Comp backend/clusters creation functionalities (#4602)

Author: sanderegg

.ruff.toml

@@ -51,7 +51,7 @@ target-version = "py310"
 
 
 [per-file-ignores]
-"**/tests/**" = [
+"{**/{tests, pytest_simcore}/**}" = [
   "T201",    # print found
   "ARG001",  # unused function argument
   "PT019",   # user pytest.mark.usefixture

.vscode/launch.template.json

@@ -46,6 +46,19 @@
       "name": "Python: Remote Attach api-server",
       "type": "python",
       "request": "attach",
+      "port": 3015,
+      "host": "127.0.0.1",
+      "pathMappings": [
+        {
+          "localRoot": "${workspaceFolder}",
+          "remoteRoot": "/devel"
+        }
+      ]
+    },
+    {
+      "name": "Python: Remote Attach clusters-keeper",
+      "type": "python",
+      "request": "attach",
       "port": 3006,
       "host": "127.0.0.1",
       "pathMappings": [

packages/pytest-simcore/src/pytest_simcore/dask_gateway.py

@@ -0,0 +1,104 @@
+# pylint: disable=unused-argument
+# pylint: disable=redefined-outer-name
+
+from collections.abc import Callable
+from typing import AsyncIterator, NamedTuple
+
+import pytest
+import traitlets.config
+from dask_gateway import Gateway, GatewayCluster, auth
+from dask_gateway_server.app import DaskGateway
+from dask_gateway_server.backends.local import UnsafeLocalBackend
+from distributed import Client
+
+
+@pytest.fixture
+def local_dask_gateway_server_config(
+    unused_tcp_port_factory: Callable,
+) -> traitlets.config.Config:
+    c = traitlets.config.Config()
+    assert isinstance(c.DaskGateway, traitlets.config.Config)
+    assert isinstance(c.ClusterConfig, traitlets.config.Config)
+    assert isinstance(c.Proxy, traitlets.config.Config)
+    assert isinstance(c.SimpleAuthenticator, traitlets.config.Config)
+    c.DaskGateway.backend_class = UnsafeLocalBackend
+    c.DaskGateway.address = f"127.0.0.1:{unused_tcp_port_factory()}"
+    c.Proxy.address = f"127.0.0.1:{unused_tcp_port_factory()}"
+    c.DaskGateway.authenticator_class = "dask_gateway_server.auth.SimpleAuthenticator"
+    c.SimpleAuthenticator.password = "qweqwe"  # noqa: S105
+    c.ClusterConfig.worker_cmd = [
+        "dask-worker",
+        "--resources",
+        f"CPU=12,GPU=1,RAM={16e9}",
+    ]
+    # NOTE: This must be set such that the local unsafe backend creates a worker with enough cores/memory
+    c.ClusterConfig.worker_cores = 12
+    c.ClusterConfig.worker_memory = "16G"
+    c.ClusterConfig.cluster_max_workers = 3
+
+    c.DaskGateway.log_level = "DEBUG"
+    return c
+
+
+class DaskGatewayServer(NamedTuple):
+    address: str
+    proxy_address: str
+    password: str
+    server: DaskGateway
+
+
+@pytest.fixture
+async def local_dask_gateway_server(
+    local_dask_gateway_server_config: traitlets.config.Config,
+) -> AsyncIterator[DaskGatewayServer]:
+    print("--> creating local dask gateway server")
+    dask_gateway_server = DaskGateway(config=local_dask_gateway_server_config)
+    dask_gateway_server.initialize([])  # that is a shitty one!
+    print("--> local dask gateway server initialized")
+    await dask_gateway_server.setup()
+    await dask_gateway_server.backend.proxy._proxy_contacted  # pylint: disable=protected-access
+
+    print("--> local dask gateway server setup completed")
+    yield DaskGatewayServer(
+        f"http://{dask_gateway_server.backend.proxy.address}",
+        f"gateway://{dask_gateway_server.backend.proxy.tcp_address}",
+        local_dask_gateway_server_config.SimpleAuthenticator.password,  # type: ignore
+        dask_gateway_server,
+    )
+    print("--> local dask gateway server switching off...")
+    await dask_gateway_server.cleanup()
+    print("...done")
+
+
+@pytest.fixture
+async def dask_gateway(
+    local_dask_gateway_server: DaskGatewayServer,
+) -> Gateway:
+    async with Gateway(
+        local_dask_gateway_server.address,
+        local_dask_gateway_server.proxy_address,
+        asynchronous=True,
+        auth=auth.BasicAuth("pytest_user", local_dask_gateway_server.password),
+    ) as gateway:
+        print(f"--> {gateway=} created")
+        cluster_options = await gateway.cluster_options()
+        gateway_versions = await gateway.get_versions()
+        clusters_list = await gateway.list_clusters()
+        print(f"--> {gateway_versions=}, {cluster_options=}, {clusters_list=}")
+        for option in cluster_options.items():
+            print(f"--> {option=}")
+        return gateway
+
+
+@pytest.fixture
+async def dask_gateway_cluster(dask_gateway: Gateway) -> AsyncIterator[GatewayCluster]:
+    async with dask_gateway.new_cluster() as cluster:
+        yield cluster
+
+
+@pytest.fixture
+async def dask_gateway_cluster_client(
+    dask_gateway_cluster: GatewayCluster,
+) -> AsyncIterator[Client]:
+    async with dask_gateway_cluster.get_client() as client:
+        yield client

services/clusters-keeper/.env-devel

@@ -0,0 +1,21 @@
+CLUSTERS_KEEPER_DEBUG=true
+CLUSTERS_KEEPER_LOGLEVEL=INFO
+CLUSTERS_KEEPER_MAX_MISSED_HEARTBEATS_BEFORE_CLUSTER_TERMINATION=60
+CLUSTERS_KEEPER_TASK_INTERVAL=30
+EC2_ACCESS_KEY_ID=XXXXXXXXXX
+EC2_INSTANCES_ALLOWED_TYPES="[\"t2.micro\"]"
+EC2_INSTANCES_AMI_ID=XXXXXXXXXX
+EC2_INSTANCES_KEY_NAME=XXXXXXXXXX
+EC2_INSTANCES_SECURITY_GROUP_IDS=XXXXXXXXXX
+EC2_INSTANCES_SUBNET_ID=XXXXXXXXXX
+EC2_SECRET_ACCESS_KEY=XXXXXXXXXX
+LOG_FORMAT_LOCAL_DEV_ENABLED=True
+RABBIT_HOST=rabbit
+RABBIT_PASSWORD=test
+RABBIT_PORT=5672
+RABBIT_SECURE=false
+RABBIT_USER=test
+REDIS_HOST=redis
+REDIS_PORT=6379
+SC_BOOT_MODE=debug-ptvsd
+SC_BUILD_TARGET=development

services/clusters-keeper/Makefile

@@ -3,3 +3,17 @@
 #
 include ../../scripts/common.Makefile
 include ../../scripts/common-service.Makefile
+
+.env: .env-devel ## creates .env file from defaults in .env-devel
+	$(if $(wildcard $@), \
+	@echo "WARNING #####  $< is newer than $@ ####"; diff -uN $@ $<; false;,\
+	@echo "WARNING ##### $@ does not exist, cloning $< as $@ ############"; cp $< $@)
+
+
+.PHONY: test-local
+up-devel: .env ## starts local test application (running bare metal against AWS)
+	# setting up dependencies
+	@docker compose up
+
+down: .env ## stops local test app dependencies (running bare metal against AWS)
+	-@docker compose down

services/clusters-keeper/docker-compose.yml

@@ -0,0 +1,51 @@
+version: "3.8"
+services:
+  rabbit:
+    image: itisfoundation/rabbitmq:3.11.2-management
+    init: true
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+      - "15692"
+    environment:
+      - RABBITMQ_DEFAULT_USER=${RABBIT_USER}
+      - RABBITMQ_DEFAULT_PASS=${RABBIT_PASSWORD}
+    healthcheck:
+      # see https://www.rabbitmq.com/monitoring.html#individual-checks for info about health-checks available in rabbitmq
+      test: rabbitmq-diagnostics -q status
+      interval: 5s
+      timeout: 30s
+      retries: 5
+      start_period: 5s
+
+  redis:
+    image: "redis:6.2.6@sha256:4bed291aa5efb9f0d77b76ff7d4ab71eee410962965d052552db1fb80576431d"
+    init: true
+    ports:
+      - "6379:6379"
+    healthcheck:
+      test: [ "CMD", "redis-cli", "ping" ]
+      interval: 5s
+      timeout: 30s
+      retries: 50
+
+  redis-commander:
+    image: rediscommander/redis-commander:latest
+    init: true
+    ports:
+      - "18081:8081"
+    environment:
+      - REDIS_HOSTS=resources:${REDIS_HOST}:${REDIS_PORT}:0,locks:${REDIS_HOST}:${REDIS_PORT}:1,validation_codes:${REDIS_HOST}:${REDIS_PORT}:2,scheduled_maintenance:${REDIS_HOST}:${REDIS_PORT}:3,user_notifications:${REDIS_HOST}:${REDIS_PORT}:4,announcements:${REDIS_HOST}:${REDIS_PORT}:5
+      # If you add/remove a db, do not forget to update the --databases entry in the docker-compose.yml
+
+  clusters-keeper:
+    image: local/clusters-keeper:development
+    init: true
+    ports:
+      - "8010:8000"
+      - "3015:3000"
+    env_file:
+      - .env
+    volumes:
+      - ./:/devel/services/clusters-keeper
+      - ../../packages:/devel/packages

services/clusters-keeper/docker/entrypoint.sh

@@ -65,7 +65,7 @@ fi
 
 if [ "${SC_BOOT_MODE}" = "debug-ptvsd" ]; then
   # NOTE: production does NOT pre-installs ptvsd
-  pip install --no-cache-dir ptvsd
+  pip install --no-cache-dir debugpy
 fi
 
 # Appends docker group if socket is mounted

services/clusters-keeper/requirements/_base.in

@@ -4,6 +4,7 @@
 # NOTE: ALL version constraints MUST be commented
 --constraint ../../../requirements/constraints.txt
 --constraint ./constraints.txt
+--constraint ../../../services/dask-sidecar/requirements/_dask-distributed.txt
 
 # intra-repo required dependencies
 --requirement ../../../packages/models-library/requirements/_base.in
@@ -13,7 +14,10 @@
 --requirement ../../../packages/service-library/requirements/_fastapi.in
 
 
+
 aioboto3
+dask[distributed]
+dask-gateway
 fastapi
 packaging
 types-aiobotocore[ec2]