Merge branch 'master' into is6318/fix-catalog

Author: pcrespov

.env-devel

@@ -25,6 +25,8 @@ TRAEFIK_API_SERVER_INFLIGHTREQ_AMOUNT=25
 
 AUTOSCALING_DASK=null
 AUTOSCALING_DRAIN_NODES_WITH_LABELS=False
+AUTOSCALING_DOCKER_JOIN_DRAINED=True
+AUTOSCALING_WAIT_FOR_CLOUD_INIT_BEFORE_WARM_BUFFER_ACTIVATION=False
 AUTOSCALING_EC2_ACCESS=null
 AUTOSCALING_EC2_INSTANCES=null
 AUTOSCALING_LOGLEVEL=WARNING

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

@@ -28,7 +28,7 @@
     _EC2_STARTUP_MAX_WAIT_TIME + _S4L_DOCKER_PULLING_MAX_TIME + _S4L_MAX_STARTUP_TIME
 )
 _S4L_STARTUP_SCREEN_MAX_TIME: Final[int] = 45 * SECOND
-
+_S4L_COPY_WORKSPACE_TIME: Final[int] = 60 * SECOND
 
 @dataclass(kw_only=True)
 class S4LWaitForWebsocket:
@@ -84,7 +84,7 @@ def __call__(self, message: str) -> bool:
         return False
 
 
-def launch_S4L(page: Page, node_id, log_in_and_out: WebSocket, autoscaled: bool) -> Dict[str, Union[WebSocket, FrameLocator]]:
+def launch_S4L(page: Page, node_id, log_in_and_out: WebSocket, autoscaled: bool, copy_workspace: bool = False) -> Dict[str, Union[WebSocket, FrameLocator]]:
     with log_context(logging.INFO, "launch S4L") as ctx:
         predicate = S4LWaitForWebsocket(logger=ctx.logger)
         with page.expect_websocket(
@@ -95,6 +95,11 @@ def launch_S4L(page: Page, node_id, log_in_and_out: WebSocket, autoscaled: bool)
                 if autoscaled
                 else _S4L_MAX_STARTUP_TIME
             )
+            + (
+                _S4L_COPY_WORKSPACE_TIME
+                if copy_workspace
+                else 0
+            )
             + 10 * SECOND,
         ) as ws_info:
             s4l_iframe = wait_for_service_running(
@@ -105,6 +110,11 @@ def launch_S4L(page: Page, node_id, log_in_and_out: WebSocket, autoscaled: bool)
                     _S4L_AUTOSCALED_MAX_STARTUP_TIME
                     if autoscaled
                     else _S4L_MAX_STARTUP_TIME
+                )
+                + (
+                    _S4L_COPY_WORKSPACE_TIME
+                    if copy_workspace
+                    else 0
                 ),
                 press_start_button=False,
             )

services/autoscaling/src/simcore_service_autoscaling/constants.py

@@ -12,9 +12,15 @@
 )
 PREPULL_COMMAND_NAME: Final[str] = "docker images pulling"
 
-DOCKER_PULL_COMMAND: Final[
-    str
-] = "docker compose -f /docker-pull.compose.yml -p buffering pull"
+DOCKER_JOIN_COMMAND_NAME: Final[str] = "docker swarm join"
+DOCKER_JOIN_COMMAND_EC2_TAG_KEY: Final[AWSTagKey] = parse_obj_as(
+    AWSTagKey, "io.simcore.autoscaling.joined_command_sent"
+)
+
+
+DOCKER_PULL_COMMAND: Final[str] = (
+    "docker compose -f /docker-pull.compose.yml -p buffering pull"
+)
 
 PRE_PULLED_IMAGES_EC2_TAG_KEY: Final[AWSTagKey] = parse_obj_as(
     AWSTagKey, "io.simcore.autoscaling.pre_pulled_images"

services/autoscaling/src/simcore_service_autoscaling/core/settings.py

@@ -112,7 +112,7 @@ class EC2InstancesSettings(BaseCustomSettings):
     )
     EC2_INSTANCES_TIME_BEFORE_TERMINATION: datetime.timedelta = Field(
         default=datetime.timedelta(minutes=1),
-        description="Time after which an EC2 instance may being the termination process (0<=T<=59 minutes, is automatically capped)"
+        description="Time after which an EC2 instance may begin the termination process (0<=T<=59 minutes, is automatically capped)"
         "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)",
     )
     EC2_INSTANCES_TIME_BEFORE_FINAL_TERMINATION: datetime.timedelta = Field(
@@ -272,6 +272,17 @@ class ApplicationSettings(BaseCustomSettings, MixinLoggingSettings):
         "but a docker node label named osparc-services-ready is attached",
     )
 
