diff --git a/mkdocs/docs/concepts/services.md b/mkdocs/docs/concepts/services.md index 84971e98b..f9f260408 100644 --- a/mkdocs/docs/concepts/services.md +++ b/mkdocs/docs/concepts/services.md @@ -1534,7 +1534,7 @@ The rolling deployment stops when all replicas are updated or when a new deploym Rolling deployment supports changes to the following properties: `port`, `probes`, `resources`, `volumes`, `docker`, `files`, `image`, `user`, `privileged`, `entrypoint`, `working_dir`, `python`, `nvcc`, `single_branch`, `env`, `shell`, `commands`, as well as changes to [repo](#repos) or [file](#files) contents. - Changes to `replicas` and `scaling` can be applied without redeploying replicas. + Changes to `priority`, `replicas`, `scaling`, and `gateway` can be applied without redeploying replicas. Changes to other properties require a full service restart. diff --git a/src/dstack/_internal/core/models/configurations.py b/src/dstack/_internal/core/models/configurations.py index 344d12aca..09fd5eec7 100644 --- a/src/dstack/_internal/core/models/configurations.py +++ b/src/dstack/_internal/core/models/configurations.py @@ -1167,7 +1167,8 @@ class ServiceConfigurationParams(CoreModel): description=( "The name of the gateway. Specify boolean `false` to run without a gateway." " Specify boolean `true` to run with the default gateway." - " Omit to run with the default gateway if there is one, or without a gateway otherwise" + " Omit to run with the default gateway if there is one, or without a gateway otherwise." + " Can be updated in-place to migrate existing services between gateways" ), ), ] = None diff --git a/src/dstack/_internal/server/services/runs/__init__.py b/src/dstack/_internal/server/services/runs/__init__.py index bf3e1c575..4d9e4a2de 100644 --- a/src/dstack/_internal/server/services/runs/__init__.py +++ b/src/dstack/_internal/server/services/runs/__init__.py @@ -675,6 +675,17 @@ async def apply_plan( raise ServerClientError( "Failed to apply plan. Resource has been changed. Try again or use force apply." ) + if ( + run_spec.configuration.type == "service" + and current_resource.run_spec.configuration.type == "service" + and run_spec.configuration.gateway != current_resource.run_spec.configuration.gateway + ): + await services.assign_service( + session=session, + run_model=current_resource_model, + run_spec=run_spec, + is_new_service_submission=False, + ) new_deployment_num = current_resource.deployment_num + 1 # FIXME: potentially long write transaction # Avoid getting run_model after update @@ -783,8 +794,9 @@ async def submit_run( ) if run_spec.configuration.type == "service": - # FIXME: Register services asynchronously in the background - await services.register_service(session, run_model, run_spec) + await services.assign_service( + session, run_model, run_spec, is_new_service_submission=True + ) service_config = run_spec.configuration global_replica_num = 0 # Global counter across all groups for unique replica_num diff --git a/src/dstack/_internal/server/services/runs/spec.py b/src/dstack/_internal/server/services/runs/spec.py index 0b5c506f5..47e3143b3 100644 --- a/src/dstack/_internal/server/services/runs/spec.py +++ b/src/dstack/_internal/server/services/runs/spec.py @@ -57,6 +57,7 @@ "replicas", "groups", "scaling", + "gateway", # rolling deployment # NOTE: keep this list in sync with the "Rolling deployment" section in services.md "port", @@ -433,7 +434,7 @@ def _check_dynamo_in_place_update_compatibility( _router_affecting_top_level_fields = tuple( f for f in _TYPE_SPECIFIC_CONF_UPDATABLE_FIELDS.get("service", []) - if f not in ("replicas", "groups", "scaling") + if f not in ("replicas", "groups", "scaling", "gateway") ) for field in _router_affecting_top_level_fields: if getattr(current_cfg, field, None) != getattr(new_cfg, field, None): diff --git a/src/dstack/_internal/server/services/services/__init__.py b/src/dstack/_internal/server/services/services/__init__.py index 8edb29efa..48a55714d 100644 --- a/src/dstack/_internal/server/services/services/__init__.py +++ b/src/dstack/_internal/server/services/services/__init__.py @@ -33,7 +33,12 @@ logger = get_logger(__name__) -async def register_service(session: AsyncSession, run_model: RunModel, run_spec: RunSpec): +async def assign_service( + session: AsyncSession, + run_model: RunModel, + run_spec: RunSpec, + is_new_service_submission: bool, +) -> None: assert isinstance(run_spec.configuration, ServiceConfiguration) if isinstance(run_spec.configuration.gateway, EntityReference) or isinstance( @@ -70,14 +75,21 @@ async def register_service(session: AsyncSession, run_model: RunModel, run_spec: "The service requires a gateway, but there is no default gateway in the project" ) + if ( + not is_new_service_submission + and (gateway.id if gateway is not None else None) == run_model.gateway_id + ): + return + if gateway is not None: - service_spec = await _register_service_in_gateway(session, run_model, run_spec, gateway) + service_spec = await _assign_service_to_gateway(session, run_model, run_spec, gateway) run_model.gateway = gateway # For faster registration for replica_model in get_gateway_replica_models(gateway): replica_model.skip_min_processing_interval = True elif not settings.FORBID_SERVICES_WITHOUT_GATEWAY: - service_spec = _register_service_in_server(session, run_model, run_spec) + service_spec = _assign_service_to_in_server_proxy(session, run_model, run_spec) + run_model.gateway = None else: raise ResourceNotExistsError( "This dstack-server installation forbids services without a gateway." @@ -86,7 +98,7 @@ async def register_service(session: AsyncSession, run_model: RunModel, run_spec: run_model.service_spec = service_spec.model_dump_json() -async def _register_service_in_gateway( +async def _assign_service_to_gateway( session: AsyncSession, run_model: RunModel, run_spec: RunSpec, gateway: GatewayModel ) -> ServiceSpec: assert run_spec.configuration.type == "service" @@ -150,7 +162,7 @@ async def _register_service_in_gateway( return service_spec -def _register_service_in_server( +def _assign_service_to_in_server_proxy( session: AsyncSession, run_model: RunModel, run_spec: RunSpec ) -> ServiceSpec: assert run_spec.configuration.type == "service" diff --git a/src/tests/_internal/server/routers/test_runs.py b/src/tests/_internal/server/routers/test_runs.py index b1e3eef18..5ba350efb 100644 --- a/src/tests/_internal/server/routers/test_runs.py +++ b/src/tests/_internal/server/routers/test_runs.py @@ -3073,19 +3073,39 @@ async def test_collects_offers_only_if_requested_by_for_offers_only( commands=["one", "two"], port=80, gateway=None, + auth=True, replicas=Range(min=1, max=1), scaling=None, ), ServiceConfiguration( commands=["one", "two"], port=8080, - gateway="test-gateway", # not updatable + gateway="test-gateway", + auth=False, # not updatable replicas=Range(min=2, max=4), scaling=ScalingSpec(metric="rps", target=5), ), "create", id="no-update-service", ), + pytest.param( + ServiceConfiguration( + commands=["one", "two"], + port=80, + gateway=None, + replicas=Range(min=1, max=1), + scaling=None, + ), + ServiceConfiguration( + commands=["one", "two"], + port=8080, + gateway="test-gateway", + replicas=Range(min=2, max=4), + scaling=ScalingSpec(metric="rps", target=5), + ), + "update", + id="update-service-gateway", + ), pytest.param( DevEnvironmentConfiguration(ide="vscode", inactivity_duration=False), DevEnvironmentConfiguration(ide="vscode", inactivity_duration="30m"), @@ -3654,6 +3674,364 @@ async def test_patches_service_configuration_probes_for_old_clients( assert response.json()["run_spec"]["configuration"]["probes"] == expected_probes +@pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True) +class TestApplyPlanServiceGatewayUpdate: + async def _create_service_run( + self, + session: AsyncSession, + project, + repo, + user, + gateway=None, + gateway_field=None, + ) -> Tuple[RunModel, RunSpec]: + run_spec = get_run_spec( + run_name="test-service", + repo_id=repo.name, + configuration=ServiceConfiguration( + type="service", + commands=["one", "two"], + port=80, + gateway=gateway_field, + replicas=Range(min=1, max=1), + ), + ) + validate_run_spec_and_set_defaults(user, run_spec) + run_model = await create_run( + session=session, + project=project, + repo=repo, + user=user, + run_name=run_spec.run_name, + run_spec=run_spec, + gateway=gateway, + ) + return run_model, run_spec + + async def _apply( + self, + client: AsyncClient, + project, + user, + run_model: RunModel, + run_spec: RunSpec, + ): + run = run_model_to_run(run_model) + return await client.post( + f"/api/project/{project.name}/runs/apply", + headers=get_auth_headers(user.token), + json=json.loads( + ApplyRunPlanRequest( + plan=ApplyRunPlanInput(run_spec=run_spec, current_resource=run), + force=False, + ).model_dump_json() + ), + ) + + @pytest.mark.asyncio + async def test_assigns_to_specified_gateway_on_field_change( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="my-gateway", + wildcard_domain="my-gateway.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=gateway.id) + + run_model, run_spec = await self._create_service_run(session, project, repo, user) + run_spec.configuration.gateway = "my-gateway" + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + assert response.json()["service"]["url"] == "https://test-service.my-gateway.example" + await session.refresh(run_model) + assert run_model.gateway_id == gateway.id + event_messages = {e.message for e in await list_events(session)} + assert "Service assigned to gateway" in event_messages + + @pytest.mark.asyncio + async def test_reassigns_between_specific_gateways_on_field_change( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + old_gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="old-gateway", + wildcard_domain="old-gateway.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=old_gateway.id) + new_gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="new-gateway", + wildcard_domain="new-gateway.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=new_gateway.id) + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway=old_gateway, gateway_field="old-gateway" + ) + run_spec.configuration.gateway = "new-gateway" + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + assert response.json()["service"]["url"] == "https://test-service.new-gateway.example" + await session.refresh(run_model) + assert run_model.gateway_id == new_gateway.id + + @pytest.mark.asyncio + async def test_unassigns_gateway_when_field_changes_to_false( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user, name="test-project") + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="my-gateway", + wildcard_domain="my-gateway.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=gateway.id) + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway=gateway, gateway_field="my-gateway" + ) + run_spec.configuration.gateway = False + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + assert response.json()["service"]["url"] == "/proxy/services/test-project/test-service/" + await session.refresh(run_model) + assert run_model.gateway_id is None + event_messages = {e.message for e in await list_events(session)} + assert "Service assigned to run without a gateway" in event_messages + + @pytest.mark.asyncio + async def test_reassigns_to_default_gateway_when_field_changes_to_true( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + default_gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="default-gateway", + wildcard_domain="default-gateway.example", + ) + await create_gateway_replica( + session=session, backend=backend, gateway_id=default_gateway.id + ) + project.default_gateway_id = default_gateway.id + await session.commit() + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway_field=False + ) + run_spec.configuration.gateway = True + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + await session.refresh(run_model) + assert run_model.gateway_id == default_gateway.id + + @pytest.mark.asyncio + async def test_reassigns_to_default_gateway_when_field_changes_to_true_even_if_assigned( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="my-gateway", + wildcard_domain="my-gateway.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=gateway.id) + default_gateway = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="default-gateway", + wildcard_domain="default-gateway.example", + ) + await create_gateway_replica( + session=session, backend=backend, gateway_id=default_gateway.id + ) + project.default_gateway_id = default_gateway.id + await session.commit() + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway=gateway, gateway_field="my-gateway" + ) + run_spec.configuration.gateway = True + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + await session.refresh(run_model) + assert run_model.gateway_id == default_gateway.id + + @pytest.mark.asyncio + async def test_no_reassignment_when_gateway_field_unchanged( + self, test_db, session: AsyncSession, client: AsyncClient + ): + """Changing which gateway is the project's default must not, by itself, move a service + configured with `gateway: true` — only an explicit change to the `gateway` field + should trigger reassignment.""" + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + backend = await create_backend(session=session, project_id=project.id) + gateway_a = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="gateway-a", + wildcard_domain="gateway-a.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=gateway_a.id) + project.default_gateway_id = gateway_a.id + await session.commit() + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway=gateway_a, gateway_field=True + ) + + # The project's default gateway changes, but the run spec's `gateway` field is untouched. + gateway_b = await create_gateway( + session=session, + project_id=project.id, + backend_id=backend.id, + status=GatewayStatus.RUNNING, + name="gateway-b", + wildcard_domain="gateway-b.example", + ) + await create_gateway_replica(session=session, backend=backend, gateway_id=gateway_b.id) + project.default_gateway_id = gateway_b.id + await session.commit() + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + await session.refresh(run_model) + assert run_model.gateway_id == gateway_a.id + + @pytest.mark.asyncio + async def test_rejects_update_to_nonexistent_gateway( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + + run_model, run_spec = await self._create_service_run(session, project, repo, user) + run_spec.configuration.gateway = "nonexistent-gateway" + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 400, response.json() + await session.refresh(run_model) + assert run_model.gateway_id is None + + @pytest.mark.asyncio + async def test_rejects_field_change_to_true_when_no_default_gateway( + self, test_db, session: AsyncSession, client: AsyncClient + ): + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway_field=False + ) + run_spec.configuration.gateway = True + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 400, response.json() + await session.refresh(run_model) + assert run_model.gateway_id is None + + @pytest.mark.asyncio + async def test_no_reassignment_when_resolved_gateway_is_unchanged( + self, test_db, session: AsyncSession, client: AsyncClient + ): + """Even though the literal `gateway` field changes, if it still resolves to the + gateway the service is already assigned to (here: no gateway, since no default exists + either way), no reassignment should happen — no event, and `service_spec` untouched.""" + user = await create_user(session=session, global_role=GlobalRole.USER) + project = await create_project(session=session, owner=user) + await add_project_member( + session=session, project=project, user=user, project_role=ProjectRole.USER + ) + repo = await create_repo(session=session, project_id=project.id) + + run_model, run_spec = await self._create_service_run( + session, project, repo, user, gateway_field=None + ) + assert run_model.service_spec is None + run_spec.configuration.gateway = False + + response = await self._apply(client, project, user, run_model, run_spec) + assert response.status_code == 200, response.json() + await session.refresh(run_model) + assert run_model.gateway_id is None + assert run_model.service_spec is None + event_messages = {e.message for e in await list_events(session)} + assert "Service assigned to gateway" not in event_messages + assert "Service assigned to run without a gateway" not in event_messages + + class TestStopRuns: @pytest.mark.asyncio @pytest.mark.parametrize("test_db", ["sqlite", "postgres"], indirect=True)