Enable orphan service removal + making stop service more solid (#1842)

GitHK · Andrei Neagu · web-flow · commit ed0b00c0b3a8 · 2020-09-30T08:10:26.000+02:00
* orpahaned services removal enabled again

* decorator makes async function calls run in sequnce

* requests to stop_service the same service are now serialized

* fixed typo

* corrected parameter name

* fix incorrect usage:
was considering all variables in the scope of the function
it will now onluy consider the function's arguments

* removed out of date comment

* function got a more appropriate name

* getting codeclimate unstuck :\

Co-authored-by: Andrei Neagu &lt;neagu@itis.swiss&gt;
diff --git a/README.md b/README.md
@@ -128,3 +128,4 @@ This project is licensed under the terms of the [MIT license](LICENSE).
 [chocolatey]:https://chocolatey.org/
 [vscode]:https://code.visualstudio.com/
 [WSL]:https://docs.microsoft.com/en-us/windows/wsl/faq
+
diff --git a/packages/service-library/src/servicelib/async_utils.py b/packages/service-library/src/servicelib/async_utils.py
@@ -0,0 +1,113 @@
+import asyncio
+from functools import wraps
+from collections import deque
+import attr
+from typing import List, Dict
+
+
+@attr.s(auto_attribs=True)
+class Context:
+    in_queue: asyncio.Queue
+    out_queue: asyncio.Queue
+    initialized: bool
+
+
+def run_sequentially_in_context(target_args: List[str] = None):
+    """All request to function with same calling context will be run sequentially.
+
+    Example:
+
+    Given the following decorated function
+
+        @run_sequentially_in_context(target_args=["param3", "param1"])
+        async def func(param1, param2, param3):
+            await asyncio.sleep(1)
+
+    The context will be formed by the values of the arguments "param3" and "param1".
+    The values must be serializable as they will be converted to string
+    and put together as storage key for the context.
+
+    The below calls will all run in a sequence:
+
+        functions = [
+            func(1, "something", 3),
+            func(1, "else", 3),
+            func(1, "here", 3),
+        ]
+        await asyncio.gather(*functions)
+
+    The following calls will run in parallel, because they have different contexts:
+
+        functions = [
+            func(1, "something", 3),
+            func(2, "else", 3),
+            func(3, "here", 3),
+        ]
+        await asyncio.gather(*functions)
+
+    """
+    target_args = [] if target_args is None else target_args
+
+    def internal(decorated_function):
+        contexts = {}
+
+        def get_context(args, kwargs: Dict) -> Context:
+            arg_names = decorated_function.__code__.co_varnames[
+                : decorated_function.__code__.co_argcount
+            ]
+            search_args = dict(zip(arg_names, args))
+            search_args.update(kwargs)
+
+            key_parts = deque()
+            for arg in target_args:
+                if arg not in search_args:
+                    message = (
+                        f"Expected '{arg}' in '{decorated_function.__name__}'"
+                        f" arguments. Got '{search_args}'"
+                    )
+                    raise ValueError(message)
+                key_parts.append(search_args[arg])
+
+            key = ":".join(map(str, key_parts))
+
+            if key not in contexts:
+                contexts[key] = Context(
+                    in_queue=asyncio.Queue(),
+                    out_queue=asyncio.Queue(),
+                    initialized=False,
+                )
+
+            return contexts[key]
+
+        @wraps(decorated_function)
+        async def wrapper(*args, **kwargs):
+            context: Context = get_context(args, kwargs)
+
+            if not context.initialized:
+                context.initialized = True
+
+                async def worker(in_q: asyncio.Queue, out_q: asyncio.Queue):
+                    while True:
+                        awaitable = await in_q.get()
+                        in_q.task_done()
+                        try:
+                            result = await awaitable
+                        except Exception as e:  # pylint: disable=broad-except
+                            result = e
+                        await out_q.put(result)
+
+                asyncio.get_event_loop().create_task(
+                    worker(context.in_queue, context.out_queue)
+                )
+
+            await context.in_queue.put(decorated_function(*args, **kwargs))
+
+            wrapped_result = await context.out_queue.get()
+            if isinstance(wrapped_result, Exception):
+                raise wrapped_result
+
+            return wrapped_result
+
+        return wrapper
+
+    return internal
diff --git a/packages/service-library/tests/test_async_utils.py b/packages/service-library/tests/test_async_utils.py
@@ -0,0 +1,150 @@
+import asyncio
+import copy
+import random
+from collections import deque
+from time import time
+from typing import Any, Dict, List
+
+import pytest
+
+from servicelib.async_utils import run_sequentially_in_context
+
+
+class LockedStore:
+    __slots__ = ("_queue", "_lock")
+
+    def __init__(self):
+        self._queue = deque()
+        self._lock = asyncio.Lock()
+
+    async def push(self, item: Any):
+        async with self._lock:
+            self._queue.append(item)
+
+    async def get_all(self) -> List[Any]:
+        async with self._lock:
+            return list(self._queue)
+
+
+async def test_context_aware_dispatch() -> None:
+    @run_sequentially_in_context(target_args=["c1", "c2", "c3"])
+    async def orderly(c1: Any, c2: Any, c3: Any, control: Any) -> None:
+        _ = (c1, c2, c3)
+        sleep_interval = random.uniform(0, 0.01)
+        await asyncio.sleep(sleep_interval)
+
+        context = dict(c1=c1, c2=c2, c3=c3)
+        await locked_stores[make_key_from_context(context)].push(control)
+
+    def make_key_from_context(context: Dict) -> str:
+        return ".".join([f"{k}:{v}" for k, v in context.items()])
+
+    def make_context():
+        return dict(
+            c1=random.randint(0, 10), c2=random.randint(0, 10), c3=random.randint(0, 10)
+        )
+
+    contexts = [make_context() for _ in range(10)]
+
+    locked_stores = {}
+    expected_outcomes = {}
+    for context in contexts:
+        key = make_key_from_context(context)
+        locked_stores[key] = LockedStore()
+        expected_outcomes[key] = deque()
+
+    tasks = deque()
+    for control in range(1000):
+        context = random.choice(contexts)
+        key = make_key_from_context(context)
+        expected_outcomes[key].append(control)
+
+        params = copy.deepcopy(context)
+        params["control"] = control
+
+        task = asyncio.get_event_loop().create_task(orderly(**params))
+        tasks.append(task)
+
+    for task in tasks:
+        await task
+
+    for context in contexts:
+        key = make_key_from_context(context)
+        assert list(expected_outcomes[key]) == await locked_stores[key].get_all()
+
+
+async def test_context_aware_function_sometimes_fails() -> None:
+    class DidFailException(Exception):
+        pass
+
+    @run_sequentially_in_context(target_args=["will_fail"])
+    async def sometimes_failing(will_fail: bool) -> None:
+        if will_fail:
+            raise DidFailException("I was instructed to fail")
+        return True
+
+    for x in range(100):
+        raise_error = x % 2 == 0
+
+        if raise_error:
+            with pytest.raises(DidFailException):
+                await sometimes_failing(raise_error)
+        else:
+            assert await sometimes_failing(raise_error) is True
+
+
+async def test_context_aware_wrong_target_args_name() -> None:
+    expected_param_name = "wrong_parameter"
+
+    # pylint: disable=unused-argument
+    @run_sequentially_in_context(target_args=[expected_param_name])
+    async def target_function(the_param: Any) -> None:
+        return None
+
+    with pytest.raises(ValueError) as excinfo:
+        await target_function("something")
+
+    message = (
+        f"Expected '{expected_param_name}' in "
+        f"'{target_function.__name__}' arguments."
+    )
+    assert str(excinfo.value).startswith(message) is True
+
+
+async def test_context_aware_measure_parallelism() -> None:
+    # expected duration 1 second
+    @run_sequentially_in_context(target_args=["control"])
+    async def sleep_for(sleep_interval: float, control: Any) -> Any:
+        await asyncio.sleep(sleep_interval)
+        return control
+
+    control_sequence = list(range(1000))
+    sleep_duration = 0.5
+    functions = [sleep_for(sleep_duration, x) for x in control_sequence]
+
+    start = time()
+    result = await asyncio.gather(*functions)
+    elapsed = time() - start
+
+    assert elapsed < sleep_duration * 2  # allow for some internal delay
+    assert control_sequence == result
+
+
+async def test_context_aware_measure_serialization() -> None:
+    # expected duration 1 second
+    @run_sequentially_in_context(target_args=["control"])
+    async def sleep_for(sleep_interval: float, control: Any) -> Any:
+        await asyncio.sleep(sleep_interval)
+        return control
+
+    control_sequence = [1 for _ in range(10)]
+    sleep_duration = 0.1
+    functions = [sleep_for(sleep_duration, x) for x in control_sequence]
+
+    start = time()
+    result = await asyncio.gather(*functions)
+    elapsed = time() - start
+
+    minimum_timelapse = (sleep_duration) * len(control_sequence)
+    assert elapsed > minimum_timelapse
+    assert control_sequence == result
diff --git a/services/director/src/simcore_service_director/producer.py b/services/director/src/simcore_service_director/producer.py
@@ -11,6 +11,7 @@
 import aiodocker
 import tenacity
 from aiohttp import ClientConnectionError, ClientSession, web
+from servicelib.async_utils import run_sequentially_in_context
 
 from servicelib.monitor_services import service_started, service_stopped
 
@@ -914,6 +915,7 @@ async def get_service_details(app: web.Application, node_uuid: str) -> Dict:
             ) from err
 
 
