Add in-place update for gateway domain (#3997)

Introduce gateway in-place update mechanism. For
now, only `domain` can be updated.

```yaml
$ dstack apply -f test.dstack.yml

Found gateway test-gateway. Detected changes that can be updated in-place:
- domain

Update the gateway? [y/n]: y
 NAME          BACKEND         HOSTNAME       DOMAIN           DEFAULT  STATUS
 test-gateway  gcp (us-west4)  34.125.56.225  new.example.com           running
```

Add new API methods:
- `/api/project/{project_name}/gateways/get_plan`
- `/api/project/{project_name}/gateways/apply`

Deprecate API methods:
- `/api/project/{project_name}/gateways/create`
- `/api/project/{project_name}/gateways/set_wildcard_domain`

Deprecate CLI arguments:
- `dstack gateway update --domain`

Author: jvstme

src/dstack/_internal/cli/commands/gateway.py

@@ -85,7 +85,7 @@ def _register(self):
         update_parser.add_argument(
             "--set-default", action="store_true", help="Set it the default gateway for the project"
         )
-        update_parser.add_argument("--domain", help="Set the domain for the gateway")
+        update_parser.add_argument("--domain", help="(deprecated) Set the domain for the gateway")
 
         get_parser = subparsers.add_parser(
             "get", help="Get a gateway", formatter_class=self._parser.formatter_class
@@ -170,6 +170,10 @@ def _delete(self, args: argparse.Namespace):
     def _update(self, args: argparse.Namespace):
         with console.status("Updating gateway..."):
             if args.domain:
+                logger.warning(
+                    "`dstack gateway update --domain` is deprecated and may be disallowed in the future."
+                    " Use `dstack apply` to update domains or other gateway configuration properties"
+                )
                 if args.name.project is not None:
                     console.print(
                         "The [code]<project>/<gateway>[/] format is not supported for gateway names"

src/dstack/_internal/cli/services/configurators/gateway.py

@@ -11,9 +11,14 @@
 )
 from dstack._internal.cli.utils.gateway import get_gateways_table
 from dstack._internal.cli.utils.rich import MultiItemStatus
-from dstack._internal.core.errors import ResourceNotExistsError
+from dstack._internal.core.errors import (
+    MethodNotAllowedError,
+    ResourceNotExistsError,
+)
+from dstack._internal.core.models.common import ApplyAction
 from dstack._internal.core.models.configurations import ApplyConfigurationType
 from dstack._internal.core.models.gateways import (
+    ApplyGatewayPlanInput,
     Gateway,
     GatewayConfiguration,
     GatewayPlan,
@@ -23,6 +28,7 @@
 from dstack._internal.core.services.diff import diff_models
 from dstack._internal.utils.common import local_time
 from dstack._internal.utils.logging import get_logger
+from dstack._internal.utils.nested_list import NestedList, NestedListItem
 from dstack.api._public import Client
 
 logger = get_logger(__name__)
@@ -51,26 +57,31 @@ def apply_configuration(
                 " https://dstack.ai/docs/concepts/services/#pd-disaggregation"
             )
         with console.status("Getting apply plan..."):
-            plan = _get_plan(api=self.api, spec=spec)
+            try:
+                plan = self.api.client.gateways.get_plan(project_name=self.api.project, spec=spec)
+                use_legacy_api = False
+            except MethodNotAllowedError:
+                # pre-0.20.27 server
+                plan = _get_plan_legacy(self.api, spec)
+                use_legacy_api = True
         _print_plan_header(plan)
 
         action_message = ""
         confirm_message = ""
+        delete_gateway_name = None
         if plan.current_resource is None:
-            if plan.spec.configuration.name is not None:
-                action_message += (
-                    f"Gateway [code]{plan.spec.configuration.name}[/] does not exist yet."
-                )
+            if plan.effective_spec.configuration.name is not None:
+                action_message += f"Gateway [code]{plan.effective_spec.configuration.name}[/] does not exist yet."
             confirm_message += "Create the gateway?"
         else:
-            action_message += f"Found gateway [code]{plan.spec.configuration.name}[/]."
+            action_message += f"Found gateway [code]{plan.effective_spec.configuration.name}[/]."
             diff = diff_models(
-                plan.spec.configuration,
                 plan.current_resource.configuration,
+                plan.effective_spec.configuration,
             )
             changed_fields = list(diff.keys())
             if (
-                plan.current_resource.configuration == plan.spec.configuration
+                plan.current_resource.configuration == plan.effective_spec.configuration
                 or changed_fields == ["default"]
             ):
                 if command_args.yes and not command_args.force:
@@ -82,38 +93,61 @@ def apply_configuration(
                     return
                 action_message += " No configuration changes detected."
                 confirm_message += "Re-create the gateway?"
+                delete_gateway_name = plan.current_resource.name
             else:
-                action_message += " Configuration changes detected."
-                confirm_message += "Re-create the gateway?"
+                formatted_diff = NestedList(
+                    children=[NestedListItem(field) for field in diff]
+                ).render()
+                if plan.action == ApplyAction.UPDATE:
+                    action_message += f" Detected changes that [code]can[/] be updated in-place:\n{formatted_diff}"
+                    confirm_message += "Update the gateway?"
+                else:
+                    action_message += f" Detected changes that [error]cannot[/] be updated in-place:\n{formatted_diff}"
+                    confirm_message += "Re-create the gateway?"
+                    delete_gateway_name = plan.current_resource.name
 
         console.print(action_message)
         if not command_args.yes and not confirm_ask(confirm_message):
             console.print("\nExiting...")
             return
 
-        if plan.current_resource is not None:
+        if delete_gateway_name is not None:
             with console.status("Deleting existing gateway..."):
                 self.api.client.gateways.delete(
                     project_name=self.api.project,
-                    gateways_names=[plan.current_resource.name],
+                    gateways_names=[delete_gateway_name],
                 )
                 # Gateway deletion is async. Wait for gateway to be deleted.
                 while True:
                     try:
                         self.api.client.gateways.get(
                             project_name=self.api.project,
-                            gateway_name=plan.current_resource.name,
+                            gateway_name=delete_gateway_name,
                         )
                     except ResourceNotExistsError:
                         break
                     else:
                         time.sleep(1)
 
-        with console.status("Creating gateway..."):
-            gateway = self.api.client.gateways.create(
-                project_name=self.api.project,
-                configuration=conf,
-            )
+        with console.status("Applying plan..."):
+            if use_legacy_api:
+                gateway = self.api.client.gateways.create(
+                    project_name=self.api.project,
+                    configuration=conf,
+                )
+            else:
+                gateway = self.api.client.gateways.apply_plan(
+                    project_name=self.api.project,
+                    plan=ApplyGatewayPlanInput(
+                        spec=spec,
+                        current_resource=plan.current_resource,
+                    ),
+                )
+
+        if plan.action == ApplyAction.UPDATE and delete_gateway_name is None:
+            console.print(get_gateways_table([gateway], current_project=self.api.project))
+            return
+
         if command_args.detach:
             console.print("Gateway configuration submitted. Exiting...")
             return
@@ -195,8 +229,7 @@ def apply_args(self, conf: GatewayConfiguration, args: argparse.Namespace):
             conf.name = args.name
 
 
-def _get_plan(api: Client, spec: GatewaySpec) -> GatewayPlan:
-    # TODO: Implement server-side /get_plan with an offer included
+def _get_plan_legacy(api: Client, spec: GatewaySpec) -> GatewayPlan:
     user = api.client.users.get_my_user()
     current_resource = None
     if spec.configuration.name is not None:
@@ -211,7 +244,9 @@ def _get_plan(api: Client, spec: GatewaySpec) -> GatewayPlan:
         project_name=api.project,
         user=user.username,
         spec=spec,
+        effective_spec=spec,
         current_resource=current_resource,
+        action=ApplyAction.CREATE,
     )
 
 
@@ -225,20 +260,22 @@ def th(s: str) -> str:
 
     configuration_table.add_row(th("Project"), plan.project_name)
     configuration_table.add_row(th("User"), plan.user)
-    configuration_table.add_row(th("Configuration"), plan.spec.configuration_path)
-    configuration_table.add_row(th("Type"), plan.spec.configuration.type)
+    configuration_table.add_row(th("Configuration"), plan.effective_spec.configuration_path)
+    configuration_table.add_row(th("Type"), plan.effective_spec.configuration.type)
 
     domain = "-"
-    if plan.spec.configuration.domain is not None:
-        domain = plan.spec.configuration.domain
+    if plan.effective_spec.configuration.domain is not None:
+        domain = plan.effective_spec.configuration.domain
 
-    configuration_table.add_row(th("Backend"), plan.spec.configuration.backend.value)
-    configuration_table.add_row(th("Region"), plan.spec.configuration.region)
+    configuration_table.add_row(th("Backend"), plan.effective_spec.configuration.backend.value)
+    configuration_table.add_row(th("Region"), plan.effective_spec.configuration.region)
     configuration_table.add_row(th("Domain"), domain)
 
-    if plan.spec.configuration.replicas is not None:
-        assert isinstance(plan.spec.configuration.replicas, int)
-        configuration_table.add_row(th("Replicas"), str(plan.spec.configuration.replicas))
+    if plan.effective_spec.configuration.replicas is not None:
+        assert isinstance(plan.effective_spec.configuration.replicas, int)
+        configuration_table.add_row(
+            th("Replicas"), str(plan.effective_spec.configuration.replicas)
+        )
 
     console.print(configuration_table)
     console.print()

src/dstack/_internal/core/models/gateways.py

@@ -7,7 +7,7 @@
 from typing_extensions import Annotated, Literal
 
 from dstack._internal.core.models.backends.base import BackendType
-from dstack._internal.core.models.common import CoreModel
+from dstack._internal.core.models.common import ApplyAction, CoreModel
 from dstack._internal.core.models.routers import AnyGatewayRouterConfig
 from dstack._internal.utils.tags import tags_validator
 
@@ -88,7 +88,8 @@ class GatewayConfiguration(CoreModel):
                 "The gateway wildcard domain name, e.g. `example.com`."
                 " Service domain names are constructed as `<run name>.<gateway domain`."
                 " The domain name can use the `${{ run.project_name }}` variable"
-                " to include the service’s project name"
+                " to include the service’s project name."
+                " Can be updated in-place. Updates do not affect existing services"
             )
         ),
     ] = None