+    AUTOSCALING_DOCKER_JOIN_DRAINED: bool = Field(
+        default=True,
+        description="If true, new nodes join the swarm as drained. If false as active.",
+    )
+
+    AUTOSCALING_WAIT_FOR_CLOUD_INIT_BEFORE_WARM_BUFFER_ACTIVATION: bool = Field(
+        default=False,
+        description="If True, then explicitely wait for cloud-init process to be completed before issuing commands. "
+        "TIP: might be useful when cheap machines are used",
+    )
+
     @cached_property
     def LOG_LEVEL(self):  # noqa: N802
         return self.AUTOSCALING_LOGLEVEL

services/autoscaling/src/simcore_service_autoscaling/modules/auto_scaling_core.py

@@ -15,11 +15,13 @@
     Resources,
 )
 from aws_library.ec2._errors import EC2TooManyInstancesError
+from aws_library.ec2._models import AWSTagValue
 from fastapi import FastAPI
 from models_library.generated_models.docker_rest_api import Node, NodeState
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.utils import limited_gather
 from servicelib.utils_formatting import timedelta_as_minute_second
+from ..constants import DOCKER_JOIN_COMMAND_EC2_TAG_KEY, DOCKER_JOIN_COMMAND_NAME
 from types_aiobotocore_ec2.literals import InstanceTypeType
 
 from ..core.errors import (
@@ -200,13 +202,17 @@ async def _make_pending_buffer_ec2s_join_cluster(
     app: FastAPI,
     cluster: Cluster,
 ) -> Cluster:
+    ec2_client = get_ec2_client(app)
     if buffer_ec2s_pending := [
         i.ec2_instance
         for i in cluster.pending_ec2s
         if is_buffer_machine(i.ec2_instance.tags)
+        and (DOCKER_JOIN_COMMAND_EC2_TAG_KEY not in i.ec2_instance.tags)
     ]:
         # started buffer instance shall be asked to join the cluster once they are running
+        app_settings = get_application_settings(app)
         ssm_client = get_ssm_client(app)
+
         buffer_ec2_connection_state = await limited_gather(
             *[
                 ssm_client.is_instance_connected_to_ssm_server(i.id)
@@ -223,27 +229,42 @@ async def _make_pending_buffer_ec2s_join_cluster(
             )
             if c is True
         ]
-        buffer_ec2_initialized = await limited_gather(
-            *[
-                ssm_client.wait_for_has_instance_completed_cloud_init(i.id)
-                for i in buffer_ec2_connected_to_ssm_server
-            ],
-            reraise=False,
-            log=_logger,
-            limit=20,
-        )
-        buffer_ec2_ready_for_command = [
-            i
-            for i, r in zip(
-                buffer_ec2_connected_to_ssm_server, buffer_ec2_initialized, strict=True
+        buffer_ec2_ready_for_command = buffer_ec2_connected_to_ssm_server
+        if app_settings.AUTOSCALING_WAIT_FOR_CLOUD_INIT_BEFORE_WARM_BUFFER_ACTIVATION:
+            buffer_ec2_initialized = await limited_gather(
+                *[
+                    ssm_client.wait_for_has_instance_completed_cloud_init(i.id)
+                    for i in buffer_ec2_connected_to_ssm_server
+                ],
+                reraise=False,
+                log=_logger,
+                limit=20,
+            )
+            buffer_ec2_ready_for_command = [
+                i
+                for i, r in zip(
+                    buffer_ec2_connected_to_ssm_server,
+                    buffer_ec2_initialized,
+                    strict=True,
+                )
+                if r is True
+            ]
+        if buffer_ec2_ready_for_command:
+            ssm_command = await ssm_client.send_command(
+                [i.id for i in buffer_ec2_ready_for_command],
+                command=await utils_docker.get_docker_swarm_join_bash_command(
+                    join_as_drained=app_settings.AUTOSCALING_DOCKER_JOIN_DRAINED
+                ),
+                command_name=DOCKER_JOIN_COMMAND_NAME,
+            )
+            await ec2_client.set_instances_tags(
+                buffer_ec2_ready_for_command,
+                tags={
+                    DOCKER_JOIN_COMMAND_EC2_TAG_KEY: AWSTagValue(
+                        ssm_command.command_id
+                    ),
+                },
             )
-            if r is True
-        ]
-        await ssm_client.send_command(
-            [i.id for i in buffer_ec2_ready_for_command],
-            command=await utils_docker.get_docker_swarm_join_bash_command(),
-            command_name="docker swarm join",
-        )
     return cluster
 
 
@@ -308,14 +329,14 @@ async def sorted_allowed_instance_types(app: FastAPI) -> list[EC2InstanceType]:
     ec2_client = get_ec2_client(app)
 
     # some instances might be able to run several tasks
-    allowed_instance_types: list[
-        EC2InstanceType
-    ] = await ec2_client.get_ec2_instance_capabilities(
-        cast(
-            set[InstanceTypeType],
-            set(
-                app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES,
-            ),
+    allowed_instance_types: list[EC2InstanceType] = (
+        await ec2_client.get_ec2_instance_capabilities(
+            cast(
+                set[InstanceTypeType],
+                set(
+                    app_settings.AUTOSCALING_EC2_INSTANCES.EC2_INSTANCES_ALLOWED_TYPES,
+                ),
+            )
         )
     )
 

services/autoscaling/src/simcore_service_autoscaling/modules/buffer_machines_pool_core.py

@@ -33,10 +33,6 @@
 from fastapi import FastAPI
 from pydantic import NonNegativeInt
 from servicelib.logging_utils import log_context
-from simcore_service_autoscaling.modules.instrumentation import (
-    get_instrumentation,
-    has_instrumentation,
-)
 from types_aiobotocore_ec2.literals import InstanceTypeType
 
 from ..constants import (
@@ -55,6 +51,7 @@
 )
 from .auto_scaling_mode_base import BaseAutoscaling
 from .ec2 import get_ec2_client
+from .instrumentation import get_instrumentation, has_instrumentation
 from .ssm import get_ssm_client
 
 _logger = logging.getLogger(__name__)
@@ -197,8 +194,9 @@ async def _terminate_instances_with_invalid_pre_pulled_images(
         ].pre_pulled_instances()
 
         for instance in all_pre_pulled_instances:
+            pre_pulled_images = load_pre_pulled_images_from_tags(instance.tags)
             if (
-                pre_pulled_images := load_pre_pulled_images_from_tags(instance.tags)
+                pre_pulled_images is not None
             ) and pre_pulled_images != ec2_boot_config.pre_pull_images:
                 _logger.info(
                     "%s",

services/autoscaling/src/simcore_service_autoscaling/utils/auto_scaling_core.py

@@ -80,7 +80,11 @@ async def ec2_startup_script(
     ec2_boot_specific: EC2InstanceBootSpecific, app_settings: ApplicationSettings
 ) -> str:
     startup_commands = ec2_boot_specific.custom_boot_scripts.copy()
-    startup_commands.append(await utils_docker.get_docker_swarm_join_bash_command())
+    startup_commands.append(
+        await utils_docker.get_docker_swarm_join_bash_command(
+            join_as_drained=app_settings.AUTOSCALING_DOCKER_JOIN_DRAINED
+        )
+    )
     if app_settings.AUTOSCALING_REGISTRY:  # noqa: SIM102
         if pull_image_cmd := utils_docker.get_docker_pull_images_on_start_bash_command(
             ec2_boot_specific.pre_pull_images

services/autoscaling/src/simcore_service_autoscaling/utils/utils_docker.py

@@ -392,7 +392,7 @@ async def compute_cluster_used_resources(
 _DOCKER_SWARM_JOIN_PATTERN = re.compile(_DOCKER_SWARM_JOIN_RE)
 
 
-async def get_docker_swarm_join_bash_command() -> str:
+async def get_docker_swarm_join_bash_command(*, join_as_drained: bool) -> str:
     """this assumes we are on a manager node"""
     command = ["docker", "swarm", "join-token", "worker"]
     process = await asyncio.create_subprocess_exec(
@@ -409,7 +409,7 @@ async def get_docker_swarm_join_bash_command() -> str:
     decoded_stdout = stdout.decode()
     if match := re.search(_DOCKER_SWARM_JOIN_PATTERN, decoded_stdout):
         capture = match.groupdict()
-        return f"{capture['command']} --availability=drain {capture['token']} {capture['address']}"
+        return f"{capture['command']} --availability={'drain' if join_as_drained  else 'active'} {capture['token']} {capture['address']}"
     msg = f"expected docker '{_DOCKER_SWARM_JOIN_RE}' command not found: received {decoded_stdout}!"
     raise RuntimeError(msg)
 

services/autoscaling/tests/manual/.env-devel

@@ -1,6 +1,9 @@
 AUTOSCALING_DEBUG=true
+AUTOSCALING_DRAIN_NODES_WITH_LABELS=False
+AUTOSCALING_DOCKER_JOIN_DRAINED=True
+AUTOSCALING_WAIT_FOR_CLOUD_INIT_BEFORE_WARM_BUFFER_ACTIVATION=False
 AUTOSCALING_LOGLEVEL=INFO
-AUTOSCALING_TASK_INTERVAL=30
+AUTOSCALING_POLL_INTERVAL=10
 AUTOSCALING_EC2_ACCESS_KEY_ID=XXXXXXXXXX
 AUTOSCALING_EC2_SECRET_ACCESS_KEY=XXXXXXXXXX
 AUTOSCALING_EC2_ENDPOINT=null
@@ -13,20 +16,21 @@ EC2_INSTANCES_MACHINES_BUFFER=0
 EC2_INSTANCES_MAX_INSTANCES=20
 EC2_INSTANCES_TIME_BEFORE_DRAINING="00:00:10"
 EC2_INSTANCES_TIME_BEFORE_TERMINATION="00:03:00"
-EC2_INSTANCES_ALLOWED_TYPES='{"t2.micro": {"ami_id": "XXXXXXXX", "custom_boot_scripts": ["whoami"], "pre_pull_images": ["ubuntu:latest"]}}'
+EC2_INSTANCES_ALLOWED_TYPES={"t2.micro": {"ami_id": "XXXXXXXX", "custom_boot_scripts": ["whoami"], "pre_pull_images": ["ubuntu:latest"]}}
 EC2_INSTANCES_ATTACHED_IAM_PROFILE=XXXXXXXXX
 EC2_INSTANCES_KEY_NAME=XXXXXXXXXX
 EC2_INSTANCES_NAME_PREFIX=testing-osparc-computational-cluster
-EC2_INSTANCES_SECURITY_GROUP_IDS="[\"XXXXXXXXXX\"]"
+EC2_INSTANCES_SECURITY_GROUP_IDS=["XXXXXXXXXX"]
 EC2_INSTANCES_SUBNET_ID=XXXXXXXXXX
-EC2_INSTANCES_CUSTOM_TAGS='{"special": "testing"}'
+EC2_INSTANCES_CUSTOM_TAGS={"special": "testing"}
 EC2_INSTANCES_TIME_BEFORE_DRAINING=00:00:20
 EC2_INSTANCES_TIME_BEFORE_TERMINATION=00:01:00
 LOG_FORMAT_LOCAL_DEV_ENABLED=True
 # define the following to activate dynamic autoscaling
-# NODES_MONITORING_NEW_NODES_LABELS="[\"testing.autoscaled-node\"]"
-# NODES_MONITORING_NODE_LABELS="[\"testing.monitored-node\"]"
-# NODES_MONITORING_SERVICE_LABELS="[\"testing.monitored-service\"]"
+# AUTOSCALING_NODES_MONITORING={}
+# NODES_MONITORING_NEW_NODES_LABELS=["testing.autoscaled-node"]
+# NODES_MONITORING_NODE_LABELS=["testing.monitored-node"]
+# NODES_MONITORING_SERVICE_LABELS=["testing.monitored-service"]
 
 # may be activated or not
 # RABBIT_HOST=rabbit

services/autoscaling/tests/manual/README.md

@@ -109,8 +109,9 @@ make up-devel # this will deploy the autoscaling stack
 docker service create \
 --name=test-service \
 --reserve-cpu=1 \
---reserve-memory=1GiB \
+--reserve-memory=512MiB \
 --constraint=node.labels.testing.monitored-node==true \
+--constraint=node.labels.io.simcore.osparc-services-ready==true \
 --label=testing.monitored-service=true \
 --container-label=io.simcore.runtime.user-id=99 \
 --container-label=io.simcore.runtime.project-id='5054a589-3ba4-46c3-829d-2e3d1a6a043f' \
@@ -120,5 +121,5 @@ docker service create \
 --container-label=io.simcore.runtime.swarm-stack-name=thestack \
 --container-label=io.simcore.runtime.memory-limit=1GB \
 --container-label=io.simcore.runtime.cpu-limit=1 \
-redis # will create a redis service reserving 4 CPUs and 1GiB of RAM
+redis # will create a redis service reserving 1 CPUs and 512MiB of RAM
 ```

services/autoscaling/tests/unit/conftest.py

@@ -147,6 +147,30 @@ def with_labelize_drain_nodes(
     )
 
 
+@pytest.fixture(
+    params=[
+        "with_AUTOSCALING_DOCKER_JOIN_DRAINED",
+        "without_AUTOSCALING_DOCKER_JOIN_DRAINED",
+    ]
+)
+def with_docker_join_drained(request: pytest.FixtureRequest) -> bool:
+    return bool(request.param == "with_AUTOSCALING_DOCKER_JOIN_DRAINED")
+
+
+@pytest.fixture
+def app_with_docker_join_drained(
+    app_environment: EnvVarsDict,
+    monkeypatch: pytest.MonkeyPatch,
+    with_docker_join_drained: bool,
+) -> EnvVarsDict:
+    return app_environment | setenvs_from_dict(
+        monkeypatch,
+        {
+            "AUTOSCALING_DOCKER_JOIN_DRAINED": f"{with_docker_join_drained}",
+        },
+    )
+
+
 @pytest.fixture(scope="session")
 def fake_ssm_settings() -> SSMSettings:
     return SSMSettings(**SSMSettings.Config.schema_extra["examples"][0])

services/autoscaling/tests/unit/test_modules_auto_scaling_computational.py

@@ -70,6 +70,7 @@ def local_dask_scheduler_server_envs(
 @pytest.fixture
 def minimal_configuration(
     with_labelize_drain_nodes: EnvVarsDict,
+    app_with_docker_join_drained: EnvVarsDict,
     docker_swarm: None,
     mocked_ec2_server_envs: EnvVarsDict,
     mocked_ssm_server_envs: EnvVarsDict,

services/autoscaling/tests/unit/test_modules_auto_scaling_dynamic.py

@@ -181,6 +181,7 @@ async def drained_host_node(
 @pytest.fixture
 def minimal_configuration(
     with_labelize_drain_nodes: EnvVarsDict,
+    app_with_docker_join_drained: EnvVarsDict,
     docker_swarm: None,
     mocked_ec2_server_envs: EnvVarsDict,
     mocked_ssm_server_envs: EnvVarsDict,
@@ -1010,6 +1011,7 @@ async def test_cluster_scaling_up_and_down_against_aws(
     skip_if_external_envfile_dict: None,
     external_ec2_instances_allowed_types: None | dict[str, EC2InstanceBootSpecific],
     with_labelize_drain_nodes: EnvVarsDict,
+    app_with_docker_join_drained: EnvVarsDict,
     docker_swarm: None,
     disabled_rabbitmq: None,
     disable_dynamic_service_background_task: None,