Support in-place update for service gateway

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.
 

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

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

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):

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"