Hint gateway pipeline

Author: r4victor

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

@@ -2,6 +2,7 @@
 
 from dstack._internal.server.background.pipeline_tasks.base import Pipeline
 from dstack._internal.server.background.pipeline_tasks.compute_groups import ComputeGroupPipeline
+from dstack._internal.server.background.pipeline_tasks.gateways import GatewayPipeline
 from dstack._internal.server.background.pipeline_tasks.placement_groups import (
     PlacementGroupPipeline,
 )
@@ -17,6 +18,7 @@ def __init__(self) -> None:
         if FeatureFlags.PIPELINE_PROCESSING_ENABLED:
             self._pipelines += [
                 ComputeGroupPipeline(),
+                GatewayPipeline(),
                 PlacementGroupPipeline(),
             ]
         self._hinter = PipelineHinter(self._pipelines)

src/dstack/_internal/server/background/scheduled_tasks/__init__.py

@@ -99,21 +99,23 @@ def start_scheduled_tasks() -> AsyncIOScheduler:
         )
         _scheduler.add_job(delete_prometheus_metrics, IntervalTrigger(minutes=5), max_instances=1)
     _scheduler.add_job(process_gateways_connections, IntervalTrigger(seconds=15))
-    _scheduler.add_job(process_gateways, IntervalTrigger(seconds=10, jitter=2), max_instances=5)
     _scheduler.add_job(
         process_submitted_volumes, IntervalTrigger(seconds=10, jitter=2), max_instances=5
     )
     _scheduler.add_job(
         process_idle_volumes, IntervalTrigger(seconds=60, jitter=10), max_instances=1
     )
-    if not FeatureFlags.PIPELINE_PROCESSING_ENABLED:
-        _scheduler.add_job(process_placement_groups, IntervalTrigger(seconds=30, jitter=5))
     _scheduler.add_job(
         process_fleets,
         IntervalTrigger(seconds=10, jitter=2),
         max_instances=1,
     )
     _scheduler.add_job(delete_instance_health_checks, IntervalTrigger(minutes=5), max_instances=1)
+    if not FeatureFlags.PIPELINE_PROCESSING_ENABLED:
+        _scheduler.add_job(
+            process_gateways, IntervalTrigger(seconds=10, jitter=2), max_instances=5
+        )
+        _scheduler.add_job(process_placement_groups, IntervalTrigger(seconds=30, jitter=5))
     for replica in range(settings.SERVER_BACKGROUND_PROCESSING_FACTOR):
         # Add multiple copies of tasks if requested.
         # max_instances=1 for additional copies to avoid running too many tasks.

src/dstack/_internal/server/models.py

@@ -524,6 +524,8 @@ class GatewayModel(PipelineModelMixin, BaseModel):
 
     __table_args__ = (UniqueConstraint("project_id", "name", name="uq_gateways_project_id_name"),)
 
+    # TODO: Add pipeline index ("ix_gateways_pipeline_fetch_q") if gateways become soft-deleted.
+
 
 class GatewayComputeModel(BaseModel):
     __tablename__ = "gateway_computes"

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

@@ -13,6 +13,7 @@
     ProjectAdmin,
     ProjectMemberOrPublicAccess,
 )
+from dstack._internal.server.services.pipelines import PipelineHinterProtocol, get_pipeline_hinter
 from dstack._internal.server.utils.routers import (
     CustomORJSONResponse,
     get_base_api_additional_responses,
@@ -54,6 +55,7 @@ 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),
 ):
     user, project = user_project
     return CustomORJSONResponse(
@@ -62,6 +64,7 @@ async def create_gateway(
             user=user,
             project=project,
             configuration=body.configuration,
+            pipeline_hinter=pipeline_hinter,
         )
     )
 

src/dstack/_internal/server/services/gateways/__init__.py

@@ -58,6 +58,7 @@
     get_locker,
     string_to_lock_id,
 )
+from dstack._internal.server.services.pipelines import PipelineHinterProtocol
 from dstack._internal.server.services.plugins import apply_plugin_policies
 from dstack._internal.server.utils.common import gather_map_async
 from dstack._internal.utils.common import get_current_datetime, run_async
@@ -184,6 +185,7 @@ async def create_gateway(
     user: UserModel,
     project: ProjectModel,
     configuration: GatewayConfiguration,
+    pipeline_hinter: PipelineHinterProtocol,
 ) -> Gateway:
     spec = await apply_plugin_policies(
         user=user.name,
@@ -238,6 +240,7 @@ async def create_gateway(
             await set_default_gateway(
                 session=session, project=project, name=configuration.name, user=user
             )
+        pipeline_hinter.hint_fetch(GatewayModel.__name__)
         return gateway_model_to_gateway(gateway)
 
 
@@ -303,7 +306,8 @@ async def delete_gateways(
         )
         gateway_models = res.scalars().all()
         if len(gateway_models) != len(gateways_ids):
-            # TODO: Make the delete endpoint fully async without lock – put the request in queue and process in background.
+            # TODO: Make the delete endpoint fully async so we don't need to lock and error:
+            # put the request in queue and process in the background.
             raise ServerClientError(
                 "Failed to delete gateways: gateways are being processed currently. Try again later."
             )

src/dstack/_internal/server/services/pipelines.py

@@ -5,8 +5,15 @@
 
 class PipelineHinterProtocol(Protocol):
     def hint_fetch(self, model_name: str) -> None:
+        """
+        Pass `Model.__name__` to hint replica's pipelines to fetch the model's items ASAP.
+        """
         pass
 
 
 def get_pipeline_hinter(request: Request) -> PipelineHinterProtocol:
+    """
+    Returns pipeline hinter that allows hinting replica's pipelines that there are new items for processing.
+    This can reduce processing latency if the processing happens rarely.
+    """
     return request.app.state.pipeline_manager.hinter