🐛 Bring back the logs (ITISFoundation#2637)

Author: sanderegg

packages/models-library/src/models_library/rabbitmq_messages.py

@@ -0,0 +1,37 @@
+from typing import List, Optional, Union
+
+from models_library.projects import ProjectID
+from models_library.projects_nodes import NodeID
+from models_library.projects_state import RunningState
+from models_library.users import UserID
+from pydantic import BaseModel
+from pydantic.types import NonNegativeFloat
+from simcore_postgres_database.models.comp_tasks import NodeClass
+
+
+class RabbitMessageBase(BaseModel):
+    node_id: NodeID
+    user_id: UserID
+    project_id: ProjectID
+
+
+class LoggerRabbitMessage(RabbitMessageBase):
+    messages: List[str]
+
+
+class ProgressRabbitMessage(RabbitMessageBase):
+    progress: NonNegativeFloat
+
+
+class InstrumentationRabbitMessage(RabbitMessageBase):
+    metrics: str
+    service_uuid: NodeID
+    service_type: NodeClass
+    service_key: str
+    service_tag: str
+    result: Optional[RunningState] = None
+
+
+RabbitMessageTypes = Union[
+    LoggerRabbitMessage, ProgressRabbitMessage, InstrumentationRabbitMessage
+]

packages/pytest-simcore/src/pytest_simcore/rabbit_service.py

@@ -6,7 +6,8 @@
 import json
 import logging
 import socket
-from typing import Any, AsyncIterator, Dict, Iterator, Optional, Tuple
+from dataclasses import dataclass
+from typing import Any, AsyncIterator, Dict, Iterator, Optional
 
 import aio_pika
 import pytest
@@ -118,11 +119,18 @@ def _channel_close_callback(sender: Any, exc: Optional[BaseException] = None):
     await channel.close()
 
 
+@dataclass
+class RabbitExchanges:
+    logs: aio_pika.Exchange
+    progress: aio_pika.Exchange
+    instrumentation: aio_pika.Exchange
+
+
 @pytest.fixture(scope="function")
-async def rabbit_exchange(
+async def rabbit_exchanges(
     rabbit_config: RabbitConfig,
     rabbit_channel: aio_pika.Channel,
-) -> Tuple[aio_pika.Exchange, aio_pika.Exchange]:
+) -> RabbitExchanges:
     """
     Declares and returns 'log' and 'instrumentation' exchange channels with rabbit
     """
@@ -134,27 +142,33 @@ async def rabbit_exchange(
     )
     assert logs_exchange
 
+    # declare progress exchange
+    PROGRESS_EXCHANGE_NAME: str = rabbit_config.channels["progress"]
+    progress_exchange = await rabbit_channel.declare_exchange(
+        PROGRESS_EXCHANGE_NAME, aio_pika.ExchangeType.FANOUT
+    )
+    assert progress_exchange
+
     # declare instrumentation exchange
     INSTRUMENTATION_EXCHANGE_NAME: str = rabbit_config.channels["instrumentation"]
     instrumentation_exchange = await rabbit_channel.declare_exchange(
         INSTRUMENTATION_EXCHANGE_NAME, aio_pika.ExchangeType.FANOUT
     )
     assert instrumentation_exchange
 
-    return logs_exchange, instrumentation_exchange
+    return RabbitExchanges(logs_exchange, progress_exchange, instrumentation_exchange)
 
 
 @pytest.fixture(scope="function")
 async def rabbit_queue(
     rabbit_channel: aio_pika.Channel,
-    rabbit_exchange: Tuple[aio_pika.Exchange, aio_pika.Exchange],
+    rabbit_exchanges: RabbitExchanges,
 ) -> AsyncIterator[aio_pika.Queue]:
-    logs_exchange, instrumentation_exchange = rabbit_exchange
-
     queue = await rabbit_channel.declare_queue(exclusive=True)
     assert queue
 
     # Binding queue to exchange
-    await queue.bind(logs_exchange)
-    await queue.bind(instrumentation_exchange)
+    await queue.bind(rabbit_exchanges.logs)
+    await queue.bind(rabbit_exchanges.progress)
+    await queue.bind(rabbit_exchanges.instrumentation)
     yield queue

packages/service-library/src/servicelib/aiohttp/rest_codecs.py

@@ -2,13 +2,14 @@
 
 """
 import inspect
+import json
 from typing import Any, Dict, List, Mapping, Optional, Tuple, Type, Union
 
 import attr
 from aiohttp import web, web_exceptions
 from aiohttp.web_exceptions import HTTPError, HTTPException
 
-from .rest_codecs import json, jsonify
+from ..json_serialization import json_dumps
 from .rest_models import ErrorItemType, ErrorType, LogMessageType
 
 ENVELOPE_KEYS = ("data", "error")
@@ -77,7 +78,7 @@ def create_data_response(
         else:
             payload = data
 
-        response = web.json_response(payload, dumps=jsonify)
+        response = web.json_response(payload, dumps=json_dumps)
     except (TypeError, ValueError) as err:
         response = create_error_response(
             [
@@ -123,7 +124,7 @@ def create_error_response(
     payload = wrap_as_envelope(error=attr.asdict(error))
 
     response = http_error_cls(
-        reason=reason, text=jsonify(payload), content_type=JSON_CONTENT_TYPE
+        reason=reason, text=json_dumps(payload), content_type=JSON_CONTENT_TYPE
     )
 
     return response
@@ -136,7 +137,9 @@ def create_log_response(msg: str, level: str) -> web.Response:
     """
     # TODO: DEPRECATE
     msg = LogMessageType(msg, level)
-    response = web.json_response(data={"data": attr.asdict(msg), "error": None})
+    response = web.json_response(
+        data={"data": attr.asdict(msg), "error": None}, dumps=json_dumps
+    )
     return response
 
 

packages/service-library/src/servicelib/aiohttp/rest_responses.py

@@ -1,25 +1,27 @@
 # FIXME: move to settings-library or refactor
 
 import logging
-from typing import Optional
+from typing import Final, Optional
 
-from tenacity import before_sleep_log, stop_after_attempt, wait_fixed
+from tenacity.before_sleep import before_sleep_log
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_fixed
 
 log = logging.getLogger(__file__)
 
 
+_MINUTE: Final[int] = 60
+
+
 class RabbitMQRetryPolicyUponInitialization:
     """Retry policy upon service initialization"""
 
-    WAIT_SECS = 2
-    ATTEMPTS_COUNT = 60
-
     def __init__(self, logger: Optional[logging.Logger] = None):
         logger = logger or log
 
         self.kwargs = dict(
-            wait=wait_fixed(self.WAIT_SECS),
-            stop=stop_after_attempt(self.ATTEMPTS_COUNT),
-            before_sleep=before_sleep_log(logger, logging.INFO),
+            wait=wait_fixed(2),
+            stop=stop_after_delay(3 * _MINUTE),
+            before_sleep=before_sleep_log(logger, logging.WARNING),
             reraise=True,
         )

packages/service-library/src/servicelib/rabbitmq_utils.py

@@ -5,13 +5,13 @@
 
 import asyncio
 import json
+from dataclasses import dataclass
 
-import attr
 from aiohttp import web
-from servicelib.aiohttp.rest_codecs import DataEncoder
+from servicelib.json_serialization import json_dumps
 
 
-@attr.s(auto_attribs=True)
+@dataclass
 class Data:
     x: int = 3
     y: str = "foo"
@@ -66,4 +66,4 @@ def get(cls, suffix, process=True):
         loop = asyncio.get_event_loop()
         data = loop.run_until_complete(coro(None))
 
-        return json.loads(json.dumps(data, cls=DataEncoder)) if process else data
+        return json.loads(json_dumps(data)) if process else data

packages/service-library/tests/aiohttp/tutils.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import logging
 from dataclasses import dataclass
 from typing import Callable, Dict, List, Tuple
@@ -13,6 +12,12 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
+from models_library.rabbitmq_messages import (
+    InstrumentationRabbitMessage,
+    LoggerRabbitMessage,
+    ProgressRabbitMessage,
+)
+from simcore_postgres_database.models.comp_tasks import NodeClass
 
 from ...core.settings import DaskSchedulerSettings
 from ...models.domains.comp_tasks import CompTaskAtDB, Image
@@ -114,19 +119,18 @@ async def _on_task_completed(self, event: TaskStateEvent) -> None:
             project_id, [node_id], event.state
         )
         # instrumentation