+@run_sequentially_in_context(target_args=["node_uuid"])
 async def stop_service(app: web.Application, node_uuid: str) -> None:
     log.debug("stopping service with uuid %s", node_uuid)
     # get the docker client
diff --git a/services/web/server/src/simcore_service_webserver/resource_manager/garbage_collector.py b/services/web/server/src/simcore_service_webserver/resource_manager/garbage_collector.py
@@ -82,12 +82,12 @@ async def garbage_collector_task(app: web.Application):
 
 async def collect_garbage(registry: RedisResourceRegistry, app: web.Application):
     """
-    Garbage collection has the task of removing trash from the system. The trash 
+    Garbage collection has the task of removing trash from the system. The trash
     can be divided in:
-    
+
     - Websockets & Redis (used to keep track of current active connections)
     - GUEST users (used for temporary access to the system which are created on the fly)
-    - deletion of users. If a user needs to be deleted it is manually marked as GUEST 
+    - deletion of users. If a user needs to be deleted it is manually marked as GUEST
         in the database
 
     The resources are Redis entries where all information regarding all the
@@ -99,7 +99,7 @@ async def collect_garbage(registry: RedisResourceRegistry, app: web.Application)
     endpoint to refresh the TTL, thus declaring that the user (websocket connection) is
     still active. The `resource_deletion_timeout_seconds` is theTTL of the key.
 
-    The field `garbage_collection_interval_seconds` defines the interval at which this 
+    The field `garbage_collection_interval_seconds` defines the interval at which this
     function will be called.
     """
     logger.info("collecting garbage...")
