Merge remote-tracking branch 'origin/master' into issue_4126_gateway_default_in_place_update

Author: jvstme

src/dstack/_internal/server/app.py

@@ -55,7 +55,6 @@
 from dstack._internal.server.services.jobs.server_connection import job_server_connections_pool
 from dstack._internal.server.services.locking import advisory_lock_ctx
 from dstack._internal.server.services.projects import get_or_create_default_project
-from dstack._internal.server.services.prometheus.client_metrics import http_metrics
 from dstack._internal.server.services.proxy.deps import ServerProxyDependencyInjector
 from dstack._internal.server.services.proxy.routers import service_proxy
 from dstack._internal.server.services.runner.pool import instance_connection_pool
@@ -299,8 +298,6 @@ async def log_request(request: Request, call_next):
         start_time = time.time()
         response: Response = await call_next(request)
         process_time = time.time() - start_time
-        # log process_time to be used in the log_http_metrics middleware
-        request.state.process_time = process_time
         logger.debug(
             "Processed request %s %s in %s. Status: %s",
             request.method,
@@ -327,42 +324,6 @@ async def profile_request(request: Request, call_next):
             else:
                 return await call_next(request)
 
-    # this middleware must be defined after the log_request middleware
-    @app.middleware("http")
-    async def log_http_metrics(request: Request, call_next):
-        def _extract_project_name(request: Request):
-            project_name = None
-            prefix = "/api/project/"
-            if request.url.path.startswith(prefix):
-                rest = request.url.path[len(prefix) :]
-                project_name = rest.split("/", 1)[0] if rest else None
-
-            return project_name
-
-        def _extract_endpoint_label(request: Request, response: Response) -> str:
-            route = request.scope.get("route")
-            route_path = getattr(route, "path", None)
-            if route_path:
-                return route_path
-            if not request.url.path.startswith("/api/"):
-                return "__non_api__"
-            if response.status_code == status.HTTP_404_NOT_FOUND:
-                return "__not_found__"
-            return "__unmatched__"
-
-        project_name = _extract_project_name(request)
-        response: Response = await call_next(request)
-        endpoint_label = _extract_endpoint_label(request, response)
-
-        http_metrics.log_request(
-            method=request.method,
-            endpoint=endpoint_label,
-            http_status=response.status_code,
-            project_name=project_name,
-            duration_seconds=request.state.process_time,
-        )
-        return response
-
     @app.get("/healthcheck")
     async def healthcheck():
         return CustomJSONResponse(content={"status": "running"})

src/dstack/_internal/server/background/pipeline_tasks/fleets.py

@@ -38,6 +38,7 @@
     InstanceModel,
     JobModel,
     PlacementGroupModel,
+    ProjectModel,
     RunModel,
 )
 from dstack._internal.server.services import events
@@ -338,7 +339,7 @@ async def _refetch_locked_fleet_for_processing(
             FleetModel.id == item.id,
             FleetModel.lock_token == item.lock_token,
         )
-        .options(joinedload(FleetModel.project))
+        .options(joinedload(FleetModel.project).joinedload(ProjectModel.owner))
         .options(
             selectinload(FleetModel.instances.and_(InstanceModel.deleted == False))
             .joinedload(InstanceModel.jobs)

src/dstack/_internal/server/background/pipeline_tasks/jobs_running.py

@@ -20,7 +20,6 @@
 )
 from dstack._internal.core.models.configurations import (
     DevEnvironmentConfiguration,
-    ServiceConfiguration,
 )
 from dstack._internal.core.models.files import FileArchiveMapping
 from dstack._internal.core.models.instances import InstanceStatus, SSHConnectionParams
@@ -34,7 +33,6 @@
     Job,
     JobProvisioningData,
     JobRuntimeData,
-    JobSpec,
     JobStatus,
     JobSubmission,
     JobTerminationReason,
@@ -366,6 +364,7 @@ class _JobUpdateMap(ItemUpdateMap, total=False):
     disconnected_at: Optional[datetime]
     inactivity_secs: Optional[int]
     exit_status: Optional[int]
+    ready: bool
     registered: bool
     image_pull_progress: Optional[str]
     skip_min_processing_interval: bool
@@ -1083,6 +1082,12 @@ def _emit_result_events(
             job_model.disconnected_at,
         ),
     )
+    _emit_readiness_change_event(
+        session=session,
+        job_model=job_model,
+        old_ready=job_model.ready,
+        new_ready=result.job_update_map.get("ready", job_model.ready),
+    )
     if result.replica_registration is not None:
         targets = [events.Target.from_model(job_model)]
         if result.replica_registration.gateway_target is not None:
