Merge branch 'master' into is8496/use-prefork-pool-for-cpu-bound-celery-tasks

Author: giancarloromeo

packages/aws-library/src/aws_library/ec2/_models.py

@@ -157,16 +157,9 @@ class EC2InstanceBootSpecific(BaseModel):
         list[DockerGenericTag],
         Field(
             default_factory=list,
-            description="a list of docker image/tags to pull on instance cold start",
+            description="a list of docker image/tags to pull on the instance",
         ),
     ] = DEFAULT_FACTORY
-    pre_pull_images_cron_interval: Annotated[
-        datetime.timedelta,
-        Field(
-            description="time interval between pulls of images (minimum is 1 minute) "
-            "(default to seconds, or see https://pydantic-docs.helpmanual.io/usage/types/#datetime-types for string formating)",
-        ),
-    ] = datetime.timedelta(minutes=30)
     buffer_count: Annotated[
         NonNegativeInt,
         Field(description="number of buffer EC2s to keep (defaults to 0)"),
@@ -180,7 +173,9 @@ def validate_bash_calls(cls, v):
                 temp_file.writelines(v)
                 temp_file.flush()
                 # NOTE: this will not capture runtime errors, but at least some syntax errors such as invalid quotes
-                sh.bash("-n", temp_file.name)  # pyright: ignore[reportCallIssue]  # sh is untyped, but this call is safe for bash syntax checking
+                sh.bash(
+                    "-n", temp_file.name
+                )  # pyright: ignore[reportCallIssue]  # sh is untyped, but this call is safe for bash syntax checking
         except sh.ErrorReturnCode as exc:
             msg = f"Invalid bash call in custom_boot_scripts: {v}, Error: {exc.stderr}"
             raise ValueError(msg) from exc
@@ -231,7 +226,7 @@ def _update_json_schema_extra(schema: JsonDict) -> None:
                             "simcore/services/dynamic/another-nice-one:2.4.5",
                             "asd",
                         ],
-                        "pre_pull_images_cron_interval": "01:00:00",
+                        "pre_pull_images_cron_interval": "01:00:00",  # retired but kept for tests
                     },
                     {
                         # AMI + pre-pull + buffer count

packages/aws-library/src/aws_library/ssm/_client.py

@@ -107,7 +107,6 @@ async def send_command(
     @log_decorator(_logger, logging.DEBUG)
     @ssm_exception_handler(_logger)
     async def get_command(self, instance_id: str, *, command_id: str) -> SSMCommand:
-
         response = await self._client.get_command_invocation(
             CommandId=command_id, InstanceId=instance_id
         )
@@ -130,6 +129,13 @@ async def get_command(self, instance_id: str, *, command_id: str) -> SSMCommand:
             ),
         )
 
+    @log_decorator(_logger, logging.DEBUG)
+    @ssm_exception_handler(_logger)
+    async def cancel_command(self, instance_id: str, *, command_id: str) -> None:
+        await self._client.cancel_command(
+            CommandId=command_id, InstanceIds=[instance_id]
+        )
+
     @log_decorator(_logger, logging.DEBUG)
     @ssm_exception_handler(_logger)
     async def is_instance_connected_to_ssm_server(self, instance_id: str) -> bool:

packages/aws-library/tests/test_ec2_client.py

