Init celery application in app once + run call to blocking call in executor (ITISFoundation#2372)

* move celery client where it is used --> activity manager
* use pytest-celery to mock celery app

Author: sanderegg

services/web/server/requirements/_test.in

@@ -11,6 +11,7 @@
 coverage
 pytest
 pytest-aiohttp  # incompatible with pytest-asyncio. See https://github.com/pytest-dev/pytest-asyncio/issues/76
+pytest-celery
 pytest-cov
 pytest-docker
 pytest-icdiff

services/web/server/requirements/_test.txt

@@ -13,6 +13,10 @@ aioresponses==0.7.2
     # via -r requirements/_test.in
 alembic==1.6.2
     # via -r requirements/_test.in
+amqp==5.0.6
+    # via
+    #   -c requirements/_base.txt
+    #   kombu
 astroid==2.5
     # via pylint
 async-timeout==3.0.1
@@ -28,8 +32,16 @@ attrs==20.3.0
     #   pytest-docker
 bcrypt==3.2.0
     # via paramiko
+billiard==3.6.4.0
+    # via
+    #   -c requirements/_base.txt
+    #   celery
 cached-property==1.5.2
     # via docker-compose
+celery[redis]==5.0.5
+    # via
+    #   -c requirements/_base.txt
+    #   pytest-celery
 certifi==2020.12.5
     # via requests
 cffi==1.14.5
@@ -43,10 +55,26 @@ chardet==3.0.4
     #   -c requirements/_base.txt
     #   aiohttp
     #   requests
+click-didyoumean==0.0.3
+    # via
+    #   -c requirements/_base.txt
+    #   celery
+click-plugins==1.1.1
+    # via
+    #   -c requirements/_base.txt
+    #   celery
+click-repl==0.1.6
+    # via
+    #   -c requirements/_base.txt
+    #   celery
 click==7.1.2
     # via
     #   -c requirements/_base.txt
     #   -r requirements/_test.in
+    #   celery
+    #   click-didyoumean
+    #   click-plugins
+    #   click-repl
 codecov==2.1.11
     # via -r requirements/_test.in
 coverage[toml]==5.5
@@ -96,6 +124,7 @@ importlib-metadata==4.0.1
     # via
     #   -c requirements/_base.txt
     #   jsonschema
+    #   kombu
     #   pluggy
     #   pytest
 iniconfig==1.1.1
@@ -113,6 +142,10 @@ jsonschema==3.2.0
     #   docker-compose
     #   openapi-schema-validator
     #   openapi-spec-validator
+kombu==5.0.2
+    # via
+    #   -c requirements/_base.txt
+    #   celery
 lazy-object-proxy==1.4.3
     # via
     #   -c requirements/_base.txt
@@ -148,6 +181,10 @@ pluggy==0.13.1
     # via pytest
 pprintpp==0.4.0
     # via pytest-icdiff
+prompt-toolkit==3.0.18
+    # via
+    #   -c requirements/_base.txt
+    #   click-repl
 psycopg2-binary==2.8.6
     # via
     #   -c requirements/_base.txt
@@ -172,6 +209,8 @@ pyrsistent==0.17.3
     #   jsonschema
 pytest-aiohttp==0.3.0
     # via -r requirements/_test.in
+pytest-celery==0.0.0
+    # via -r requirements/_test.in
 pytest-cov==2.12.0
     # via -r requirements/_test.in
 pytest-docker==0.10.1
@@ -206,6 +245,10 @@ python-dotenv==0.17.1
     #   docker-compose
 python-editor==1.0.4
     # via alembic
+pytz==2021.1
+    # via
+    #   -c requirements/_base.txt
+    #   celery
 pyyaml==5.4.1
     # via
     #   -c requirements/../../../../requirements/constraints.txt
@@ -216,6 +259,7 @@ redis==3.5.3
     # via
     #   -c requirements/_base.txt
     #   -r requirements/_test.in
+    #   celery
 requests==2.25.1
     # via
     #   codecov
@@ -226,6 +270,7 @@ six==1.16.0
     # via
     #   -c requirements/_base.txt
     #   bcrypt
+    #   click-repl
     #   dockerpty
     #   isodate
     #   jsonschema
@@ -267,6 +312,15 @@ urllib3==1.26.4
     # via
     #   -c requirements/../../../../requirements/constraints.txt
     #   requests
+vine==5.0.0
+    # via
+    #   -c requirements/_base.txt
+    #   amqp
+    #   celery
+wcwidth==0.2.5
+    # via
+    #   -c requirements/_base.txt
+    #   prompt-toolkit
 websocket-client==0.59.0
     # via
     #   docker

services/web/server/src/simcore_service_webserver/activity/celery_client.py

@@ -0,0 +1,62 @@
+import logging
+from typing import Optional
+
+from aiohttp import web
+from celery import Celery
+from tenacity import AsyncRetrying, before_log, stop_after_attempt, wait_fixed
+
+from ..computation_config import ComputationSettings, get_settings
+from .config import CONFIG_SECTION_NAME
+
+__APP_CLIENT_CELERY_CLIENT_KEY = ".".join(
+    [__name__, CONFIG_SECTION_NAME, "celery_client"]
+)
+
+log = logging.getLogger(__name__)
+
+retry_upon_init_policy = dict(
+    stop=stop_after_attempt(4),
+    wait=wait_fixed(1.5),
+    before=before_log(log, logging.WARNING),
+    reraise=True,
+)
+
+
+def _get_computation_settings(app: web.Application) -> ComputationSettings:
+    return get_settings(app)
+
+
+async def _celery_app(app: web.Application):
+    comp_settings: ComputationSettings = _get_computation_settings(app)
+
+    celery_app: Optional[Celery] = None
+    async for attempt in AsyncRetrying(**retry_upon_init_policy):
+        with attempt:
+            celery_app = Celery(
+                comp_settings.task_name,
+                broker=comp_settings.broker_url,
+                backend=comp_settings.result_backend,
+            )
+            if not celery_app:
+                raise ValueError(
+                    "Expected celery client app instance, got {celery_app}"
+                )
+
+    app[__APP_CLIENT_CELERY_CLIENT_KEY] = celery_app
+
+    yield
+
+    if celery_app is not app[__APP_CLIENT_CELERY_CLIENT_KEY]:
+        log.critical("Invalid celery client in app")
+
+    celery_app.close()
+
+
+def setup(app: web.Application):
+    app[__APP_CLIENT_CELERY_CLIENT_KEY] = None
+
+    app.cleanup_ctx.append(_celery_app)
+
+
+def get_celery_client(app: web.Application) -> Celery:
+    return app[__APP_CLIENT_CELERY_CLIENT_KEY]

services/web/server/src/simcore_service_webserver/activity/config.py

@@ -2,15 +2,17 @@
     - config-file schema
     - prometheus endpoint information
 """
-from aiohttp.web import Application
+from typing import Dict
+
 import trafaret as T
+from aiohttp.web import Application
 from models_library.basic_types import PortInt, VersionTag
 from pydantic import BaseSettings
 from servicelib.application_keys import APP_CONFIG_KEY
-from typing import Dict
 
 CONFIG_SECTION_NAME = "activity"
 
+
 schema = T.Dict(
     {
         T.Key("enabled", default=True, optional=True): T.Bool(),

services/web/server/src/simcore_service_webserver/activity/handlers.py

@@ -7,8 +7,8 @@
 from servicelib.request_keys import RQT_USERID_KEY
 from yarl import URL
 
-from ..computation_api import get_celery
 from ..login.decorators import login_required
+from .celery_client import get_celery_client
 from .config import CONFIG_SECTION_NAME
 
 
@@ -19,7 +19,7 @@ async def query_prometheus(session, url, query):
 
 
 def celery_reserved(app):
-    return get_celery(app).control.inspect().reserved()
+    return get_celery_client(app).control.inspect().reserved()
 
 
 async def get_cpu_usage(session, url, user_id):
@@ -33,7 +33,8 @@ async def get_memory_usage(session, url, user_id):
 
 
 async def get_celery_reserved(app):
-    return celery_reserved(app)
+    # this can take a bit of time, blocks for a second. so it runs in an executor
+    return await asyncio.get_event_loop().run_in_executor(None, celery_reserved, app)
 
 
 async def get_container_metric_for_labels(session, url, user_id):

services/web/server/src/simcore_service_webserver/activity/module_setup.py

@@ -5,7 +5,7 @@
 from servicelib.rest_routing import iter_path_operations, map_handlers_with_operations
 
 from ..rest_config import APP_OPENAPI_SPECS_KEY
-from . import handlers
+from . import celery_client, handlers
 from .config import assert_valid_config
 
 logger = logging.getLogger(__name__)
@@ -19,11 +19,11 @@
 )
 def setup_activity(app: web.Application):
 
-    #----------------------------------------------
+    # ----------------------------------------------
     # TODO: temporary, just to check compatibility between
     # trafaret and pydantic schemas
     assert_valid_config(app)
-    #---------------------------------------------
+    # ---------------------------------------------
 
     # setup routes ------------
     specs = app[APP_OPENAPI_SPECS_KEY]
@@ -38,3 +38,5 @@ def include_path(tup_object):
         handlers_dict, filter(include_path, iter_path_operations(specs)), strict=True
     )
     app.router.add_routes(routes)
+
+    celery_client.setup(app)

services/web/server/src/simcore_service_webserver/computation_comp_tasks_listening_task.py

@@ -21,7 +21,7 @@
 from simcore_postgres_database.webserver_models import DB_CHANNEL_NAME, projects
 from sqlalchemy.sql import select
 
-from .computation_api import convert_state_from_db
+from .computation_utils import convert_state_from_db
 from .projects import projects_api, projects_exceptions
 from .projects.projects_utils import project_get_depending_nodes
 

services/web/server/src/simcore_service_webserver/computation_api.py renamed to services/web/server/src/simcore_service_webserver/computation_utils.py

@@ -1,35 +1,15 @@
 """ API of computation subsystem within this application
 
 """
-# pylint: disable=too-many-arguments
-import logging
 
-from aiohttp import web
-from celery import Celery
 from models_library.projects_state import RunningState
 from simcore_postgres_database.models.comp_pipeline import StateType
 
-from .computation_config import ComputationSettings
-from .computation_config import get_settings as get_computation_settings
-
-log = logging.getLogger(__file__)
-
-
 #
 # API ------------------------------------------
 #
 
 
-def get_celery(app: web.Application) -> Celery:
-    comp_settings: ComputationSettings = get_computation_settings(app)
-    celery_app = Celery(
-        comp_settings.task_name,
-        broker=comp_settings.broker_url,
-        backend=comp_settings.result_backend,
-    )
-    return celery_app
-
-
 DB_TO_RUNNING_STATE = {
     StateType.FAILED: RunningState.FAILED,
     StateType.PENDING: RunningState.PENDING,

services/web/server/tests/unit/isolated/test_activity.py

@@ -9,16 +9,25 @@
 import yaml
 from aiohttp import web
 from aiohttp.client_exceptions import ClientConnectionError
-
+from celery import Celery
 from pytest_simcore.helpers.utils_assert import assert_status
 from pytest_simcore.helpers.utils_mock import future_with_result
 from servicelib.application import create_safe_application
 from simcore_service_webserver.activity import handlers, setup_activity
+from simcore_service_webserver.computation_config import ComputationSettings
 from simcore_service_webserver.rest import setup_rest
 from simcore_service_webserver.security import setup_security
 from simcore_service_webserver.session import setup_session
 
 
+@pytest.fixture
+def mocked_celery_client(celery_app: Celery, mocker):
+    mocker.patch(
+        "simcore_service_webserver.activity.celery_client._get_computation_settings",
+        return_value=ComputationSettings.create_from_env(),
+    )
+
+
 @pytest.fixture
 def mocked_login_required(mocker):
     mock = mocker.patch(
@@ -82,7 +91,9 @@ def app_config(fake_data_dir: Path, osparc_simcore_root_dir: Path):
 
 
 @pytest.fixture
-def client(loop, aiohttp_client, app_config, mock_orphaned_services):
+def client(
+    mocked_celery_client, loop, aiohttp_client, app_config, mock_orphaned_services
+):
     app = create_safe_application(app_config)
 
     setup_session(app)

services/web/server/tests/unit/with_dbs/04/test_computation_api.py renamed to services/web/server/tests/unit/with_dbs/04/test_computation_utils.py

@@ -6,7 +6,7 @@
 import pytest
 from models_library.projects_state import RunningState
 from simcore_postgres_database.models.comp_pipeline import StateType
-from simcore_service_webserver.computation_api import convert_state_from_db
+from simcore_service_webserver.computation_utils import convert_state_from_db
 
 
 @pytest.mark.parametrize(