@@ -1175,13 +1180,32 @@ async def _maybe_register_replica(
 ) -> None:
     if (
         context.run.run_spec.configuration.type != "service"
-        or _get_result_registered(context.job_model, result)
         or context.job_model.job_num != 0
         or result.new_probe_models
-        or not is_job_ready(context.job_model.probes, context.job.job_spec.probes)
     ):
         return
 
+    is_ready = is_job_ready(context.job_model.probes, context.job.job_spec.probes)
+    if is_ready and not context.job_model.ready:
+        result.job_update_map["ready"] = True
+
+    router_group = next(
+        (g for g in context.run.run_spec.configuration.replica_groups if g.router is not None),
+        None,
+    )
+    is_router_replica = (
+        router_group is not None and context.job.job_spec.replica_group == router_group.name
+    )
+    # non-router replicas aren't registered if the service has a router
+    if router_group is not None and not is_router_replica:
+        if context.job_model.registered:
+            # migration edge case: a pre-0.21.1 server replica incorrectly set registered=True
+            result.job_update_map["registered"] = False
+        return
+
+    if not is_ready or _get_result_registered(context.job_model, result):
+        return
+
     ssh_head_proxy: Optional[SSHConnectionParams] = None
     ssh_head_proxy_private_key: Optional[str] = None
     instance = get_or_error(context.job_model.instance)
@@ -1220,23 +1244,6 @@ async def _register_service_replica(
 ) -> Optional[events.Target]:
     if context.run_model.gateway_id is None:
         return None