@@ -604,7 +604,11 @@ async def test_set_instance_tags(
     # now remove some real ones
     tag_key_to_remove = random.choice(list(new_tags))  # noqa: S311
     await simcore_ec2_api.remove_instances_tags(
-        created_instances, tag_keys=[tag_key_to_remove]
+        created_instances,
+        tag_keys=[
+            tag_key_to_remove,
+            TypeAdapter(AWSTagKey).validate_python("whatever_i_dont_exist"),
+        ],
     )
     new_tags.pop(tag_key_to_remove)
     await _assert_instances_in_ec2(

packages/aws-library/tests/test_ssm_client.py

@@ -37,8 +37,7 @@ async def simcore_ssm_api(
     await ec2.close()
 
 
-async def test_ssm_client_lifespan(simcore_ssm_api: SimcoreSSMAPI):
-    ...
+async def test_ssm_client_lifespan(simcore_ssm_api: SimcoreSSMAPI): ...
 
 
 async def test_aiobotocore_ssm_client_when_ssm_server_goes_up_and_down(
@@ -125,6 +124,30 @@ async def test_send_command(
         )
 
 
+async def test_cancel_command(
+    mocked_aws_server: ThreadedMotoServer,
+    simcore_ssm_api: SimcoreSSMAPI,
+    faker: Faker,
+):
+    command_name = faker.word()
+    target_instance_id = faker.pystr()
+    sent_command = await simcore_ssm_api.send_command(
+        instance_ids=[target_instance_id],
+        command=faker.text(),
+        command_name=command_name,
+    )
+    assert sent_command
+    assert sent_command.command_id
+    assert sent_command.name == command_name
+    assert sent_command.instance_ids == [target_instance_id]
+    assert sent_command.status == "Success"
+
+    # cancelling a finished command is a no-op but is a bit of a joke as moto does not implement cancel command yet
+    await simcore_ssm_api.cancel_command(
+        instance_id=target_instance_id, command_id=sent_command.command_id
+    )
+
+
 async def test_is_instance_connected_to_ssm_server(
     mocked_aws_server: ThreadedMotoServer,
     simcore_ssm_api: SimcoreSSMAPI,

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

@@ -1,7 +1,7 @@
 import base64
 from collections.abc import Sequence
 
-from common_library.json_serialization import json_dumps
+from common_library.json_serialization import json_loads
 from models_library.docker import DockerGenericTag
 from types_aiobotocore_ec2 import EC2Client
 from types_aiobotocore_ec2.literals import InstanceStateNameType, InstanceTypeType
@@ -46,6 +46,7 @@ async def assert_autoscaled_dynamic_ec2_instances(
     expected_instance_type: InstanceTypeType,
     expected_instance_state: InstanceStateNameType,
     expected_additional_tag_keys: list[str],
+    expected_pre_pulled_images: list[DockerGenericTag] | None = None,
     instance_filters: Sequence[FilterTypeDef] | None,
     expected_user_data: list[str] | None = None,
     check_reservation_index: int | None = None,
@@ -64,6 +65,7 @@ async def assert_autoscaled_dynamic_ec2_instances(
             *expected_additional_tag_keys,
         ],
         expected_user_data=expected_user_data,
+        expected_pre_pulled_images=expected_pre_pulled_images,
         instance_filters=instance_filters,
         check_reservation_index=check_reservation_index,
     )
@@ -142,10 +144,9 @@ def _by_pre_pull_image(ec2_tag: TagTypeDef) -> bool:
                 iter(filter(_by_pre_pull_image, instance["Tags"]))
             )
             assert "Value" in instance_pre_pulled_images_aws_tag
-            assert (
-                instance_pre_pulled_images_aws_tag["Value"]
-                == f"{json_dumps(expected_pre_pulled_images)}"
-            )
+            assert sorted(
+                json_loads(instance_pre_pulled_images_aws_tag["Value"])
+            ) == sorted(expected_pre_pulled_images)
 
         assert "PrivateDnsName" in instance
         instance_private_dns_name = instance["PrivateDnsName"]

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

@@ -55,6 +55,16 @@ def _patch_describe_instance_information(
     return {"InstanceInformationList": [{"PingStatus": "Online"}]}
 
 
+def _patch_cancel_command(self, operation_name, api_params) -> dict[str, Any]:
+    warnings.warn(
+        "moto is missing the cancel_command function, therefore it is manually mocked."
+        "TIP: periodically check if it gets updated https://docs.getmoto.org/en/latest/docs/services/ssm.html#ssm",
+        UserWarning,
+        stacklevel=1,
+    )
+    return {}
+
+
 # Mocked aiobotocore _make_api_call function
 async def patched_aiobotocore_make_api_call(self, operation_name, api_params):
     # For example for the Access Analyzer service
@@ -63,6 +73,8 @@ async def patched_aiobotocore_make_api_call(self, operation_name, api_params):
     # Rationale -> https://github.com/boto/botocore/blob/develop/botocore/client.py#L810:L816
     if operation_name == "SendCommand":
         return await _patch_send_command(self, operation_name, api_params)
+    if operation_name == "CancelCommand":
+        return _patch_cancel_command(self, operation_name, api_params)
     if operation_name == "DescribeInstanceInformation":
         return _patch_describe_instance_information(self, operation_name, api_params)
 

services/autoscaling/src/simcore_service_autoscaling/constants.py

@@ -1,13 +1,14 @@
 import re
+from pathlib import Path
 from typing import Final
 
 from aws_library.ec2._models import AWSTagKey, AWSTagValue, EC2Tags
 from pydantic import TypeAdapter
 
-BUFFER_MACHINE_PULLING_EC2_TAG_KEY: Final[AWSTagKey] = TypeAdapter(
-    AWSTagKey
-).validate_python("pulling")
-BUFFER_MACHINE_PULLING_COMMAND_ID_EC2_TAG_KEY: Final[AWSTagKey] = TypeAdapter(
+MACHINE_PULLING_EC2_TAG_KEY: Final[AWSTagKey] = TypeAdapter(AWSTagKey).validate_python(
+    "pulling"
+)
+MACHINE_PULLING_COMMAND_ID_EC2_TAG_KEY: Final[AWSTagKey] = TypeAdapter(
     AWSTagKey
 ).validate_python("ssm-command-id")
 PREPULL_COMMAND_NAME: Final[str] = "docker images pulling"
@@ -17,10 +18,14 @@
     AWSTagKey
 ).validate_python("io.simcore.autoscaling.joined_command_sent")
 
+DOCKER_COMPOSE_CMD: Final[str] = "docker compose"
+PRE_PULL_COMPOSE_PATH: Final[Path] = Path("/docker-pull.compose.yml")
+DOCKER_COMPOSE_PULL_SCRIPT_PATH: Final[Path] = Path("/docker-pull-script.sh")
 
-DOCKER_PULL_COMMAND: Final[
-    str
-] = "docker compose -f /docker-pull.compose.yml -p buffering pull"
+
+DOCKER_PULL_COMMAND: Final[str] = (
+    f"{DOCKER_COMPOSE_CMD} -f {PRE_PULL_COMPOSE_PATH} -p buffering pull"
+)
 
 PRE_PULLED_IMAGES_EC2_TAG_KEY: Final[AWSTagKey] = TypeAdapter(
     AWSTagKey

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

@@ -1,5 +1,6 @@
 import logging
 
+from common_library.json_serialization import json_dumps
 from fastapi import FastAPI
 from servicelib.fastapi.tracing import (
     initialize_fastapi_app_tracing,
@@ -32,7 +33,7 @@
 from ..modules.ssm import setup as setup_ssm
 from .settings import ApplicationSettings
 
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
 
 
 def create_app(settings: ApplicationSettings, tracing_config: TracingConfig) -> FastAPI:
@@ -49,6 +50,10 @@ def create_app(settings: ApplicationSettings, tracing_config: TracingConfig) ->
     app.state.settings = settings
     app.state.tracing_config = tracing_config
     assert app.state.settings.API_VERSION == API_VERSION  # nosec
+    _logger.info(
+        "Application settings: %s",
+        json_dumps(settings, indent=2, sort_keys=True),
+    )
 
     # PLUGINS SETUP
     if tracing_config.tracing_enabled:

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

@@ -3,11 +3,12 @@
 from typing import Annotated, Final, Self, cast
 
 from aws_library.ec2 import EC2InstanceBootSpecific, EC2Tags
+from common_library.basic_types import DEFAULT_FACTORY
 from common_library.logging.logging_utils_filtering import LoggerName, MessageSubstring
 from fastapi import FastAPI
 from models_library.basic_types import LogLevel, PortInt, VersionTag
 from models_library.clusters import ClusterAuthentication
-from models_library.docker import DockerLabelKey
+from models_library.docker import DockerGenericTag, DockerLabelKey
 from pydantic import (
     AliasChoices,
     AnyUrl,
@@ -63,6 +64,14 @@ class EC2InstancesSettings(BaseCustomSettings):
         ),
     ]
 
+    EC2_INSTANCES_COLD_START_DOCKER_IMAGES_PRE_PULLING: Annotated[
+        list[DockerGenericTag],
+        Field(
+            description="List of docker images to pre-pull on cold started new EC2 instances",
+            default_factory=list,
+        ),
+    ] = DEFAULT_FACTORY
+
     EC2_INSTANCES_KEY_NAME: Annotated[
         str,
         Field(