Merge branch 'master' into is7961/upgrade-gc-periodic-task

Author: pcrespov

packages/celery-library/src/celery_library/signals.py

@@ -4,7 +4,7 @@
 
 from celery import Celery  # type: ignore[import-untyped]
 from celery.worker.worker import WorkController  # type: ignore[import-untyped]
-from servicelib.celery.app_server import STARTUP_TIMEOUT, BaseAppServer
+from servicelib.celery.app_server import BaseAppServer
 from servicelib.logging_utils import log_context
 from settings_library.celery import CelerySettings
 
@@ -26,8 +26,6 @@ def _init(startup_complete_event: threading.Event) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
 
-        shutdown_event = asyncio.Event()
-
         async def _setup_task_manager():
             assert sender.app  # nosec
             assert isinstance(sender.app, Celery)  # nosec
@@ -42,9 +40,7 @@ async def _setup_task_manager():
         app_server.event_loop = loop
 
         loop.run_until_complete(_setup_task_manager())
-        loop.run_until_complete(
-            app_server.startup(startup_complete_event, shutdown_event)
-        )
+        loop.run_until_complete(app_server.lifespan(startup_complete_event))
 
     thread = threading.Thread(
         group=None,
@@ -55,12 +51,12 @@ async def _setup_task_manager():
     )
     thread.start()
 
-    startup_complete_event.wait(STARTUP_TIMEOUT * 1.1)
+    startup_complete_event.wait()
 
 
 def on_worker_shutdown(sender, **_kwargs) -> None:
     with log_context(_logger, logging.INFO, "Worker shutdown"):
         assert isinstance(sender.app, Celery)
         app_server = get_app_server(sender.app)
 
-        app_server.event_loop.run_until_complete(app_server.shutdown())
+        app_server.shutdown_event.set()

packages/celery-library/tests/conftest.py

@@ -2,6 +2,7 @@
 # pylint: disable=unused-argument
 
 import datetime
+import threading
 from collections.abc import AsyncIterator, Callable
 from functools import partial
 from typing import Any
@@ -30,10 +31,7 @@
 
 
 class FakeAppServer(BaseAppServer):
-    async def on_startup(self) -> None:
-        pass
-
-    async def on_shutdown(self) -> None:
+    async def lifespan(self, startup_completed_event: threading.Event) -> None:
         pass
 
 

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

@@ -40,6 +40,7 @@
     "whoami",
     "sto-worker",
     "sto-worker-cpu-bound",
+    "traefik-configuration-placeholder",
 }
 # TODO: unify healthcheck policies see  https://github.com/ITISFoundation/osparc-simcore/pull/2281
 DEFAULT_SERVICE_HEALTHCHECK_ENTRYPOINT: Final[str] = "/v0/"

packages/service-library/src/servicelib/celery/app_server.py

@@ -1,21 +1,18 @@
 import asyncio
-import datetime
 import threading
 from abc import ABC, abstractmethod
 from asyncio import AbstractEventLoop
-from typing import Final, Generic, TypeVar
+from typing import Generic, TypeVar
 
 from servicelib.celery.task_manager import TaskManager
 
-STARTUP_TIMEOUT: Final[float] = datetime.timedelta(minutes=1).total_seconds()
-
 T = TypeVar("T")
 
 
 class BaseAppServer(ABC, Generic[T]):
     def __init__(self, app: T) -> None:
         self._app: T = app
-        self._shutdown_event: asyncio.Event | None = None
+        self._shutdown_event: asyncio.Event = asyncio.Event()
 
     @property
     def app(self) -> T:
@@ -29,6 +26,10 @@ def event_loop(self) -> AbstractEventLoop:
     def event_loop(self, loop: AbstractEventLoop) -> None:
         self._event_loop = loop
 
+    @property
+    def shutdown_event(self) -> asyncio.Event:
+        return self._shutdown_event
+
     @property
     def task_manager(self) -> TaskManager:
         return self._task_manager
@@ -38,23 +39,8 @@ def task_manager(self, manager: TaskManager) -> None:
         self._task_manager = manager
 
     @abstractmethod