-
-    job_spec = validate_json_extra_ignore(JobSpec, context.job_model.job_spec_data)
-
-    # For router-based services (e.g. PD disaggregation), only router replicas should be
-    # registered with the gateway. Worker replicas are discovered by the router-worker
-    # sync pipeline and should not be routed to directly by the gateway.
-    config = context.run.run_spec.configuration
-    assert isinstance(config, ServiceConfiguration)
-    router_group = next((g for g in config.replica_groups if g.router is not None), None)
-    is_router_replica = router_group is not None and job_spec.replica_group == router_group.name
-    if router_group is not None and not is_router_replica:
-        logger.debug(
-            "%s: skipping gateway replica registration (non-router replica)",
-            fmt(context.job_model),
-        )
-        return None
-
     async with get_session_ctx() as session:
         gateway_model, connections = await get_or_add_gateway_connections(
             session, context.run_model.gateway_id
@@ -1261,7 +1268,7 @@ async def _register_service_replica(
             async with conn.client() as gateway_client:
                 await gateway_client.register_replica(
                     run=context.run,
-                    job_spec=job_spec,
+                    job_spec=context.job.job_spec,
                     job_submission=job_submission,
                     instance_project_ssh_private_key=instance_project_ssh_private_key,
                     ssh_head_proxy=ssh_head_proxy,
@@ -1877,6 +1884,23 @@ def _emit_reachability_change_event(
         )
 
 
+def _emit_readiness_change_event(
+    session: AsyncSession,
+    job_model: JobModel,
+    old_ready: bool,
+    new_ready: bool,
+) -> None:
+    # ready: False -> True
+    if not old_ready and new_ready:
+        events.emit(
+            session,
+            "Service replica ready to receive requests",
+            actor=events.SystemActor(),
+            targets=[events.Target.from_model(job_model)],
+        )
+    # ready: True -> False is not possible as of this writing
+
+
 def _terminate_job(
     job_model: JobModel,
     job_update_map: _JobUpdateMap,

src/dstack/_internal/server/background/pipeline_tasks/runs/active.py

@@ -686,11 +686,11 @@ async def _build_rolling_deployment_maps(
                 max_new = max(j.replica_num for j in new_jobs)
                 next_replica_num = max(next_replica_num, max_new + 1)
 
-        # Scale down: terminate unregistered out-of-date + excess registered replicas
-        replicas_to_stop = state.unregistered_out_of_date_replica_count
+        # Scale down: terminate unready out-of-date + excess ready replicas
+        replicas_to_stop = state.unready_out_of_date_replica_count
         replicas_to_stop += max(
             0,
-            state.registered_non_terminating_replica_count - group_desired,
+            state.ready_non_terminating_replica_count - group_desired,
         )
         if replicas_to_stop > 0:
             scale_down_maps = _build_scale_down_job_update_maps(

src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py

@@ -233,13 +233,12 @@ async def process(self, item: ServiceRouterWorkerSyncPipelineItem) -> None:
                     selectinload(
                         RunModel.jobs.and_(
                             JobModel.status == JobStatus.RUNNING,
-                            JobModel.registered == True,
+                            JobModel.ready == True,
                         )
                     )
                     .load_only(
                         JobModel.id,
                         JobModel.status,
-                        JobModel.registered,
                         JobModel.job_spec_data,
                         JobModel.job_provisioning_data,
                         JobModel.job_runtime_data,

src/dstack/_internal/server/migrations/versions/2026/08_04_0702_72cfa56364ad_add_jobmodel_ready.py

@@ -0,0 +1,120 @@
+"""Add JobModel.ready
+
+Revision ID: 72cfa56364ad
+Revises: a1c3f5e7b209
+Create Date: 2026-08-04 07:02:02.041000+00:00
+
+"""
+
+import json
+import uuid
+from typing import Optional
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy_utils import UUIDType
+
+# revision identifiers, used by Alembic.
+revision = "72cfa56364ad"
+down_revision = "a1c3f5e7b209"
+branch_labels = None
+depends_on = None
+
+# Partial table descriptions - only columns needed for the data migration below.
+jobs_table = sa.Table(
+    "jobs",
+    sa.MetaData(),
+    sa.Column("id", UUIDType(binary=False), primary_key=True, default=uuid.uuid4),
+    sa.Column("run_id", UUIDType(binary=False)),
+    sa.Column("status", sa.String(100)),
+    sa.Column("registered", sa.Boolean()),
+    sa.Column("ready", sa.Boolean()),
+    sa.Column("job_spec_data", sa.Text()),
+)
+runs_table = sa.Table(
+    "runs",
+    sa.MetaData(),
+    sa.Column("id", UUIDType(binary=False), primary_key=True, default=uuid.uuid4),
+    sa.Column("run_spec", sa.Text()),
+    sa.Column("service_spec", sa.Text(), nullable=True),
+)
+
+
+def _get_router_group_name(run_spec_data: str) -> Optional[str]:
+    configuration = json.loads(run_spec_data).get("configuration") or {}
+    if configuration.get("type") != "service":
+        return None
+    replica_groups = configuration.get("replicas")
+    if not isinstance(replica_groups, list):
+        return None
+    for group in replica_groups:
+        if isinstance(group, dict) and group.get("router") is not None:
+            return group.get("name")
+    return None
+
+
+def _get_job_replica_group(job_spec_data: str) -> str:
+    return json.loads(job_spec_data).get("replica_group", "0")
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("jobs", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column("ready", sa.Boolean(), server_default=sa.false(), nullable=False)
+        )
+
+    # ### end Alembic commands ###
+
+    bind = op.get_bind()
+
+    # backfill ready=True for existing registered replicas
+
+    bind.execute(jobs_table.update().where(jobs_table.c.registered == True).values(ready=True))
+
+    # set registered=False for non-router replicas in services with a router
+
+    candidate_jobs = bind.execute(
+        sa.select(jobs_table.c.id, jobs_table.c.run_id, jobs_table.c.job_spec_data)
+        .select_from(jobs_table.join(runs_table, jobs_table.c.run_id == runs_table.c.id))
+        .where(
+            jobs_table.c.registered == True,
+            jobs_table.c.status == "RUNNING",
+            runs_table.c.service_spec.is_not(None),
+            sa.or_(
+                runs_table.c.run_spec.like('%"sglang"%'),
+                runs_table.c.run_spec.like('%"dynamo"%'),
+            ),
+        )
+    ).all()
+
+    run_ids = {row.run_id for row in candidate_jobs}
+    router_group_name_by_run_id = {}
+    if run_ids:
+        for row in bind.execute(
+            sa.select(runs_table.c.id, runs_table.c.run_spec).where(runs_table.c.id.in_(run_ids))
+        ).all():
+            router_group_name_by_run_id[row.id] = _get_router_group_name(row.run_spec)
+
+    non_router_job_ids = []
+    for row in candidate_jobs:
+        router_group_name = router_group_name_by_run_id.get(row.run_id)
+        if router_group_name is None:
+            continue
+        if _get_job_replica_group(row.job_spec_data) != router_group_name:
+            non_router_job_ids.append(row.id)
+
+    if non_router_job_ids:
+        bind.execute(
+            jobs_table.update()
+            .where(jobs_table.c.id.in_(non_router_job_ids))
+            .values(registered=False)
+        )
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("jobs", schema=None) as batch_op:
+        batch_op.drop_column("ready")
+
+    # ### end Alembic commands ###