Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

♻️ reroute get project inactivity via dynamic-scheduler #6949

Open
wants to merge 14 commits into
base: master
Choose a base branch
from
Original file line number Diff line number Diff line change
Expand Up @@ -79,3 +79,5 @@ class DynamicServiceCreate(ServiceDetails):

class GetProjectInactivityResponse(BaseModel):
is_inactive: bool

model_config = ConfigDict(json_schema_extra={"example": {"is_inactive": "false"}})
GitHK marked this conversation as resolved.
Show resolved Hide resolved
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler import DYNAMIC_SCHEDULER_RPC_NAMESPACE
Expand Down Expand Up @@ -99,6 +100,24 @@ async def stop_dynamic_service(
assert result is None # nosec


@log_decorator(_logger, level=logging.DEBUG)
async def get_project_inactivity(
rabbitmq_rpc_client: RabbitMQRPCClient,
*,
project_id: ProjectID,
max_inactivity_seconds: NonNegativeInt,
) -> GetProjectInactivityResponse:
result = await rabbitmq_rpc_client.request(
DYNAMIC_SCHEDULER_RPC_NAMESPACE,
_RPC_METHOD_NAME_ADAPTER.validate_python("get_project_inactivity"),
project_id=project_id,
max_inactivity_seconds=max_inactivity_seconds,
timeout_s=_RPC_DEFAULT_TIMEOUT_S,
)
assert isinstance(result, GetProjectInactivityResponse) # nosec
return result


@log_decorator(_logger, level=logging.DEBUG)
async def restart_user_services(
rabbitmq_rpc_client: RabbitMQRPCClient,
Expand Down
5 changes: 4 additions & 1 deletion services/director-v2/openapi.json
Original file line number Diff line number Diff line change
Expand Up @@ -2058,7 +2058,10 @@
"required": [
"is_inactive"
],
"title": "GetProjectInactivityResponse"
"title": "GetProjectInactivityResponse",
"example": {
"is_inactive": "false"
}
},
"HTTPValidationError": {
"properties": {
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from fastapi import FastAPI
from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
Expand All @@ -12,6 +13,7 @@
from models_library.projects_nodes_io import NodeID
from models_library.services_types import ServicePortKey
from models_library.users import UserID
from pydantic import NonNegativeInt
from servicelib.rabbitmq import RPCRouter
from servicelib.rabbitmq.rpc_interfaces.dynamic_scheduler.errors import (
ServiceWaitingForManualInterventionError,
Expand Down Expand Up @@ -62,6 +64,15 @@ async def stop_dynamic_service(
)


@router.expose()
async def get_project_inactivity(
app: FastAPI, *, project_id: ProjectID, max_inactivity_seconds: NonNegativeInt
) -> GetProjectInactivityResponse:
return await scheduler_interface.get_project_inactivity(
app, project_id=project_id, max_inactivity_seconds=max_inactivity_seconds
)


@router.expose()
async def restart_user_services(app: FastAPI, *, node_id: NodeID) -> None:
await scheduler_interface.restart_user_services(app, node_id=node_id)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from fastapi import FastAPI, status
from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
Expand All @@ -14,7 +15,7 @@
from models_library.projects_nodes_io import NodeID
from models_library.services_types import ServicePortKey
from models_library.users import UserID
from pydantic import TypeAdapter
from pydantic import NonNegativeInt, TypeAdapter
from servicelib.fastapi.app_state import SingletonInAppStateMixin
from servicelib.fastapi.http_client import AttachLifespanMixin, HasClientSetupInterface
from servicelib.fastapi.http_client_thin import UnexpectedStatusError
Expand Down Expand Up @@ -125,6 +126,16 @@ async def list_tracked_dynamic_services(
)
return TypeAdapter(list[DynamicServiceGet]).validate_python(response.json())

async def get_project_inactivity(
self, *, project_id: ProjectID, max_inactivity_seconds: NonNegativeInt
) -> GetProjectInactivityResponse:
response = await self.thin_client.get_projects_inactivity(
project_id=project_id, max_inactivity_seconds=max_inactivity_seconds
)
return TypeAdapter(GetProjectInactivityResponse).validate_python(
response.json()
)

async def restart_user_services(self, *, node_id: NodeID) -> None:
await self.thin_client.post_restart(node_id=node_id)

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
from models_library.services_resources import ServiceResourcesDictHelpers
from models_library.services_types import ServicePortKey
from models_library.users import UserID
from pydantic import NonNegativeInt
from servicelib.common_headers import (
X_DYNAMIC_SIDECAR_REQUEST_DNS,
X_DYNAMIC_SIDECAR_REQUEST_SCHEME,
Expand Down Expand Up @@ -143,6 +144,15 @@ async def get_dynamic_services(
)

@retry_on_errors()
@expect_status(status.HTTP_200_OK)
async def get_projects_inactivity(
self, *, project_id: ProjectID, max_inactivity_seconds: NonNegativeInt
) -> Response:
return await self.client.get(
f"/dynamic_services/projects/{project_id}/inactivity",
params={"max_inactivity_seconds": max_inactivity_seconds},
)

@expect_status(status.HTTP_204_NO_CONTENT)
async def post_restart(self, *, node_id: NodeID) -> Response:
return await self.client.post(f"/dynamic_services/{node_id}:restart")
Expand Down
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from fastapi import FastAPI
from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
Expand All @@ -12,6 +13,7 @@
from models_library.projects_nodes_io import NodeID
from models_library.services_types import ServicePortKey
from models_library.users import UserID
from pydantic import NonNegativeInt

from ..core.settings import ApplicationSettings
from .director_v2 import DirectorV2Client
Expand Down Expand Up @@ -79,6 +81,22 @@ async def stop_dynamic_service(
await set_request_as_stopped(app, dynamic_service_stop)


async def get_project_inactivity(
app: FastAPI, *, project_id: ProjectID, max_inactivity_seconds: NonNegativeInt
) -> GetProjectInactivityResponse:
settings: ApplicationSettings = app.state.settings
if settings.DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER:
raise NotImplementedError

director_v2_client = DirectorV2Client.get_from_app_state(app)
response: GetProjectInactivityResponse = (
await director_v2_client.get_project_inactivity(
project_id=project_id, max_inactivity_seconds=max_inactivity_seconds
)
)
return response


async def restart_user_services(app: FastAPI, *, node_id: NodeID) -> None:
settings: ApplicationSettings = app.state.settings
if settings.DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from fastapi.encoders import jsonable_encoder
from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
Expand Down Expand Up @@ -493,6 +494,40 @@ async def test_stop_dynamic_service_serializes_generic_errors(
)


@pytest.fixture
def inactivity_response() -> GetProjectInactivityResponse:
return TypeAdapter(GetProjectInactivityResponse).validate_python(
GetProjectInactivityResponse.model_json_schema()["example"]
)


@pytest.fixture
def mock_director_v2_get_project_inactivity(
project_id: ProjectID, inactivity_response: GetProjectInactivityResponse
) -> Iterator[None]:
with respx.mock(
base_url="http://director-v2:8000/v2",
assert_all_called=False,
assert_all_mocked=True, # IMPORTANT: KEEP always True!
) as mock:
mock.get(f"/dynamic_services/projects/{project_id}/inactivity").respond(
status.HTTP_200_OK, text=inactivity_response.model_dump_json()
)
yield None


async def test_get_project_inactivity(
mock_director_v2_get_project_inactivity: None,
rpc_client: RabbitMQRPCClient,
project_id: ProjectID,
inactivity_response: GetProjectInactivityResponse,
):
result = await services.get_project_inactivity(
rpc_client, project_id=project_id, max_inactivity_seconds=5
)
assert result == inactivity_response


@pytest.fixture
def mock_director_v2_restart_user_services(node_id: NodeID) -> Iterator[None]:
with respx.mock(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9952,6 +9952,8 @@ components:
required:
- is_inactive
title: GetProjectInactivityResponse
example:
is_inactive: 'false'
GetWalletAutoRecharge:
properties:
enabled:
Expand Down

This file was deleted.

Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
is_pipeline_running,
stop_pipeline,
)
from ._core_dynamic_services import get_project_inactivity
from ._core_utils import is_healthy
from .exceptions import DirectorServiceError

Expand All @@ -28,7 +27,6 @@
"DirectorServiceError",
"get_batch_tasks_outputs",
"get_computation_task",
"get_project_inactivity",
"get_project_run_policy",
"is_healthy",
"is_pipeline_running",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from aiohttp import web
from models_library.api_schemas_directorv2.dynamic_services import (
DynamicServiceGet,
GetProjectInactivityResponse,
RetrieveDataOutEnveloped,
)
from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
Expand All @@ -23,6 +24,7 @@
from models_library.rabbitmq_messages import ProgressRabbitMessageProject, ProgressType
from models_library.services import ServicePortKey
from models_library.users import UserID
from pydantic import NonNegativeInt
from pydantic.types import PositiveInt
from servicelib.progress_bar import ProgressBarData
from servicelib.rabbitmq import RabbitMQClient, RPCServerError
Expand Down Expand Up @@ -154,6 +156,19 @@ async def stop_dynamic_services_in_project(
await logged_gather(*services_to_stop)


async def get_project_inactivity(
app: web.Application,
*,
project_id: ProjectID,
max_inactivity_seconds: NonNegativeInt,
) -> GetProjectInactivityResponse:
return await services.get_project_inactivity(
get_rabbitmq_rpc_client(app),
project_id=project_id,
max_inactivity_seconds=max_inactivity_seconds,
)


async def restart_user_services(app: web.Application, *, node_id: NodeID) -> None:
"""Restarts the user service(s) started by the the node_uuid's sidecar

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1895,13 +1895,12 @@ async def get_project_inactivity(
app: web.Application, project_id: ProjectID
) -> GetProjectInactivityResponse:
project_settings: ProjectsSettings = get_plugin_settings(app)
project_inactivity = await director_v2_api.get_project_inactivity(
return await dynamic_scheduler_api.get_project_inactivity(
app,
project_id,
project_id=project_id,
# NOTE: project is considered inactive if all services exposing an /inactivity
# endpoint were inactive since at least PROJECTS_INACTIVITY_INTERVAL
max_inactivity_seconds=int(
project_settings.PROJECTS_INACTIVITY_INTERVAL.total_seconds()
),
)
return GetProjectInactivityResponse.model_validate(project_inactivity)
Loading