✨Computational clusters: connect autoscaling to RabbitMQ (#7485)

sanderegg · web-flow · commit f4076447c762 · 2025-04-23T07:19:55.000Z
diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md
@@ -21,7 +21,7 @@ This document provides guidelines and best practices for using GitHub Copilot in
 - ensure we use `sqlalchemy` >2 compatible code.
 - ensure we use `pydantic` >2 compatible code.
 - ensure we use `fastapi` >0.100 compatible code
-
+- use f-string formatting
 
 ## Node.js-Specific Instructions
 
diff --git a/packages/models-library/src/models_library/clusters.py b/packages/models-library/src/models_library/clusters.py
@@ -3,6 +3,7 @@
 from typing import Literal, TypeAlias
 
 from pydantic import AnyUrl, BaseModel, ConfigDict, Field, HttpUrl, field_validator
+from pydantic.config import JsonDict
 from pydantic.types import NonNegativeInt
 
 from .groups import GroupID
@@ -36,18 +37,22 @@ class TLSAuthentication(_AuthenticationBase):
     tls_client_cert: Path
     tls_client_key: Path
 
-    model_config = ConfigDict(
-        json_schema_extra={
-            "examples": [
-                {
-                    "type": "tls",
-                    "tls_ca_file": "/path/to/ca_file",
-                    "tls_client_cert": "/path/to/cert_file",
-                    "tls_client_key": "/path/to/key_file",
-                },
-            ]
-        }
-    )
+    @staticmethod
+    def _update_json_schema_extra(schema: JsonDict) -> None:
+        schema.update(
+            {
+                "examples": [
+                    {
+                        "type": "tls",
+                        "tls_ca_file": "/path/to/ca_file",
+                        "tls_client_cert": "/path/to/cert_file",
+                        "tls_client_key": "/path/to/key_file",
+                    },
+                ]
+            }
+        )
+
+    model_config = ConfigDict(json_schema_extra=_update_json_schema_extra)
 
 
 ClusterAuthentication: TypeAlias = NoAuthentication | TLSAuthentication
@@ -71,36 +76,41 @@ class BaseCluster(BaseModel):
         create_enums_pre_validator(ClusterTypeInModel)
     )
 