@@ -176,7 +177,22 @@ class GatewayPlan(CoreModel):
     project_name: str
     user: str
     spec: GatewaySpec
+    effective_spec: GatewaySpec
     current_resource: Optional[Gateway] = None
+    action: ApplyAction
+
+
+class ApplyGatewayPlanInput(CoreModel):
+    spec: GatewaySpec
+    current_resource: Annotated[
+        Optional[Gateway],
+        Field(
+            description=(
+                "The expected current resource."
+                " If the resource has changed, the apply fails unless `force: true`."
+            )
+        ),
+    ] = None
 
 
 class GatewayComputeConfiguration(CoreModel):

src/dstack/_internal/server/compatibility/gateways.py

@@ -2,7 +2,7 @@
 
 from packaging.version import Version
 
-from dstack._internal.core.models.gateways import Gateway
+from dstack._internal.core.models.gateways import Gateway, GatewayPlan
 
 
 def patch_gateway(gateway: Gateway, client_version: Optional[Version]) -> None:
@@ -21,3 +21,10 @@ def patch_gateway(gateway: Gateway, client_version: Optional[Version]) -> None:
                 replica.region = ""
             if replica.backend is None:
                 replica.backend = gateway.configuration.backend
+
+
+def patch_gateway_plan(plan: GatewayPlan, client_version: Optional[Version]) -> None:
+    if client_version is None:
+        return
+    if plan.current_resource is not None:
+        patch_gateway(plan.current_resource, client_version)

