Merge branch 'feature/event-driven-patches' of github.com:odeimaiz/osparc-simcore into feature/event-driven-patches

Author: odeimaiz

.github/workflows/ci-testing-deploy.yml

@@ -1903,20 +1903,27 @@ jobs:
         run: ./ci/github/system-testing/e2e.bash install
       - name: test
         run: ./ci/github/system-testing/e2e.bash test
+      - name: wait before dumping logs
+        if: ${{ failure() }}
+        run: sleep 5
       - name: dump docker logs
+        if: ${{ failure() }}
         id: docker_logs_dump
         run: ./ci/github/system-testing/e2e.bash dump_docker_logs
       - name: upload docker logs
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ github.job }}_docker_logs
           path: ./tests/e2e/test_failures
       - name: upload screenshots
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ github.job }}_screenshots
           path: tests/e2e/screenshots
       - name: upload e2e logs
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ github.job }}_logs
@@ -1955,15 +1962,21 @@ jobs:
       - name: test
         run: |
           ./ci/github/system-testing/e2e-playwright.bash test
+      - name: wait before dumping logs
+        if: ${{ failure() }}
+        run: sleep 5
       - name: dump docker logs
+        if: ${{ failure() }}
         id: docker_logs_dump
         run: ./ci/github/system-testing/e2e-playwright.bash dump_docker_logs
       - name: upload docker logs
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ github.job }}_docker_logs
           path: ./tests/e2e-playwright/test_failures
       - name: upload tracing if failed
+        if: ${{ failure() }}
         uses: actions/upload-artifact@v4
         with:
           name: ${{ github.job }}_tracing

Makefile

@@ -378,8 +378,8 @@ up-devel-frontend: .stack-simcore-development-frontend.yml .init-swarm ## Every
 	@$(MAKE_C) services/dask-sidecar certificates
 	# Deploy stack $(SWARM_STACK_NAME)  [back-end]
 	@docker stack deploy --detach=true --with-registry-auth -c $< $(SWARM_STACK_NAME)
-	@$(MAKE) .deploy-vendors
 	@$(MAKE) .deploy-ops
+	@$(MAKE) .deploy-vendors
 	@$(_show_endpoints)
 	@$(MAKE_C) services/static-webserver/client follow-dev-logs
 
@@ -389,8 +389,8 @@ ifeq ($(target),)
 	@$(MAKE_C) services/dask-sidecar certificates
 	# Deploy stack $(SWARM_STACK_NAME)
 	@docker stack deploy --detach=true --with-registry-auth -c $< $(SWARM_STACK_NAME)
-	@$(MAKE) .deploy-vendors
 	@$(MAKE) .deploy-ops
+	@$(MAKE) .deploy-vendors
 else
 	# deploys ONLY $(target) service
 	@docker compose --file $< up --detach $(target)
@@ -671,6 +671,7 @@ local-registry: .env ## creates a local docker registry and configure simcore to
 					sudo mv /tmp/daemon.json /etc/docker/daemon.json &&\
 					echo restarting engine... &&\
 					sudo service docker restart &&\