-        message = {
-            "metrics": "service_stopped",
-            "user_id": user_id,
-            "project_id": f"{project_id}",
-            "service_uuid": f"{node_id}",
-            "service_type": "COMPUTATIONAL",
-            "service_key": service_key,
-            "service_tag": service_version,
-            "result": "SUCCESS" if event.state == RunningState.SUCCESS else "FAILURE",
-        }
-        await self.rabbitmq_client.publish_message(
-            "instrumentation", json.dumps(message)
+        message = InstrumentationRabbitMessage(
+            metrics="service_stopped",
+            user_id=user_id,
+            project_id=project_id,
+            node_id=node_id,
+            service_uuid=node_id,
+            service_type=NodeClass.COMPUTATIONAL,
+            service_key=service_key,
+            service_tag=service_version,
+            result=event.state,
         )
+        await self.rabbitmq_client.publish_message(message)
         self._wake_up_scheduler_now()
 
     async def _task_state_change_handler(self, event: str) -> None:
@@ -140,18 +144,17 @@ async def _task_state_change_handler(self, event: str) -> None:
         )
 
         if task_state_event.state == RunningState.STARTED:
-            message = {
-                "metrics": "service_started",
-                "user_id": user_id,
-                "project_id": f"{project_id}",
-                "service_uuid": f"{node_id}",
-                "service_type": "COMPUTATIONAL",
-                "service_key": service_key,
-                "service_tag": service_version,
-            }
-            await self.rabbitmq_client.publish_message(
-                "instrumentation", json.dumps(message)
+            message = InstrumentationRabbitMessage(
+                metrics="service_started",
+                user_id=user_id,
+                project_id=project_id,
+                node_id=node_id,
+                service_uuid=node_id,
+                service_type=NodeClass.COMPUTATIONAL,
+                service_key=service_key,
+                service_tag=service_version,
             )
+            await self.rabbitmq_client.publish_message(message)
 
         await CompTasksRepository(self.db_engine).set_project_tasks_state(
             project_id, [node_id], task_state_event.state
@@ -161,28 +164,23 @@ async def _task_progress_change_handler(self, event: str) -> None:
         task_progress_event = TaskProgressEvent.parse_raw(event)
         logger.debug("received task progress update: %s", task_progress_event)
         *_, user_id, project_id, node_id = parse_dask_job_id(task_progress_event.job_id)
-        message = {
-            "user_id": user_id,
-            "project_id": f"{project_id}",
-            "node_id": f"{node_id}",
-            "progress": task_progress_event.progress,
-            "channel": "progress",
-        }
-        await self.rabbitmq_client.publish_message(
-            task_progress_event.topic_name(), json.dumps(message)
+        message = ProgressRabbitMessage(
+            user_id=user_id,
+            project_id=project_id,
+            node_id=node_id,
+            progress=task_progress_event.progress,
         )
+        await self.rabbitmq_client.publish_message(message)
 
     async def _task_log_change_handler(self, event: str) -> None:
         task_log_event = TaskLogEvent.parse_raw(event)
         logger.debug("received task log update: %s", task_log_event)
         *_, user_id, project_id, node_id = parse_dask_job_id(task_log_event.job_id)
-        message = {
-            "user_id": user_id,
-            "project_id": f"{project_id}",
-            "node_id": f"{node_id}",
-            "messages": [task_log_event.log],
-            "channel": "logger",
-        }
-        await self.rabbitmq_client.publish_message(
-            task_log_event.topic_name(), json.dumps(message)
+        message = LoggerRabbitMessage(
+            user_id=user_id,
+            project_id=project_id,
+            node_id=node_id,
+            messages=[task_log_event.log],
         )
+
+        await self.rabbitmq_client.publish_message(message)