clean unused stuff

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/dask_client.py

@@ -12,10 +12,9 @@
 import logging
 import traceback
 from collections.abc import Callable, Iterable
-from copy import deepcopy
 from dataclasses import dataclass
 from http.client import HTTPException
-from typing import Any, Final, cast
+from typing import Final, cast
 
 import distributed
 from aiohttp import ClientResponseError
@@ -49,17 +48,16 @@
     create_ec2_resource_constraint_key,
 )
 from fastapi import FastAPI
-from models_library.api_schemas_directorv2.clusters import ClusterDetails, Scheduler
 from models_library.clusters import ClusterAuthentication, ClusterTypeInModel
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.projects_state import RunningState
 from models_library.resource_tracker import HardwareInfo
 from models_library.services import ServiceRunID
 from models_library.users import UserID
-from pydantic import TypeAdapter, ValidationError
+from pydantic import ValidationError
 from pydantic.networks import AnyUrl
-from servicelib.logging_utils import log_catch, log_context
+from servicelib.logging_utils import log_context
 from settings_library.s3 import S3Settings
 from simcore_sdk.node_ports_common.exceptions import NodeportsException
 from simcore_sdk.node_ports_v2 import FileLinkType
@@ -538,50 +536,3 @@ async def release_task_result(self, job_id: str) -> None:
 
         except KeyError:
             _logger.warning("Unknown task cannot be unpublished: %s", f"{job_id=}")
-
-    async def get_cluster_details(self) -> ClusterDetails:
-        dask_utils.check_scheduler_is_still_the_same(
-            self.backend.scheduler_id, self.backend.client
-        )
-        dask_utils.check_communication_with_scheduler_is_open(self.backend.client)
-        dask_utils.check_scheduler_status(self.backend.client)
-        scheduler_info = self.backend.client.scheduler_info()
-        scheduler_status = self.backend.client.status
-        dashboard_link = self.backend.client.dashboard_link
-
-        def _get_worker_used_resources(
-            dask_scheduler: distributed.Scheduler,
-        ) -> dict[str, dict]:
-            used_resources = {}
-            for worker_name, worker_state in dask_scheduler.workers.items():
-                used_resources[worker_name] = worker_state.used_resources
-            return used_resources
-
-        with log_catch(_logger, reraise=False):
-            # NOTE: this runs directly on the dask-scheduler and may rise exceptions
-            used_resources_per_worker: dict[str, dict[str, Any]] = (
-                await dask_utils.wrap_client_async_routine(
-                    self.backend.client.run_on_scheduler(_get_worker_used_resources)
-                )
-            )
-
-            # let's update the scheduler info, with default to 0s since sometimes
-            # workers are destroyed/created without us knowing right away
-            for worker_name, worker_info in scheduler_info.get("workers", {}).items():
-                used_resources: dict[str, float] = deepcopy(
-                    worker_info.get("resources", {})
-                )
-                # reset default values
-                for res_name in used_resources:
-                    used_resources[res_name] = 0
-                # if the scheduler has info, let's override them
-                used_resources = used_resources_per_worker.get(
-                    worker_name, used_resources
-                )
-                worker_info.update(used_resources=used_resources)
-
-        assert dashboard_link  # nosec
-        return ClusterDetails(
-            scheduler=Scheduler(status=scheduler_status, **scheduler_info),
-            dashboard_link=TypeAdapter(AnyUrl).validate_python(dashboard_link),
-        )

services/director-v2/tests/unit/test_modules_dask_client.py

@@ -1162,108 +1162,3 @@ def fake_remote_fct(
                 (mock.ANY, "my name is progress")
             )
     await _assert_wait_for_cb_call(mocked_user_completed_cb)
-
-
-async def test_get_cluster_details(
-    dask_client: DaskClient,
-    user_id: UserID,
-    project_id: ProjectID,
-    image_params: ImageParams,
-    _mocked_node_ports: None,
-    mocked_user_completed_cb: mock.AsyncMock,
-    mocked_storage_service_api: respx.MockRouter,
-    comp_run_metadata: RunMetadataDict,
-    empty_hardware_info: HardwareInfo,
-    faker: Faker,
-    resource_tracking_run_id: ServiceRunID,
-):
-    cluster_details = await dask_client.get_cluster_details()
-    assert cluster_details
-
-    _DASK_EVENT_NAME = faker.pystr()
-
-    # send a fct that uses resources
-    def fake_sidecar_fct(
-        task_parameters: ContainerTaskParameters,
-        docker_auth: DockerBasicAuth,
-        log_file_url: LogFileUploadURL,
-        s3_settings: S3Settings | None,
-        expected_annotations,
-    ) -> TaskOutputData:
-        # get the task data
-        worker = get_worker()
-        task = worker.state.tasks.get(worker.get_current_task())
-        assert task is not None
-        assert task.annotations == expected_annotations
-        assert task_parameters.command == ["run"]
-        event = distributed.Event(_DASK_EVENT_NAME)
-        event.wait(timeout=25)
-
-        return TaskOutputData.model_validate({"some_output_key": 123})
-
-    # NOTE: We pass another fct so it can run in our localy created dask cluster
-    published_computation_task = await dask_client.send_computation_tasks(
-        user_id=user_id,
-        project_id=project_id,
-        tasks=image_params.fake_tasks,
-        callback=mocked_user_completed_cb,
-        remote_fct=functools.partial(
-            fake_sidecar_fct, expected_annotations=image_params.expected_annotations
-        ),
-        metadata=comp_run_metadata,
-        hardware_info=empty_hardware_info,
-        resource_tracking_run_id=resource_tracking_run_id,
-    )
-    assert published_computation_task
-    assert len(published_computation_task) == 1
-
-    assert published_computation_task[0].node_id in image_params.fake_tasks
-
-    # check status goes to PENDING/STARTED
-    await _assert_wait_for_task_status(
-        published_computation_task[0].job_id,
-        dask_client,
-        expected_status=RunningState.STARTED,
-    )
-
-    # check we have one worker using the resources
-    # one of the workers should now get the job and use the resources
-    worker_with_the_task: AnyUrl | None = None
-    async for attempt in AsyncRetrying(reraise=True, stop=stop_after_delay(10)):
-        with attempt:
-            cluster_details = await dask_client.get_cluster_details()
-            assert cluster_details
-            assert (
-                cluster_details.scheduler.workers
-            ), f"there are no workers in {cluster_details.scheduler=!r}"
-            for worker_url, worker_data in cluster_details.scheduler.workers.items():
-                if all(
-                    worker_data.used_resources.get(res_name) == res_value
-                    for res_name, res_value in image_params.expected_used_resources.items()
-                ):
-                    worker_with_the_task = worker_url
-            assert (
-                worker_with_the_task is not None
-            ), f"there is no worker in {cluster_details.scheduler.workers.keys()=} consuming {image_params.expected_annotations=!r}"
-
-    # using the event we let the remote fct continue
-    event = distributed.Event(_DASK_EVENT_NAME, client=dask_client.backend.client)
-    await event.set()  # type: ignore
-
-    # wait for the task to complete
-    await _assert_wait_for_task_status(
-        published_computation_task[0].job_id,
-        dask_client,
-        expected_status=RunningState.SUCCESS,
-    )
-
-    # check the resources are released
-    cluster_details = await dask_client.get_cluster_details()
-    assert cluster_details
-    assert cluster_details.scheduler.workers
-    assert worker_with_the_task
-    currently_used_resources = cluster_details.scheduler.workers[
-        worker_with_the_task
-    ].used_resources
-
-    assert all(res == 0.0 for res in currently_used_resources.values())