@@ -122,7 +122,7 @@ async def collect_garbage(registry: RedisResourceRegistry, app: web.Application)
     # Temporary disabling GC to until the dynamic service
     # safe function is invoked by the GC. This will avoid
     # data loss for current users.
-    # await remove_orphaned_services(registry, app)
+    await remove_orphaned_services(registry, app)
 
 
 async def remove_disconnected_user_resources(
@@ -189,7 +189,8 @@ async def remove_disconnected_user_resources(
                 # if this user was a GUEST also remove it from the database
                 # with the only associated project owned
                 await remove_guest_user_with_all_its_resources(
-                    app=app, user_id=int(dead_key["user_id"]),
+                    app=app,
+                    user_id=int(dead_key["user_id"]),
                 )
 
 
@@ -218,7 +219,8 @@ async def remove_users_manually_marked_as_guests(
             continue
 
         await remove_guest_user_with_all_its_resources(
-            app=app, user_id=guest_user_id,
+            app=app,
+            user_id=guest_user_id,
         )
 
 
@@ -283,11 +285,11 @@ async def remove_guest_user_with_all_its_resources(
 async def remove_all_projects_for_user(app: web.Application, user_id: int) -> None:
     """
     Goes through all the projects and will try to remove them but first it will check if
-    the project is shared with others. 
+    the project is shared with others.
     Based on the given access rights it will deltermine the action to take:
     - if other users have read access & execute access it will get deleted
     - if other users have write access the project's owner will be changed to a new owner:
-        - if the project is directly shared with a one or more users, one of these 
+        - if the project is directly shared with a one or more users, one of these
             will be picked as the new owner
         - if the project is not shared with any user but with groups of users, one
             of the users inside the group (which currently exists) will be picked as
@@ -309,12 +311,15 @@ async def remove_all_projects_for_user(app: web.Application, user_id: int) -> No
         APP_PROJECT_DBAPI
     ].list_all_projects_by_uuid_for_user(user_id=user_id)
     logger.info(
-        "Project uuids, to clean, for user '%s': '%s'", user_id, user_project_uuids,
+        "Project uuids, to clean, for user '%s': '%s'",
+        user_id,
+        user_project_uuids,
     )
 
     for project_uuid in user_project_uuids:
         logger.debug(
-            "Removing or transfering project '%s'", project_uuid,
+            "Removing or transfering project '%s'",
+            project_uuid,
         )
         try:
             project: Dict = await get_project_for_user(
@@ -411,7 +416,9 @@ async def get_new_project_owner_gid(
     # fallback to the groups search if the user does not exist
     if len(standard_groups) > 0 and new_project_owner_gid is None:
         new_project_owner_gid = await fetch_new_project_owner_from_groups(
-            app=app, standard_groups=standard_groups, user_id=user_id,
+            app=app,
+            standard_groups=standard_groups,
+            user_id=user_id,
         )
 
     logger.info(
@@ -483,7 +490,8 @@ async def replace_current_owner(
     # syncing back project data
     try:
         await app[APP_PROJECT_DBAPI].update_project_without_enforcing_checks(
-            project_data=project, project_uuid=project_uuid,
+            project_data=project,
+            project_uuid=project_uuid,
         )
     except Exception:  # pylint: disable=broad-except
         logger.exception(
@@ -500,4 +508,3 @@ async def remove_user(app: web.Application, user_id: int) -> None:
         logger.warning(
             "User '%s' still has some projects, could not be deleted", user_id
         )
-
