✨Ui will now receive logs from dynamic-services (ITISFoundation#2602)

* added rabbitmq

* first draft for logs fetching

* interface cahanged

* refacted setup

* changed entire interface and setup

* added typing

* fixing testsuite

* reverting to older settings scheme

* added rabbit env vars

* first iteraction of log fecthing

* new log format

* refactor log format

* fixing director-v2 tests

* using proper log format

* not sending twice

* updated logging message

* added rabbitmq tests

* @pcrespov refactor logging module usage

* added more tests

* cleanup container after test

* fixed test

* removed dataclass

* finished settings refactor

* reverted back

* fxied tests

* removing test

* better name for the channel

* enabling logs only for flaky test

* fixed test

* removed traces of old impelemtation for settings

* refacrtor rabbitmq to not be setup by the docker_logs module

* refactor internals to make is simpler to use

* using queues to dispatch messages at a interval

* refactored test

* refactor code

* sending logs to the UI from dynamic-sidecar

* fix pylint issues

* refactored dy-sidecar env vars

* fix test

* fixing message

* changed level of logs

* no need to relog messages

* rename

* removed rabbit settings

* renamed and removed thins which are not strictly required

* fixing timeout and method

* using fstrings

* fixes after merge

* refactor env vars assembly

* fixing broken test

* using monkeypatch

* fix fixture definition

* fixed test

* replacing docker IP with get_ip

* using get_ip

* fix import order

* swapped json serializer

* no need for these

* fix unit tests

* downgrading to debug

* fixed an issue when zipping

* reducing log noise

* putting [sidecar] in front of logs

* fix test

* update help

* minor fixes

* replacing mocking objects

* refactor test

* fix pylint

* refactor no longer blocks

* added registry availablity check

* fix failing test, reduced timeout

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

ci/github/integration-testing/director-v2.bash

@@ -23,8 +23,7 @@ test() {
   echo "testing in services/director-v2/tests/integration/$1"
   pytest --cov=simcore_service_director_v2 --durations=10 --cov-append \
     --color=yes --cov-report=term-missing --cov-report=xml --cov-config=.coveragerc \
-    -vvv \
-    -v -m "not travis" "services/director-v2/tests/integration/$1"
+    -v -m "not travis" "services/director-v2/tests/integration/$1" --log-level=DEBUG
 }
 
 clean_up() {

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/__init__.py

@@ -1,11 +1,11 @@
+from .proxy import get_dynamic_proxy_spec
 from .settings import (
     MATCH_SERVICE_VERSION,
     MATCH_SIMCORE_REGISTRY,
     inject_settings_to_create_service_params,
     merge_settings_before_use,
 )
-from .spec_dynamic_sidecar import (
+from .sidecar import (
     extract_service_port_from_compose_start_spec,
     get_dynamic_sidecar_spec,
 )
-from .spec_proxy import get_dynamic_proxy_spec

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/spec_proxy.py renamed to services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/proxy.py

@@ -1,15 +1,13 @@
-import json
 import logging
 from pathlib import Path
 from typing import Any, Dict
 
 from models_library.service_settings_labels import SimcoreServiceSettingsLabel
-from settings_library.docker_registry import RegistrySettings
+from servicelib.json_serialization import json_dumps
 
 from ....core.settings import AppSettings, DynamicSidecarSettings
 from ....models.schemas.constants import DYNAMIC_SIDECAR_SERVICE_PREFIX
 from ....models.schemas.dynamic_services import SchedulerData, ServiceType
-from ....utils.registry import get_dynamic_sidecar_env_vars
 from ..volumes_resolver import DynamicSidecarVolumesPathsResolver
 from .settings import inject_settings_to_create_service_params
 
@@ -22,13 +20,17 @@ def extract_service_port_from_compose_start_spec(
     return create_service_params["labels"]["service_port"]
 
 
-def _get_dy_sidecar_env_vars(
-    scheduler_data: SchedulerData, app_settings: AppSettings
+def _get_environment_variables(
+    compose_namespace: str, scheduler_data: SchedulerData, app_settings: AppSettings
 ) -> Dict[str, str]:
+    registry_settings = app_settings.DIRECTOR_V2_DOCKER_REGISTRY
+    rabbit_settings = app_settings.CELERY.CELERY_RABBIT
     return {
+        "SIMCORE_HOST_NAME": scheduler_data.service_name,
+        "DYNAMIC_SIDECAR_COMPOSE_NAMESPACE": compose_namespace,
         "DY_SIDECAR_PATH_INPUTS": f"{scheduler_data.paths_mapping.inputs_path}",
         "DY_SIDECAR_PATH_OUTPUTS": f"{scheduler_data.paths_mapping.outputs_path}",
-        "DY_SIDECAR_STATE_PATHS": json.dumps(
+        "DY_SIDECAR_STATE_PATHS": json_dumps(
             [f"{x}" for x in scheduler_data.paths_mapping.state_paths]
         ),
         "DY_SIDECAR_USER_ID": f"{scheduler_data.user_id}",
@@ -41,12 +43,22 @@ def _get_dy_sidecar_env_vars(
         "POSTGRES_USER": f"{app_settings.POSTGRES.POSTGRES_USER}",
         "POSTGRES_DB": f"{app_settings.POSTGRES.POSTGRES_DB}",
         "STORAGE_ENDPOINT": app_settings.STORAGE_ENDPOINT,
+        "REGISTRY_AUTH": f"{registry_settings.REGISTRY_AUTH}",
+        "REGISTRY_PATH": f"{registry_settings.REGISTRY_PATH}",
+        "REGISTRY_URL": f"{registry_settings.REGISTRY_URL}",
+        "REGISTRY_USER": f"{registry_settings.REGISTRY_USER}",
+        "REGISTRY_PW": f"{registry_settings.REGISTRY_PW.get_secret_value()}",
+        "REGISTRY_SSL": f"{registry_settings.REGISTRY_SSL}",
+        "RABBIT_HOST": f"{rabbit_settings.RABBIT_HOST}",
+        "RABBIT_PORT": f"{rabbit_settings.RABBIT_PORT}",
+        "RABBIT_USER": f"{rabbit_settings.RABBIT_USER}",
+        "RABBIT_PASSWORD": f"{rabbit_settings.RABBIT_PASSWORD.get_secret_value()}",
+        "RABBIT_CHANNELS": json_dumps(rabbit_settings.RABBIT_CHANNELS),
     }
 
 
 async def get_dynamic_sidecar_spec(
     scheduler_data: SchedulerData,
-    docker_registry_settings: RegistrySettings,
     dynamic_sidecar_settings: DynamicSidecarSettings,
     dynamic_sidecar_network_id: str,
     swarm_network_id: str,
@@ -156,19 +168,16 @@ async def get_dynamic_sidecar_spec(
             "service_key": scheduler_data.key,
             "service_tag": scheduler_data.version,
             "paths_mapping": scheduler_data.paths_mapping.json(),
-            "compose_spec": json.dumps(scheduler_data.compose_spec),
+            "compose_spec": json_dumps(scheduler_data.compose_spec),
             "container_http_entry": scheduler_data.container_http_entry,
         },
         "name": scheduler_data.service_name,
         "networks": [swarm_network_id, dynamic_sidecar_network_id],
         "task_template": {
             "ContainerSpec": {
-                "Env": {
-                    "SIMCORE_HOST_NAME": scheduler_data.service_name,
-                    "DYNAMIC_SIDECAR_COMPOSE_NAMESPACE": compose_namespace,
-                    **get_dynamic_sidecar_env_vars(docker_registry_settings),
-                    **_get_dy_sidecar_env_vars(scheduler_data, app_settings),
-                },
+                "Env": _get_environment_variables(
+                    compose_namespace, scheduler_data, app_settings
+                ),
                 "Hosts": [],
                 "Image": dynamic_sidecar_settings.DYNAMIC_SIDECAR_IMAGE,
                 "Init": True,

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/spec_dynamic_sidecar.py renamed to services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/docker_service_specs/sidecar.py

@@ -11,7 +11,6 @@
 from tenacity.before_sleep import before_sleep_log
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_exponential, wait_fixed
-from settings_library.docker_registry import RegistrySettings
 
 from ....core.settings import DynamicSidecarSettings
 from ....models.schemas.dynamic_services import (
@@ -84,9 +83,6 @@ async def will_trigger(cls, app: FastAPI, scheduler_data: SchedulerData) -> bool
 
     @classmethod
     async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
-        docker_registry_settings: RegistrySettings = (
-            app.state.settings.DIRECTOR_V2_DOCKER_REGISTRY
-        )
         dynamic_sidecar_settings: DynamicSidecarSettings = (
             app.state.settings.DYNAMIC_SERVICES.DYNAMIC_SIDECAR
         )
@@ -125,7 +121,6 @@ async def action(cls, app: FastAPI, scheduler_data: SchedulerData) -> None:
         # start dynamic-sidecar and run the proxy on the same node
         dynamic_sidecar_create_service_params = await get_dynamic_sidecar_spec(
             scheduler_data=scheduler_data,
-            docker_registry_settings=docker_registry_settings,
             dynamic_sidecar_settings=dynamic_sidecar_settings,
             dynamic_sidecar_network_id=dynamic_sidecar_network_id,
             swarm_network_id=swarm_network_id,

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/scheduler/events.py

@@ -30,6 +30,7 @@
 import httpx
 import pytest
 import sqlalchemy as sa
+from _pytest.monkeypatch import MonkeyPatch
 from aiodocker.containers import DockerContainer
 from asgi_lifespan import LifespanManager
 from fastapi import FastAPI
@@ -40,6 +41,7 @@
 from models_library.settings.redis import RedisConfig
 from py._path.local import LocalPath
 from pytest_mock.plugin import MockerFixture
+from pytest_simcore.helpers.utils_docker import get_ip
 from shared_comp_utils import (
     assert_computation_task_out_obj,
     assert_pipeline_status,
@@ -237,7 +239,7 @@ async def db_manager(postgres_dsn: Dict[str, str]) -> AsyncIterable[DBManager]:
 
 @pytest.fixture
 async def fast_api_app(
-    minimal_configuration: None, network_name: str, monkeypatch
+    minimal_configuration: None, network_name: str, monkeypatch: MonkeyPatch
 ) -> FastAPI:
     # Works as below line in docker.compose.yml
     # ${DOCKER_REGISTRY:-itisfoundation}/dynamic-sidecar:${DOCKER_IMAGE_TAG:-latest}
@@ -261,6 +263,10 @@ async def fast_api_app(
     monkeypatch.setenv("DIRECTOR_V2_CELERY_SCHEDULER_ENABLED", "false")
     monkeypatch.setenv("DYNAMIC_SIDECAR_TRAEFIK_ACCESS_LOG", "true")
     monkeypatch.setenv("DYNAMIC_SIDECAR_TRAEFIK_LOGLEVEL", "debug")
+    # patch host for dynamic-sidecar, not reachable via localhost
+    # the dynamic-sidecar (running inside a container) will use
+    # this address to reach the rabbit service
+    monkeypatch.setenv("RABBIT_HOST", f"{get_ip()}")
 
     settings = AppSettings.create_from_envs()
 

services/director-v2/src/simcore_service_director_v2/utils/registry.py

@@ -3,16 +3,18 @@
 
 import asyncio
 import logging
-from typing import Any, Dict
+from typing import Any, AsyncIterable, Dict
 from uuid import uuid4
 
 import aiodocker
 import pytest
 from async_asgi_testclient import TestClient
 from async_asgi_testclient.response import Response
 from async_timeout import timeout
+from models_library.settings.rabbit import RabbitConfig
 from pydantic import PositiveInt
 from pytest_mock.plugin import MockerFixture
+from pytest_simcore.helpers.utils_docker import get_ip
 from simcore_service_director_v2.core.application import init_app
 from simcore_service_director_v2.core.settings import AppSettings
 from utils import ensure_network_cleanup, patch_dynamic_service_url
@@ -21,13 +23,17 @@
 
 logger = logging.getLogger(__name__)
 
-pytest_simcore_core_services_selection = ["director"]
+pytest_simcore_core_services_selection = [
+    "director",
+    "rabbit",
+]
 
 
 @pytest.fixture
 def minimal_configuration(
     dy_static_file_server_dynamic_sidecar_service: Dict,
     simcore_services_ready: None,
+    rabbit_service: RabbitConfig,
 ):
     pass
 
@@ -76,7 +82,7 @@ async def test_client(
     mock_env: None,
     network_name: str,
     monkeypatch,
-) -> TestClient:
+) -> AsyncIterable[TestClient]:
     monkeypatch.setenv("SC_BOOT_MODE", "production")
     monkeypatch.setenv("DYNAMIC_SIDECAR_EXPOSE_PORT", "true")
     monkeypatch.setenv("SIMCORE_SERVICES_NETWORK_NAME", network_name)
@@ -92,6 +98,11 @@ async def test_client(
     monkeypatch.setenv("POSTGRES_DB", "mocked_db")
     monkeypatch.setenv("DIRECTOR_V2_POSTGRES_ENABLED", "false")
 
+    # patch host for dynamic-sidecar, not reachable via localhost
+    # the dynamic-sidecar (running inside a container) will use
+    # this address to reach the rabbit service
+    monkeypatch.setenv("RABBIT_HOST", f"{get_ip()}")
+
     settings = AppSettings.create_from_envs()
 
     app = init_app(settings)

services/director-v2/tests/integration/02/test__dynamic_sidecar_nodeports_integration.py

@@ -16,6 +16,7 @@
 from models_library.projects import ProjectAtDB
 from models_library.settings.rabbit import RabbitConfig
 from models_library.settings.redis import RedisConfig
+from pytest_simcore.helpers.utils_docker import get_ip
 from simcore_sdk.node_ports_common import config as node_ports_config
 from simcore_service_director_v2.core.application import init_app
 from simcore_service_director_v2.core.settings import AppSettings
@@ -155,6 +156,10 @@ async def director_v2_client(
     monkeypatch.setenv("POSTGRES_PASSWORD", "mocked_password")
     monkeypatch.setenv("POSTGRES_DB", "mocked_db")
     monkeypatch.setenv("DIRECTOR_V2_POSTGRES_ENABLED", "false")
+    # patch host for dynamic-sidecar, not reachable via localhost
+    # the dynamic-sidecar (running inside a container) will use
+    # this address to reach the rabbit service
+    monkeypatch.setenv("RABBIT_HOST", f"{get_ip()}")
 
     settings = AppSettings.create_from_envs()
 

services/director-v2/tests/integration/02/test_dynamic_services_routes.py

@@ -11,6 +11,7 @@
 from fastapi import FastAPI
 from models_library.projects import Node
 from pydantic import PositiveInt
+from pytest_simcore.helpers.utils_docker import get_ip
 from simcore_service_director_v2.models.schemas.constants import (
     DYNAMIC_PROXY_SERVICE_PREFIX,
     DYNAMIC_SIDECAR_SERVICE_PREFIX,
@@ -23,7 +24,7 @@
 from tenacity.wait import wait_fixed
 
 SERVICE_WAS_CREATED_BY_DIRECTOR_V2 = 20
-SERVICES_ARE_READY_TIMEOUT = 10 * 60
+SERVICES_ARE_READY_TIMEOUT = 2 * 60
 SEPARATOR = "=" * 50
 
 
@@ -57,7 +58,7 @@ async def patch_dynamic_service_url(app: FastAPI, node_uuid: str) -> str:
     Normally director-v2 talks via docker-netwoks with the dynamic-sidecar.
     Since the director-v2 was started outside docker and is not
     running in a container, the service port needs to be exposed and the
-    url needs to be changed to 172.17.0.1 (docker localhost)
+    url needs to be changed to get_ip()
 
     returns: the local endpoint
     """
@@ -85,11 +86,11 @@ async def patch_dynamic_service_url(app: FastAPI, node_uuid: str) -> str:
     async with scheduler._lock:  # pylint: disable=protected-access
         for entry in scheduler._to_observe.values():  # pylint: disable=protected-access
             if entry.scheduler_data.service_name == service_name:
-                entry.scheduler_data.dynamic_sidecar.hostname = "172.17.0.1"
+                entry.scheduler_data.dynamic_sidecar.hostname = f"{get_ip()}"
                 entry.scheduler_data.dynamic_sidecar.port = port
 
                 endpoint = entry.scheduler_data.dynamic_sidecar.endpoint
-                assert endpoint == f"http://172.17.0.1:{port}"
+                assert endpoint == f"http://{get_ip()}:{port}"
                 break
 
     assert endpoint is not None
@@ -101,7 +102,7 @@ async def _get_proxy_port(node_uuid: str) -> PositiveInt:
     Normally director-v2 talks via docker-netwoks with the started proxy.
     Since the director-v2 was started outside docker and is not
     running in a container, the service port needs to be exposed and the
-    url needs to be changed to 172.17.0.1 (docker localhost)
+    url needs to be changed to get_ip()
 
     returns: the local endpoint
     """
@@ -328,9 +329,9 @@ async def assert_service_is_available(  # pylint: disable=redefined-outer-name
     exposed_port: PositiveInt, is_legacy: bool, service_uuid: str
 ) -> None:
     service_address = (
-        f"http://172.17.0.1:{exposed_port}/x/{service_uuid}"
+        f"http://{get_ip()}:{exposed_port}/x/{service_uuid}"
         if is_legacy
-        else f"http://172.17.0.1:{exposed_port}"
+        else f"http://{get_ip()}:{exposed_port}"
     )
     print(f"checking service @ {service_address}")