-    async def on_startup(self) -> None:
-        raise NotImplementedError
-
-    async def startup(
-        self, completed_event: threading.Event, shutdown_event: asyncio.Event
+    async def lifespan(
+        self,
+        startup_completed_event: threading.Event,
     ) -> None:
-        self._shutdown_event = shutdown_event
-        completed_event.set()
-        await self.on_startup()
-        await self._shutdown_event.wait()
-
-    @abstractmethod
-    async def on_shutdown(self) -> None:
         raise NotImplementedError
-
-    async def shutdown(self) -> None:
-        if self._shutdown_event is not None:
-            self._shutdown_event.set()
-
-        await self.on_shutdown()

packages/service-library/src/servicelib/fastapi/celery/app_server.py

@@ -1,29 +1,29 @@
-from datetime import timedelta
+import datetime
+import logging
+import threading
 from typing import Final
 
 from asgi_lifespan import LifespanManager
 from fastapi import FastAPI
 
 from ...celery.app_server import BaseAppServer
 
-_SHUTDOWN_TIMEOUT: Final[float] = timedelta(seconds=10).total_seconds()
-_STARTUP_TIMEOUT: Final[float] = timedelta(minutes=1).total_seconds()
+_SHUTDOWN_TIMEOUT: Final[float] = datetime.timedelta(seconds=10).total_seconds()
+
+_logger = logging.getLogger(__name__)
 
 
 class FastAPIAppServer(BaseAppServer[FastAPI]):
     def __init__(self, app: FastAPI):
         super().__init__(app)
         self._lifespan_manager: LifespanManager | None = None
 
-    async def on_startup(self) -> None:
-        self._lifespan_manager = LifespanManager(
+    async def lifespan(self, startup_completed_event: threading.Event) -> None:
+        async with LifespanManager(
             self.app,
-            startup_timeout=_STARTUP_TIMEOUT,
+            startup_timeout=None,  # waits for full app initialization (DB migrations, etc.)
             shutdown_timeout=_SHUTDOWN_TIMEOUT,
-        )
-        await self._lifespan_manager.__aenter__()
-
-    async def on_shutdown(self) -> None:
-        if self._lifespan_manager is None:
-            return
-        await self._lifespan_manager.__aexit__(None, None, None)
+        ):
+            _logger.info("fastapi app initialized")
+            startup_completed_event.set()
+            await self.shutdown_event.wait()  # NOTE: wait here until shutdown is requested

services/docker-compose.yml

@@ -57,6 +57,7 @@ services:
         - traefik.http.services.${SWARM_STACK_NAME}_api-server.loadbalancer.healthcheck.path=/
         - traefik.http.services.${SWARM_STACK_NAME}_api-server.loadbalancer.healthcheck.interval=2000ms
         - traefik.http.services.${SWARM_STACK_NAME}_api-server.loadbalancer.healthcheck.timeout=1000ms
+        # NOTE: keep in sync with fallback router (rule and entrypoint)
         - traefik.http.routers.${SWARM_STACK_NAME}_api-server.rule=(Path(`/`) || Path(`/v0`) ||  PathPrefix(`/v0/`) || Path(`/api/v0/openapi.json`))
         - traefik.http.routers.${SWARM_STACK_NAME}_api-server.entrypoints=simcore_api
         - traefik.http.routers.${SWARM_STACK_NAME}_api-server.priority=3
@@ -628,6 +629,7 @@ services:
         - traefik.http.services.${SWARM_STACK_NAME}_static_webserver.loadbalancer.healthcheck.interval=2000ms
         - traefik.http.services.${SWARM_STACK_NAME}_static_webserver.loadbalancer.healthcheck.timeout=1000ms
         - traefik.http.middlewares.${SWARM_STACK_NAME}_static_webserver_retry.retry.attempts=2
+        # NOTE: keep in sync with fallback router (rule and entrypoint)
         - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver.rule=(Path(`/osparc`) || Path(`/s4l`) || Path(`/s4llite`) || Path(`/s4lacad`) || Path(`/s4lengine`) || Path(`/s4ldesktop`) || Path(`/s4ldesktopacad`) || Path(`/tis`) || Path(`/tiplite`) || Path(`/transpiled`) || Path(`/resource`) || PathPrefix(`/osparc/`) || PathPrefix(`/s4l/`) || PathPrefix(`/s4llite/`) || PathPrefix(`/s4lacad/`) || PathPrefix(`/s4lengine/`) || PathPrefix(`/s4ldesktop/`) || PathPrefix(`/s4ldesktopacad/`) || PathPrefix(`/tis/`) || PathPrefix(`/tiplite/`) || PathPrefix(`/transpiled/`) || PathPrefix(`/resource/`))
         - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver.service=${SWARM_STACK_NAME}_static_webserver
         - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver.entrypoints=http
@@ -871,6 +873,7 @@ services:
         # NOTE: stickyness must remain only for specific endpoints, see https://github.com/ITISFoundation/osparc-simcore/pull/4180
         - traefik.http.middlewares.${SWARM_STACK_NAME}_webserver_retry.retry.attempts=2
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver.service=${SWARM_STACK_NAME}_webserver
+        # NOTE: keep in sync with fallback router (rule and entrypoint)
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver.rule=(Path(`/`) || Path(`/v0`) || Path(`/socket.io/`) || Path(`/static-frontend-data.json`) || PathRegexp(`^/study/(?P<study_uuid>\b[0-9a-f]{8}\b-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-\b[0-9a-f]{12}\b)`) || Path(`/view`) || Path(`/#/view`) || Path(`/#/error`) || PathPrefix(`/v0/`))
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver.entrypoints=http
         - traefik.http.routers.${SWARM_STACK_NAME}_webserver.priority=6
@@ -1456,6 +1459,62 @@ services:
       - default
       - interactive_services_subnet # for legacy dynamic services
 
+  # use to define fallback routes for simcore services
+  # if docker healthcheck fails, container's traefik configuration is removed
+  # leading to 404 https://github.com/traefik/traefik/issues/7842
+  #
+  # use fallback routes to return proper 503 (instead of 404)
+  # this service must be running at all times
+  traefik-configuration-placeholder:
+    image: busybox:1.35.0
+    command: sleep infinity
+    networks:
+      - default
+    deploy:
+      labels:
+        # route to internal traefik
+        - traefik.enable=true
+        - io.simcore.zone=${TRAEFIK_SIMCORE_ZONE}
+
+        ### Fallback for api-server
+        - traefik.http.routers.${SWARM_STACK_NAME}_api-server_fallback.rule=(Path(`/`) || Path(`/v0`) ||  PathPrefix(`/v0/`) || Path(`/api/v0/openapi.json`))
+        - traefik.http.routers.${SWARM_STACK_NAME}_api-server_fallback.service=${SWARM_STACK_NAME}_api-server_fallback
+        - traefik.http.routers.${SWARM_STACK_NAME}_api-server_fallback.entrypoints=simcore_api
+        - traefik.http.routers.${SWARM_STACK_NAME}_api-server_fallback.priority=1
+        # always fail and return 503 via unhealthy loadbalancer healthcheck
+        - traefik.http.services.${SWARM_STACK_NAME}_api-server_fallback.loadbalancer.server.port=0 # port is required (otherwise traefik service is not created)
+        - traefik.http.services.${SWARM_STACK_NAME}_api-server_fallback.loadbalancer.healthcheck.path=/some/invalid/path/to/generate/a/503
+        - traefik.http.services.${SWARM_STACK_NAME}_api-server_fallback.loadbalancer.healthcheck.interval=10s
+        - traefik.http.services.${SWARM_STACK_NAME}_api-server_fallback.loadbalancer.healthcheck.timeout=1ms
+
+        ### Fallback for webserver
+        - traefik.http.routers.${SWARM_STACK_NAME}_webserver_fallback.service=${SWARM_STACK_NAME}_webserver_fallback
+        - traefik.http.routers.${SWARM_STACK_NAME}_webserver_fallback.rule=(Path(`/`) || Path(`/v0`) || Path(`/socket.io/`) || Path(`/static-frontend-data.json`) || PathRegexp(`^/study/(?P<study_uuid>\b[0-9a-f]{8}\b-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-\b[0-9a-f]{12}\b)`) || Path(`/view`) || Path(`/#/view`) || Path(`/#/error`) || PathPrefix(`/v0/`))
+        - traefik.http.routers.${SWARM_STACK_NAME}_webserver_fallback.entrypoints=http
+        - traefik.http.routers.${SWARM_STACK_NAME}_webserver_fallback.priority=1
+        # always fail and return 503 via unhealthy loadbalancer healthcheck
+        - traefik.http.services.${SWARM_STACK_NAME}_webserver_fallback.loadbalancer.server.port=0
+        - traefik.http.services.${SWARM_STACK_NAME}_webserver_fallback.loadbalancer.healthcheck.path=/v0/
+        - traefik.http.services.${SWARM_STACK_NAME}_webserver_fallback.loadbalancer.healthcheck.interval=10s
+        - traefik.http.services.${SWARM_STACK_NAME}_webserver_fallback.loadbalancer.healthcheck.timeout=1ms
+
+        ### Fallback for static-webserver
+        - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver_fallback.rule=(Path(`/osparc`) || Path(`/s4l`) || Path(`/s4llite`) || Path(`/s4lacad`) || Path(`/s4lengine`) || Path(`/s4ldesktop`) || Path(`/s4ldesktopacad`) || Path(`/tis`) || Path(`/tiplite`) || Path(`/transpiled`) || Path(`/resource`) || PathPrefix(`/osparc/`) || PathPrefix(`/s4l/`) || PathPrefix(`/s4llite/`) || PathPrefix(`/s4lacad/`) || PathPrefix(`/s4lengine/`) || PathPrefix(`/s4ldesktop/`) || PathPrefix(`/s4ldesktopacad/`) || PathPrefix(`/tis/`) || PathPrefix(`/tiplite/`) || PathPrefix(`/transpiled/`) || PathPrefix(`/resource/`))
+        - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver_fallback.service=${SWARM_STACK_NAME}_static_webserver_fallback
+        - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver_fallback.entrypoints=http
+        - traefik.http.routers.${SWARM_STACK_NAME}_static_webserver_fallback.priority=1
+        # always fail and return 503 via unhealthy loadbalancer healthcheck
+        - traefik.http.services.${SWARM_STACK_NAME}_static_webserver_fallback.loadbalancer.server.port=0
+        - traefik.http.services.${SWARM_STACK_NAME}_static_webserver_fallback.loadbalancer.healthcheck.path=/some/invalid/path/to/generate/a/503
+        - traefik.http.services.${SWARM_STACK_NAME}_static_webserver_fallback.loadbalancer.healthcheck.interval=10s
+        - traefik.http.services.${SWARM_STACK_NAME}_static_webserver_fallback.loadbalancer.healthcheck.timeout=1ms
+    healthcheck:
+      test: command -v sleep
+      interval: 10s
+      timeout: 1s
+      start_period: 1s
+      retries: 3
+
 volumes:
   postgres_data:
     name: ${SWARM_STACK_NAME}_postgres_data

services/static-webserver/client/source/class/osparc/conversation/AddMessage.js

@@ -36,8 +36,8 @@ qx.Class.define("osparc.conversation.AddMessage", {
   },
 
   events: {
-    "commentAdded": "qx.event.type.Data",
-    "messageEdited": "qx.event.type.Data",
+    "messageAdded": "qx.event.type.Data",
+    "messageUpdated": "qx.event.type.Data",
   },
 
   members: {
@@ -227,7 +227,7 @@ qx.Class.define("osparc.conversation.AddMessage", {
       if (content) {
         osparc.study.Conversations.addMessage(this.__studyData["uuid"], this.__conversationId, content)
           .then(data => {
-            this.fireDataEvent("commentAdded", data);
+            this.fireDataEvent("messageAdded", data);
             commentField.getChildControl("text-area").setValue("");
           });
       }
@@ -239,7 +239,7 @@ qx.Class.define("osparc.conversation.AddMessage", {
       if (content) {
         osparc.study.Conversations.editMessage(this.__studyData["uuid"], this.__conversationId, this.__message["messageId"], content)
           .then(data => {
-            this.fireDataEvent("messageEdited", data);
+            this.fireDataEvent("messageUpdated", data);
             commentField.getChildControl("text-area").setValue("");
           });
       }
@@ -249,7 +249,7 @@ qx.Class.define("osparc.conversation.AddMessage", {
       if (userGid) {
         osparc.study.Conversations.notifyUser(this.__studyData["uuid"], this.__conversationId, userGid)
           .then(data => {
-            this.fireDataEvent("commentAdded", data);
+            this.fireDataEvent("messageAdded", data);
             const potentialCollaborators = osparc.store.Groups.getInstance().getPotentialCollaborators();
             if (userGid in potentialCollaborators) {
               if ("getUserId" in potentialCollaborators[userGid]) {