+					sleep 5 &&\
 					echo done)
 
 	@$(if $(shell docker ps --format="{{.Names}}" | grep registry),,\

ci/github/system-testing/e2e-playwright.bash

@@ -37,7 +37,7 @@ dump_docker_logs() {
   for service_id in $(docker service ls -q); do
     service_name=$(docker service inspect "$service_id" --format="{{.Spec.Name}}")
     echo "Dumping logs for $service_name"
-    (timeout 30 docker service logs --timestamps --tail=400 --details "$service_id" >"$out_dir/$service_name.log" 2>&1) || true
+    (timeout 30 docker service logs --timestamps --tail=500 --details "$service_id" >"$out_dir/$service_name.log" 2>&1) || true
   done
 }
 

packages/pytest-simcore/src/pytest_simcore/helpers/logging_tools.py

@@ -1,10 +1,11 @@
 import datetime
 import logging
-from collections.abc import Iterator
+import warnings
+from collections.abc import Callable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass, field
 from types import SimpleNamespace
-from typing import TypeAlias
+from typing import Final, TypeAlias
 
 
 def _timedelta_as_minute_second_ms(delta: datetime.timedelta) -> str:
@@ -29,6 +30,19 @@ def _timedelta_as_minute_second_ms(delta: datetime.timedelta) -> str:
     return f"{sign}{result.strip()}"
 
 
+def _resolve(val: str | Callable[[], str], prefix: str, suffix: str) -> str:
+    try:
+        return f"{prefix}{val if isinstance(val, str) else val()}{suffix}"
+    except Exception as exc:  # pylint: disable=broad-exception-caught
+        warnings.warn(
+            f"Failed to generate {val} message: {exc!r}. "
+            f"Fix the callable to return a string without raising exceptions.",
+            UserWarning,
+            stacklevel=3,
+        )
+        return f"❌❌❌ [{val} message generation failed TIP: Check how the {val} message is generated!] ❌❌❌"
+
+
 class DynamicIndentFormatter(logging.Formatter):
     indent_char: str = "    "
     _cls_indent_level: int = 0
@@ -74,15 +88,26 @@ def setup(cls, logger: logging.Logger) -> None:
 DynamicIndentFormatter.setup(test_logger)
 
 
+# Message formatting constants
+_STARTING_PREFIX: Final[str] = "--> "
+_STARTING_SUFFIX: Final[str] = " ⏳"
+_DONE_PREFIX: Final[str] = "<-- "
+_DONE_SUFFIX: Final[str] = " ✅"
+_RAISED_PREFIX: Final[str] = "❌❌❌ Error: "
+_RAISED_SUFFIX: Final[str] = " ❌❌❌"
+
+
 @dataclass
 class ContextMessages:
-    starting: str
-    done: str
-    raised: str = field(default="")
+    starting: str | Callable[[], str]
+    done: str | Callable[[], str]
+    raised: str | Callable[[], str] = field(default="")
 
     def __post_init__(self):
         if not self.raised:
-            self.raised = f"{self.done} [with error]"
+            self.raised = (
+                lambda: f"{self.done if isinstance(self.done, str) else self.done()} [with raised error]"
+            )
 
 
 LogLevelInt: TypeAlias = int
@@ -127,9 +152,9 @@ def log_context(
 
     if isinstance(msg, str):
         ctx_msg = ContextMessages(
-            starting=f"-> {msg} starting ...",
-            done=f"<- {msg} done",
-            raised=f"! {msg} raised",
+            starting=f"{msg}",
+            done=f"{msg}",
+            raised=f"{msg}",
         )
     elif isinstance(msg, tuple):
         ctx_msg = ContextMessages(*msg)
@@ -140,13 +165,16 @@ def log_context(
     try:
         DynamicIndentFormatter.cls_increase_indent()
 
-        logger.log(level, ctx_msg.starting, *args, **kwargs)
+        logger.log(
+            level,
+            _resolve(ctx_msg.starting, _STARTING_PREFIX, _STARTING_SUFFIX),
+            *args,
+            **kwargs,
+        )
         with _increased_logger_indent(logger):
             yield SimpleNamespace(logger=logger, messages=ctx_msg)
         elapsed_time = datetime.datetime.now(tz=datetime.UTC) - started_time
-        done_message = (
-            f"{ctx_msg.done} ({_timedelta_as_minute_second_ms(elapsed_time)})"
-        )
+        done_message = f"{_resolve(ctx_msg.done, _DONE_PREFIX, _DONE_SUFFIX)} ({_timedelta_as_minute_second_ms(elapsed_time)})"
         logger.log(
             level,
             done_message,
@@ -156,9 +184,7 @@ def log_context(
 
     except:
         elapsed_time = datetime.datetime.now(tz=datetime.UTC) - started_time
-        error_message = (
-            f"{ctx_msg.raised} ({_timedelta_as_minute_second_ms(elapsed_time)})"
-        )
+        error_message = f"{_resolve(ctx_msg.raised, _RAISED_PREFIX, _RAISED_SUFFIX)} ({_timedelta_as_minute_second_ms(elapsed_time)})"
         logger.exception(
             error_message,
             *args,

packages/pytest-simcore/src/pytest_simcore/helpers/playwright.py

@@ -43,7 +43,7 @@
     wait_fixed,
 )
 
-from .logging_tools import log_context
+from .logging_tools import ContextMessages, log_context
 
 _logger = logging.getLogger(__name__)
 
@@ -532,9 +532,10 @@ def wait_for_pipeline_state(
     if current_state in if_in_states:
         with log_context(
             logging.INFO,
-            msg=(
-                f"pipeline is in {current_state=}, waiting for one of {expected_states=}",
-                f"pipeline is now in {current_state=}",
+            msg=ContextMessages(
+                starting=f"wait for one of {expected_states=}",
+                done=lambda: f"wait for one of {expected_states=}, pipeline reached {current_state=}",
+                raised=lambda: f"pipeline failed or timed out with {current_state}. Expected one of {expected_states=}",
             ),
         ):
             waiter = SocketIOProjectStateUpdatedWaiter(
@@ -551,7 +552,7 @@ def wait_for_pipeline_state(
                 and current_state not in expected_states
             ):
                 pytest.fail(
-                    f"❌ Pipeline failed with state {current_state}. Expected one of {expected_states} ❌"
+                    f"❌ Pipeline failed fast with state {current_state}. Expected one of {expected_states} ❌"
                 )
     return current_state
 

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

@@ -70,7 +70,7 @@ async def web_server(
 
 @pytest.fixture
 async def server_url(web_server: URL) -> str:
-    return f'{web_server.with_path("/")}'
+    return f"{web_server.with_path('/')}"
 
 
 @pytest.fixture

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/webserver/functions/functions_rpc_interface.py

@@ -17,6 +17,8 @@
     RegisteredFunctionJobCollection,
 )
 from models_library.functions import (
+    FunctionJobStatus,
+    FunctionOutputs,
     FunctionUserAccessRights,
     FunctionUserApiAccessRights,
 )
@@ -300,6 +302,82 @@ async def get_function_job(
     return TypeAdapter(RegisteredFunctionJob).validate_python(result)
 
 
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_function_job_status(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    function_job_id: FunctionJobID,
+    product_name: ProductName,
+) -> FunctionJobStatus:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("get_function_job_status"),
+        function_job_id=function_job_id,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionJobStatus).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_function_job_outputs(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    function_job_id: FunctionJobID,
+    product_name: ProductName,
+) -> FunctionOutputs:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("get_function_job_outputs"),
+        function_job_id=function_job_id,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionOutputs).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def update_function_job_status(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_id: FunctionJobID,
+    job_status: FunctionJobStatus,
+) -> FunctionJobStatus:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("update_function_job_status"),
+        function_job_id=function_job_id,
+        job_status=job_status,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionJobStatus).validate_python(result)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def update_function_job_outputs(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    user_id: UserID,
+    product_name: ProductName,
+    function_job_id: FunctionJobID,
+    outputs: FunctionOutputs,
+) -> FunctionOutputs:
+    result = await rabbitmq_rpc_client.request(
+        WEBSERVER_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("update_function_job_outputs"),
+        function_job_id=function_job_id,
+        outputs=outputs,
+        user_id=user_id,
+        product_name=product_name,
+    )
+    return TypeAdapter(FunctionOutputs).validate_python(result)
+
+
 @log_decorator(_logger, level=logging.DEBUG)
 async def delete_function_job(
     rabbitmq_rpc_client: RabbitMQRPCClient,

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

@@ -1,4 +1,4 @@
-""" Common utilities for python-socketio library
+"""Common utilities for python-socketio library
 
 
 NOTE: we intentionally avoided importing socketio here to avoid adding an extra dependency at
@@ -9,7 +9,6 @@
 
 
 async def cleanup_socketio_async_pubsub_manager(server_manager):
-
     # NOTE: this is ugly. It seems though that python-socketio does not
     # cleanup its background tasks properly.
     # https://github.com/miguelgrinberg/python-socketio/discussions/1092
@@ -35,6 +34,7 @@ async def cleanup_socketio_async_pubsub_manager(server_manager):
             for coro_name in [
                 "AsyncServer._service_task",
                 "AsyncSocket.schedule_ping",
+                "AsyncSocket._send_ping",
                 "AsyncPubSubManager._thread",
             ]
         ):

scripts/maintenance/computational-clusters/autoscaled_monitor/core.py

@@ -615,7 +615,11 @@ async def _cancel_all_jobs(
                     the_cluster,
                     dask_task.job_id,
                 )
-        if comp_task is not None and abort_in_db:
+        if (
+            comp_task is not None
+            and comp_task.state not in ["FAILED", "SUCCESS", "ABORTED"]
+            and abort_in_db
+        ):
             await db.abort_job_in_db(state, comp_task.project_id, comp_task.node_id)
 
         rich.print("cancelled all tasks")