src/dstack/_internal/server/routers/gateways.py

@@ -1,4 +1,4 @@
-from typing import List, Optional, Tuple
+from typing import Annotated, List, Optional, Tuple
 
 from fastapi import APIRouter, Depends
 from packaging.version import Version
@@ -9,7 +9,7 @@
 import dstack._internal.server.services.gateways as gateways
 from dstack._internal.core.errors import ResourceNotExistsError
 from dstack._internal.core.models.common import EntityReference
-from dstack._internal.server.compatibility.gateways import patch_gateway
+from dstack._internal.server.compatibility.gateways import patch_gateway, patch_gateway_plan
 from dstack._internal.server.db import get_session
 from dstack._internal.server.deps import Project
 from dstack._internal.server.models import ProjectModel, UserModel
@@ -71,14 +71,63 @@ async def get_gateway(
     return CustomORJSONResponse(gateway)
 
 
-@router.post("/create", summary="Create gateway", response_model=models.Gateway)
+@router.post("/get_plan", summary="Get gateway plan", response_model=models.GatewayPlan)
+async def get_plan(
+    body: schemas.GetGatewayPlanRequest,
+    session: Annotated[AsyncSession, Depends(get_session)],
+    user_project: Annotated[tuple[UserModel, ProjectModel], Depends(ProjectAdmin())],
+    client_version: Annotated[Optional[Version], Depends(get_client_version)],
+):
+    """
+    Returns a gateway plan for the given gateway spec.
+    This is an optional step before calling `/apply`.
+    """
+    user, project = user_project
+    plan = await gateways.get_plan(
+        session=session,
+        project=project,
+        user=user,
+        spec=body.spec,
+    )
+    patch_gateway_plan(plan, client_version)
+    return CustomORJSONResponse(plan)
+
+
+@router.post("/apply", summary="Apply gateway plan", response_model=models.Gateway)
+async def apply_plan(
+    body: schemas.ApplyGatewayPlanRequest,
+    session: Annotated[AsyncSession, Depends(get_session)],
+    user_project: Annotated[tuple[UserModel, ProjectModel], Depends(ProjectAdmin())],
+    pipeline_hinter: Annotated[PipelineHinterProtocol, Depends(get_pipeline_hinter)],
+    client_version: Annotated[Optional[Version], Depends(get_client_version)],
+):
+    """
+    Creates a new gateway or updates an existing gateway in-place.
+    """
+    user, project = user_project
+    gateway = await gateways.apply_plan(
+        session=session,
+        user=user,
+        project=project,
+        plan=body.plan,
+        force=body.force,
+        pipeline_hinter=pipeline_hinter,
+    )
+    patch_gateway(gateway, client_version)
+    return CustomORJSONResponse(gateway)
+
+
+@router.post("/create", summary="Create gateway", response_model=models.Gateway, deprecated=True)
 async def create_gateway(
     body: schemas.CreateGatewayRequest,
     session: AsyncSession = Depends(get_session),
     user_project: Tuple[UserModel, ProjectModel] = Depends(ProjectAdmin()),
     pipeline_hinter: PipelineHinterProtocol = Depends(get_pipeline_hinter),
     client_version: Optional[Version] = Depends(get_client_version),
 ):
+    """
+    Deprecated in favor of `/apply`.
+    """
     user, project = user_project
     gateway = await gateways.create_gateway(
         session=session,
@@ -121,13 +170,21 @@ async def set_default_gateway(
     )
 
 
-@router.post("/set_wildcard_domain", summary="Set wildcard domain", response_model=models.Gateway)
+@router.post(
+    "/set_wildcard_domain",
+    summary="Set wildcard domain",
+    response_model=models.Gateway,
+    deprecated=True,
+)
 async def set_gateway_wildcard_domain(
     body: schemas.SetWildcardDomainRequest,
     session: AsyncSession = Depends(get_session),
     user_project: Tuple[UserModel, ProjectModel] = Depends(ProjectAdmin()),
     client_version: Optional[Version] = Depends(get_client_version),
 ):
+    """
+    Deprecated in favor of `/apply` (in-place update).
+    """
     user, project = user_project
     gateway = await gateways.set_gateway_wildcard_domain(
         session=session,