-    model_config = ConfigDict(
-        use_enum_values=True,
-        json_schema_extra={
-            "examples": [
-                {
-                    "name": "My awesome cluster",
-                    "type": ClusterTypeInModel.ON_PREMISE,
-                    "owner": 12,
-                    "endpoint": "https://registry.osparc-development.fake.dev",
-                    "authentication": {
-                        "type": "tls",
-                        "tls_ca_file": "/path/to/ca_file",
-                        "tls_client_cert": "/path/to/cert_file",
-                        "tls_client_key": "/path/to/key_file",
+    @staticmethod
+    def _update_json_schema_extra(schema: JsonDict) -> None:
+        schema.update(
+            {
+                "examples": [
+                    {
+                        "name": "My awesome cluster",
+                        "type": ClusterTypeInModel.ON_PREMISE,
+                        "owner": 12,
+                        "endpoint": "https://registry.osparc-development.fake.dev",
+                        "authentication": {
+                            "type": "tls",
+                            "tls_ca_file": "/path/to/ca_file",
+                            "tls_client_cert": "/path/to/cert_file",
+                            "tls_client_key": "/path/to/key_file",
+                        },
                     },
-                },
-                {
-                    "name": "My AWS cluster",
-                    "type": ClusterTypeInModel.AWS,
-                    "owner": 154,
-                    "endpoint": "https://registry.osparc-development.fake.dev",
-                    "authentication": {
-                        "type": "tls",
-                        "tls_ca_file": "/path/to/ca_file",
-                        "tls_client_cert": "/path/to/cert_file",
-                        "tls_client_key": "/path/to/key_file",
+                    {
+                        "name": "My AWS cluster",
+                        "type": ClusterTypeInModel.AWS,
+                        "owner": 154,
+                        "endpoint": "https://registry.osparc-development.fake.dev",
+                        "authentication": {
+                            "type": "tls",
+                            "tls_ca_file": "/path/to/ca_file",
+                            "tls_client_cert": "/path/to/cert_file",
+                            "tls_client_key": "/path/to/key_file",
+                        },
                     },
-                },
-            ]
-        },
+                ]
+            }
+        )
+
+    model_config = ConfigDict(
+        use_enum_values=True, json_schema_extra=_update_json_schema_extra
     )
 
 
diff --git a/packages/pytest-simcore/src/pytest_simcore/helpers/monkeypatch_envs.py b/packages/pytest-simcore/src/pytest_simcore/helpers/monkeypatch_envs.py
@@ -30,6 +30,9 @@ def setenvs_from_dict(
         if isinstance(value, bool):
             v = "true" if value else "false"
 
+        if isinstance(value, int | float):
+            v = f"{value}"
+
         assert isinstance(v, str), (
             "caller MUST explicitly stringify values since some cannot be done automatically"
             f"e.g. json-like values. Check {key=},{value=}"
diff --git a/packages/service-library/tests/fastapi/test_rabbitmq_lifespan.py b/packages/service-library/tests/fastapi/test_rabbitmq_lifespan.py
@@ -39,7 +39,7 @@ def mock_rabbitmq_connection(mocker: MockerFixture) -> MockType:
 def mock_rabbitmq_rpc_client_class(mocker: MockerFixture) -> MockType:
     mock_rpc_client_instance = mocker.AsyncMock()
     mocker.patch.object(
-        servicelib.rabbitmq._client_rpc.RabbitMQRPCClient,
+        servicelib.rabbitmq._client_rpc.RabbitMQRPCClient,  # noqa: SLF001
         "create",
         return_value=mock_rpc_client_instance,
     )
@@ -87,7 +87,6 @@ async def my_app_rpc_server(app: FastAPI, state: State) -> AsyncIterator[State]:
 
     # setup rpc-client using rabbitmq_rpc_client_context
     async def my_app_rpc_client(app: FastAPI, state: State) -> AsyncIterator[State]:
-
         assert "RABBIT_CONNECTIVITY_LIFESPAN_NAME" in state
 
         async with rabbitmq_rpc_client_context(
@@ -122,7 +121,6 @@ async def test_lifespan_rabbitmq_in_an_app(
         startup_timeout=None if is_pdb_enabled else 10,
         shutdown_timeout=None if is_pdb_enabled else 10,
     ):
-
         # Verify that RabbitMQ responsiveness was checked
         mock_rabbitmq_connection.assert_called_once_with(
             app.state.settings.RABBITMQ.dsn
diff --git a/packages/settings-library/src/settings_library/rabbit.py b/packages/settings-library/src/settings_library/rabbit.py
@@ -1,5 +1,7 @@
 from functools import cached_property
+from typing import ClassVar
 
+from pydantic.config import JsonDict
 from pydantic.networks import AnyUrl
 from pydantic.types import SecretStr
 from pydantic_settings import SettingsConfigDict
@@ -9,7 +11,7 @@
 
 
 class RabbitDsn(AnyUrl):
-    allowed_schemes = {"amqp", "amqps"}
+    allowed_schemes: ClassVar[set[str]] = {"amqp", "amqps"}
 
 
 class RabbitSettings(BaseCustomSettings):
@@ -35,16 +37,31 @@ def dsn(self) -> str:
         )
         return rabbit_dsn
 
+    @staticmethod
+    def _update_json_schema_extra(schema: JsonDict) -> None:
+        schema.update(
+            {
+                "examples": [
+                    {
+                        "RABBIT_HOST": "rabbitmq.example.com",
+                        "RABBIT_USER": "guest",
+                        "RABBIT_PASSWORD": "guest-password",
+                        "RABBIT_SECURE": False,
+                        "RABBIT_PORT": 5672,
+                    },
+                    {
+                        "RABBIT_HOST": "secure.rabbitmq.example.com",
+                        "RABBIT_USER": "guest",
+                        "RABBIT_PASSWORD": "guest-password",
+                        "RABBIT_SECURE": True,
+                        "RABBIT_PORT": 15672,
+                    },
+                ]
+            }
+        )
+
     model_config = SettingsConfigDict(
-        json_schema_extra={
-            "examples": [
-                # minimal required
-                {
-                    "RABBIT_SECURE": "1",
-                    "RABBIT_HOST": "localhost",
-                    "RABBIT_USER": "user",
-                    "RABBIT_PASSWORD": "foobar",  # NOSONAR
-                }
-            ],
-        }
+        extra="ignore",
+        populate_by_name=True,
+        json_schema_extra=_update_json_schema_extra,
     )
diff --git a/services/autoscaling/tests/unit/test_modules_dask.py b/services/autoscaling/tests/unit/test_modules_dask.py
@@ -43,7 +43,7 @@
 _authentication_types = [
     NoAuthentication(),
     TLSAuthentication.model_construct(
-        **TLSAuthentication.model_config["json_schema_extra"]["examples"][0]
+        **TLSAuthentication.model_json_schema()["examples"][0]
     ),
 ]
 
@@ -267,7 +267,9 @@ def _add_fct(x: int, y: int) -> int:
         )
         assert isinstance(exc, RuntimeError)
     else:
-        result = await future_queued_task.result(timeout=_DASK_SCHEDULER_REACTION_TIME_S)  # type: ignore
+        result = await future_queued_task.result(
+            timeout=_DASK_SCHEDULER_REACTION_TIME_S
+        )  # type: ignore
         assert result == 7
 
     await _wait_for_dask_scheduler_to_change_state()
diff --git a/services/clusters-keeper/src/simcore_service_clusters_keeper/core/settings.py b/services/clusters-keeper/src/simcore_service_clusters_keeper/core/settings.py
@@ -252,6 +252,14 @@ class PrimaryEC2InstancesSettings(BaseCustomSettings):
         ),
     ] = "172.20.0.0/14"  # nosec
 
+    PRIMARY_EC2_INSTANCES_RABBIT: Annotated[
+        RabbitSettings | None,
+        Field(
+            description="defines the Rabbit settings for the primary instance (may be disabled)",
+            json_schema_extra={"auto_default_from_env": True},
+        ),
+    ]
+
     @field_validator("PRIMARY_EC2_INSTANCES_ALLOWED_TYPES")
     @classmethod
     def _check_valid_instance_names(
diff --git a/services/clusters-keeper/src/simcore_service_clusters_keeper/data/docker-compose.yml b/services/clusters-keeper/src/simcore_service_clusters_keeper/data/docker-compose.yml
@@ -91,6 +91,7 @@ services:
       AUTOSCALING_EC2_SECRET_ACCESS_KEY: ${CLUSTERS_KEEPER_EC2_SECRET_ACCESS_KEY}
       AUTOSCALING_NODES_MONITORING: null
       AUTOSCALING_POLL_INTERVAL: 00:00:10
+      AUTOSCALING_RABBITMQ: ${AUTOSCALING_RABBITMQ}
       DASK_MONITORING_URL: tls://dask-scheduler:8786
       DASK_SCHEDULER_AUTH: '{"type":"tls","tls_ca_file":"${DASK_TLS_CA_FILE}","tls_client_cert":"${DASK_TLS_CERT}","tls_client_key":"${DASK_TLS_KEY}"}'
       EC2_INSTANCES_ALLOWED_TYPES: ${WORKERS_EC2_INSTANCES_ALLOWED_TYPES}
@@ -188,6 +189,7 @@ services:
 networks:
   cluster:
 
+
 configs:
   prometheus-config:
     file: ./prometheus.yml
@@ -200,6 +202,7 @@ volumes:
   redis-data:
   prometheus-data:
 
+
 secrets:
   dask_tls_ca:
     file: ${DASK_TLS_CA_FILE}
diff --git a/services/clusters-keeper/src/simcore_service_clusters_keeper/utils/clusters.py b/services/clusters-keeper/src/simcore_service_clusters_keeper/utils/clusters.py
@@ -9,6 +9,8 @@
 import yaml
 from aws_library.ec2 import EC2InstanceBootSpecific, EC2InstanceData, EC2Tags
 from aws_library.ec2._models import CommandStr
+from common_library.json_serialization import json_dumps
+from common_library.serialization import model_dump_with_secrets
 from fastapi.encoders import jsonable_encoder
 from models_library.api_schemas_clusters_keeper.clusters import (
     ClusterState,
@@ -81,6 +83,8 @@ def _convert_to_env_list(entries: list[Any]) -> str:
     def _convert_to_env_dict(entries: dict[str, Any]) -> str:
         return f"'{json.dumps(jsonable_encoder(entries))}'"
 
+    assert app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES  # nosec
+
     return [
         f"CLUSTERS_KEEPER_EC2_ACCESS_KEY_ID={app_settings.CLUSTERS_KEEPER_EC2_ACCESS.EC2_ACCESS_KEY_ID}",
         f"CLUSTERS_KEEPER_EC2_ENDPOINT={app_settings.CLUSTERS_KEEPER_EC2_ACCESS.EC2_ENDPOINT or 'null'}",
@@ -102,6 +106,7 @@ def _convert_to_env_dict(entries: dict[str, Any]) -> str:
         f"WORKERS_EC2_INSTANCES_SUBNET_ID={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_SUBNET_ID}",
         f"WORKERS_EC2_INSTANCES_TIME_BEFORE_DRAINING={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_TIME_BEFORE_DRAINING}",
         f"WORKERS_EC2_INSTANCES_TIME_BEFORE_TERMINATION={app_settings.CLUSTERS_KEEPER_WORKERS_EC2_INSTANCES.WORKERS_EC2_INSTANCES_TIME_BEFORE_TERMINATION}",
+        f"AUTOSCALING_RABBITMQ={json_dumps(model_dump_with_secrets(app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES.PRIMARY_EC2_INSTANCES_RABBIT, show_secrets=True)) if app_settings.CLUSTERS_KEEPER_PRIMARY_EC2_INSTANCES.PRIMARY_EC2_INSTANCES_RABBIT else 'null'}",
     ]
 
 
diff --git a/services/clusters-keeper/tests/unit/conftest.py b/services/clusters-keeper/tests/unit/conftest.py
@@ -224,6 +224,7 @@ def disable_clusters_management_background_task(
 @pytest.fixture
 def disabled_rabbitmq(app_environment: EnvVarsDict, monkeypatch: pytest.MonkeyPatch):
     monkeypatch.setenv("CLUSTERS_KEEPER_RABBITMQ", "null")
+    monkeypatch.delenv("RABBIT_HOST", raising=False)
 
 
 @pytest.fixture
@@ -331,7 +332,7 @@ async def _do(num: int) -> list[str]:
                     "Tags": [
                         {
                             "Key": "Name",
-                            "Value": f"{get_cluster_name(app_settings,user_id=user_id,wallet_id=wallet_id,is_manager=False)}_blahblah",
+                            "Value": f"{get_cluster_name(app_settings, user_id=user_id, wallet_id=wallet_id, is_manager=False)}_blahblah",
                         }
                     ],
                 }
diff --git a/services/clusters-keeper/tests/unit/test_modules_dask.py b/services/clusters-keeper/tests/unit/test_modules_dask.py
@@ -25,7 +25,7 @@
 _authentication_types = [
     NoAuthentication(),
     TLSAuthentication.model_construct(
-        **TLSAuthentication.model_config["json_schema_extra"]["examples"][0]
+        **TLSAuthentication.model_json_schema()["examples"][0]
     ),
 ]
 
diff --git a/services/clusters-keeper/tests/unit/test_utils_clusters.py b/services/clusters-keeper/tests/unit/test_utils_clusters.py
diff --git a/services/docker-compose.yml b/services/docker-compose.yml

Original file line number	Diff line number	Diff line change
`@@ -224,6 +224,7 @@ def disable_clusters_management_background_task(`
`224`	`224`	`@pytest.fixture`
`225`	`225`	`def disabled_rabbitmq(app_environment: EnvVarsDict, monkeypatch: pytest.MonkeyPatch):`
`226`	`226`	`monkeypatch.setenv("CLUSTERS_KEEPER_RABBITMQ", "null")`
	`227`	`+ monkeypatch.delenv("RABBIT_HOST", raising=False)`
`227`	`228`
`228`	`229`
`229`	`230`	`@pytest.fixture`
`@@ -331,7 +332,7 @@ async def _do(num: int) -> list[str]:`
`331`	`332`	`"Tags": [`
`332`	`333`	`{`
`333`	`334`	`"Key": "Name",`
`334`		`- "Value": f"{get_cluster_name(app_settings,user_id=user_id,wallet_id=wallet_id,is_manager=False)}_blahblah",`
	`335`	`+ "Value": f"{get_cluster_name(app_settings, user_id=user_id, wallet_id=wallet_id, is_manager=False)}_blahblah",`
`335`	`336`	`}`
`336`	`337`	`],`
`337`	`338`	`}`