diff --git a/components/src/dynamo/planner/core/__init__.py b/components/src/dynamo/planner/core/__init__.py index 63b2bb13fde8..5584e2d9b9a5 100644 --- a/components/src/dynamo/planner/core/__init__.py +++ b/components/src/dynamo/planner/core/__init__.py @@ -3,9 +3,14 @@ from dynamo.planner.core.state_machine import PlannerScalingState from dynamo.planner.core.types import ( + BatchDispatcherFeedback, + BatchDrainLimitDecision, + BatchJobDemand, + BatchSchedulingObservation, EngineCapabilities, FpmObservations, PlannerEffects, + PoolTrafficDemand, ScalingDecision, ScheduledTick, TickInput, @@ -15,10 +20,15 @@ ) __all__ = [ + "BatchDispatcherFeedback", + "BatchDrainLimitDecision", + "BatchJobDemand", + "BatchSchedulingObservation", "EngineCapabilities", "FpmObservations", "PlannerEffects", "PlannerScalingState", + "PoolTrafficDemand", "ScalingDecision", "ScheduledTick", "TickInput", diff --git a/components/src/dynamo/planner/core/types.py b/components/src/dynamo/planner/core/types.py index a7676ac760f4..93b142dfd534 100644 --- a/components/src/dynamo/planner/core/types.py +++ b/components/src/dynamo/planner/core/types.py @@ -11,6 +11,7 @@ from __future__ import annotations +import math from dataclasses import dataclass, field from typing import TYPE_CHECKING, Optional @@ -88,6 +89,144 @@ class FpmObservations: decode: Optional[dict[tuple[str, int], ForwardPassMetrics]] = None +@dataclass +class BatchJobDemand: + """Batch job demand observed for one inference pool. + + ``observed_at_s`` and ``deadline_at_s`` are absolute wall-clock + timestamps. ``deadline_at_s=None`` means the job has no SLA deadline. + ``work_class`` is an opaque request/workload class understood by the + capacity model; it is deliberately not tied to a Batch Gateway enum. + + The raw counters and status are retained for auditability. + ``remaining_requests`` is derived so contradictory observations cannot + enter the planner contract. + """ + + observed_at_s: float + pool_id: str + job_id: str + status: str + total_requests: int + completed_requests: int + failed_requests: int + deadline_at_s: Optional[float] + work_class: str + + def __post_init__(self) -> None: + _require_finite_non_negative("observed_at_s", self.observed_at_s) + _require_non_empty_string("pool_id", self.pool_id) + _require_non_empty_string("job_id", self.job_id) + _require_non_empty_string("status", self.status) + _require_non_empty_string("work_class", self.work_class) + counters = ( + self.total_requests, + self.completed_requests, + self.failed_requests, + ) + if any( + isinstance(count, bool) or not isinstance(count, int) or count < 0 + for count in counters + ): + raise ValueError("batch request counters must be non-negative integers") + if self.completed_requests + self.failed_requests > self.total_requests: + raise ValueError( + "completed_requests + failed_requests must not exceed total_requests" + ) + if self.deadline_at_s is not None: + _require_finite_non_negative("deadline_at_s", self.deadline_at_s) + + @property + def remaining_requests(self) -> int: + """Requests not yet reported completed or failed.""" + + return self.total_requests - self.completed_requests - self.failed_requests + + +@dataclass +class PoolTrafficDemand: + """Online (non-batch) offered load for one inference pool. + + This is intentionally separate from Batch Gateway state. The Planner + derives safe capacity from its existing worker and capability inputs. + ``observed_at_s`` is an absolute wall-clock timestamp. + """ + + observed_at_s: float + pool_id: str + online_offered_rps: float + # Current frontend concurrency for the selected pool. ``None`` preserves + # compatibility with observation producers that only report offered load; + # ``0`` is an explicit, observed idle frontend. + frontend_active_requests: Optional[int] = None + + def __post_init__(self) -> None: + _require_finite_non_negative("observed_at_s", self.observed_at_s) + _require_non_empty_string("pool_id", self.pool_id) + _require_finite_non_negative("online_offered_rps", self.online_offered_rps) + if self.frontend_active_requests is not None and ( + isinstance(self.frontend_active_requests, bool) + or not isinstance(self.frontend_active_requests, int) + or self.frontend_active_requests < 0 + ): + raise ValueError( + "frontend_active_requests must be a non-negative integer or None" + ) + + +@dataclass +class BatchDispatcherFeedback: + """Observed dispatcher state for one inference pool. + + ``actual_dispatch_rps`` is the achieved batch admission rate over + ``observation_window_s``. ``applied_max_admission_rps=None`` means the + dispatcher did not report an applied Planner cap; ``0.0`` means an + explicit pause. ``observed_at_s`` is an absolute wall-clock timestamp. + """ + + observed_at_s: float + pool_id: str + observation_window_s: float + queued_requests: int + inflight_requests: int + actual_dispatch_rps: float + applied_max_admission_rps: Optional[float] = None + + def __post_init__(self) -> None: + _require_finite_non_negative("observed_at_s", self.observed_at_s) + _require_non_empty_string("pool_id", self.pool_id) + _require_positive_finite("observation_window_s", self.observation_window_s) + if any( + isinstance(count, bool) or not isinstance(count, int) or count < 0 + for count in (self.queued_requests, self.inflight_requests) + ): + raise ValueError("dispatcher request counts must be non-negative integers") + _require_finite_non_negative("actual_dispatch_rps", self.actual_dispatch_rps) + if self.applied_max_admission_rps is not None: + _require_finite_non_negative( + "applied_max_admission_rps", self.applied_max_admission_rps + ) + + +@dataclass +class BatchSchedulingObservation: + """Batch scheduling inputs sampled independently by their adapters. + + The durable job source is authoritative: a present observation means its + ``job_demands`` snapshot succeeded, including the valid empty-list case. + Traffic and dispatcher entries are independently optional per pool. An + absent pool entry means that source is unknown or unusable for this tick; + it must never be interpreted as a numeric zero. A known idle pool is + represented explicitly by ``PoolTrafficDemand.online_offered_rps == 0``; + frontend concurrency is independently unknown when + ``frontend_active_requests is None`` and explicitly idle when it is zero. + """ + + job_demands: list[BatchJobDemand] = field(default_factory=list) + pool_traffic: list[PoolTrafficDemand] = field(default_factory=list) + dispatcher_feedback: list[BatchDispatcherFeedback] = field(default_factory=list) + + @dataclass class TickInput: """What the adapter provides to the core on each tick. @@ -100,6 +239,7 @@ class TickInput: traffic: Optional[TrafficObservation] = None worker_counts: Optional[WorkerCounts] = None fpm_observations: Optional[FpmObservations] = None + batch: Optional[BatchSchedulingObservation] = None @dataclass @@ -112,6 +252,49 @@ class ScalingDecision: num_decode: Optional[int] = None +@dataclass +class BatchDrainLimitDecision: + """Leased batch admission limit for one inference pool. + + ``max_admission_rps`` is the maximum rate at which new batch requests + may be admitted; ``0.0`` explicitly pauses admission. ``valid_until_s`` + is an absolute wall-clock lease expiry. The dispatcher must stop using + the limit after expiry. ``decision_id`` identifies the decision for + idempotent application and audit correlation. + """ + + pool_id: str + max_admission_rps: float + valid_until_s: float + decision_id: str + + def __post_init__(self) -> None: + _require_non_empty_string("pool_id", self.pool_id) + _require_non_empty_string("decision_id", self.decision_id) + _require_finite_non_negative("max_admission_rps", self.max_admission_rps) + _require_finite_non_negative("valid_until_s", self.valid_until_s) + + +def _require_non_empty_string(name: str, value: object) -> None: + if not isinstance(value, str) or not value: + raise ValueError(f"{name} must be a non-empty string") + + +def _require_finite_non_negative(name: str, value: object) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{name} must be numeric") + normalized = float(value) + if not math.isfinite(normalized) or normalized < 0: + raise ValueError(f"{name} must be finite and non-negative") + return normalized + + +def _require_positive_finite(name: str, value: object) -> None: + normalized = _require_finite_non_negative(name, value) + if normalized <= 0: + raise ValueError(f"{name} must be positive and finite") + + @dataclass class TickDiagnostics: """Intermediate decision data populated by the planner core for @@ -201,6 +384,7 @@ class PlannerEffects: scale_to: Optional[ScalingDecision] = None next_tick: Optional[ScheduledTick] = None diagnostics: TickDiagnostics = field(default_factory=TickDiagnostics) + batch_drain_limits: list[BatchDrainLimitDecision] = field(default_factory=list) @dataclass diff --git a/components/src/dynamo/planner/plugins/_proto_bridge.py b/components/src/dynamo/planner/plugins/_proto_bridge.py index cad393c2af45..1361633958a3 100644 --- a/components/src/dynamo/planner/plugins/_proto_bridge.py +++ b/components/src/dynamo/planner/plugins/_proto_bridge.py @@ -57,6 +57,10 @@ pyd.TrafficMetrics: pb.TrafficMetrics, pyd.FpmData: pb.FpmData, pyd.WorkerState: pb.WorkerState, + pyd.BatchJobDemand: pb.BatchJobDemand, + pyd.PoolTrafficDemand: pb.PoolTrafficDemand, + pyd.BatchDispatcherFeedback: pb.BatchDispatcherFeedback, + pyd.BatchSchedulingData: pb.BatchSchedulingData, pyd.ObservationData: pb.ObservationData, pyd.PredictionData: pb.PredictionData, pyd.ComponentTarget: pb.ComponentTarget, diff --git a/components/src/dynamo/planner/plugins/orchestrator/engine_adapter.py b/components/src/dynamo/planner/plugins/orchestrator/engine_adapter.py index e4e995c19cf2..432b2ce91cc0 100644 --- a/components/src/dynamo/planner/plugins/orchestrator/engine_adapter.py +++ b/components/src/dynamo/planner/plugins/orchestrator/engine_adapter.py @@ -77,6 +77,7 @@ ) from dynamo.planner.core.state_machine import PlannerScalingState from dynamo.planner.core.types import ( + BatchSchedulingObservation, FpmObservations, PlannerEffects, ScalingDecision, @@ -106,15 +107,20 @@ from dynamo.planner.plugins.scheduler import PluginScheduler from dynamo.planner.plugins.transport.config import make_transport_for_endpoint from dynamo.planner.plugins.types import ( + BatchDispatcherFeedback as BatchDispatcherFeedbackData, +) +from dynamo.planner.plugins.types import BatchJobDemand as BatchJobDemandData +from dynamo.planner.plugins.types import ( + BatchSchedulingData, ComponentTarget, FpmData, ObservationData, OverrideResult, OverrideType, PipelineContext, - TrafficMetrics, - WorkerState, ) +from dynamo.planner.plugins.types import PoolTrafficDemand as PoolTrafficDemandData +from dynamo.planner.plugins.types import TrafficMetrics, WorkerState log = logging.getLogger(__name__) @@ -854,10 +860,62 @@ def _tick_input_to_context(self, ti: TickInput) -> PipelineContext: # and could not implement load-based decisions through # the public PipelineContext API. fpm = self._encode_fpm(ti.fpm_observations) + batch = self._encode_batch(ti.batch) return PipelineContext( request_id=f"tick-{ti.now_s}", decision_id=f"d-{ti.now_s}", - observations=ObservationData(traffic=traffic, fpm=fpm, workers=workers), + observations=ObservationData( + traffic=traffic, + fpm=fpm, + workers=workers, + batch=batch, + ), + ) + + @staticmethod + def _encode_batch( + obs: Optional[BatchSchedulingObservation], + ) -> Optional[BatchSchedulingData]: + """Project generic core batch observations onto the plugin contract.""" + + if obs is None: + return None + return BatchSchedulingData( + job_demands=[ + BatchJobDemandData( + observed_at_s=job.observed_at_s, + pool_id=job.pool_id, + job_id=job.job_id, + status=job.status, + total_requests=job.total_requests, + completed_requests=job.completed_requests, + failed_requests=job.failed_requests, + deadline_at_s=job.deadline_at_s, + work_class=job.work_class, + remaining_requests=job.remaining_requests, + ) + for job in obs.job_demands + ], + pool_traffic=[ + PoolTrafficDemandData( + observed_at_s=pool.observed_at_s, + pool_id=pool.pool_id, + online_offered_rps=pool.online_offered_rps, + ) + for pool in obs.pool_traffic + ], + dispatcher_feedback=[ + BatchDispatcherFeedbackData( + observed_at_s=feedback.observed_at_s, + pool_id=feedback.pool_id, + observation_window_s=feedback.observation_window_s, + queued_requests=feedback.queued_requests, + inflight_requests=feedback.inflight_requests, + actual_dispatch_rps=feedback.actual_dispatch_rps, + applied_max_admission_rps=feedback.applied_max_admission_rps, + ) + for feedback in obs.dispatcher_feedback + ], ) @staticmethod diff --git a/components/src/dynamo/planner/plugins/proto/v1/README.md b/components/src/dynamo/planner/plugins/proto/v1/README.md index 3e159af99cf5..d829c88737f6 100644 --- a/components/src/dynamo/planner/plugins/proto/v1/README.md +++ b/components/src/dynamo/planner/plugins/proto/v1/README.md @@ -1,3 +1,8 @@ + + # Plugin Proto v1 Plugin RPC contract for the Dynamo Planner plugin framework. @@ -38,7 +43,7 @@ Total: **6 services / 33 messages / 3 enums** ### Messages — by category - **PluginRegistry**: `RegisterRequest` / `RegisterResponse` / `HeartbeatRequest` / `HeartbeatResponse` / `UnregisterRequest` / `UnregisterResponse` / `ListPluginsRequest` / `ListPluginsResponse` / `PluginInfo` -- **PipelineContext + observation**: `PipelineContext` / `ObservationData` / `TrafficMetrics` / `FpmData` / `WorkerState` / `PredictionData` / `ScalingProposal` / `ComponentTarget` / `OverrideResult` / `AcceptResult` / `RejectResult` +- **PipelineContext + observation**: `PipelineContext` / `ObservationData` / `TrafficMetrics` / `FpmData` / `WorkerState` / `BatchSchedulingData` / `BatchJobDemand` / `PoolTrafficDemand` / `BatchDispatcherFeedback` / `PredictionData` / `ScalingProposal` / `ComponentTarget` / `OverrideResult` / `AcceptResult` / `RejectResult` - **Stage request/response**: `PredictStageRequest` / `PredictStageResponse` / `ProposeStageRequest` / `ProposeStageResponse` / `ProposeResult` / `ReconcileStageRequest` / `ReconcileStageResponse` / `ConstrainStageRequest` / `ConstrainStageResponse` - **PluginLifecycle**: `BootstrapRequest` / `BootstrapResponse` / `ResetRequest` / `ResetResponse` @@ -235,6 +240,15 @@ Plugins that declare `needs=["observations.fpm"]` receive the per-engine maps when available; absence of the field means no FPM data was collected for that tick. +## Batch scheduling observations + +`ObservationData.batch` carries flat, independently timestamped job demand, +online pool traffic, and dispatcher feedback. Plugins can declare +`needs=["observations.batch"]` and join records by `pool_id`. Job deadlines and +all `observed_at_s` values are absolute wall-clock timestamps. Dispatcher +`applied_max_admission_rps` preserves proto presence: unset means no applied cap +was reported, while `0.0` means admission was explicitly paused. + ## References - `dynamo/planner/plugins/types.py` — Pydantic v2 mirror diff --git a/components/src/dynamo/planner/plugins/proto/v1/plugin.proto b/components/src/dynamo/planner/plugins/proto/v1/plugin.proto index c9a9f4f39bd3..1f01e9aad3b0 100644 --- a/components/src/dynamo/planner/plugins/proto/v1/plugin.proto +++ b/components/src/dynamo/planner/plugins/proto/v1/plugin.proto @@ -234,6 +234,63 @@ message ObservationData optional TrafficMetrics traffic = 1; optional FpmData fpm = 2; optional WorkerState workers = 3; + optional BatchSchedulingData batch = 4; +} + +// Mirrors BatchSchedulingObservation (core/types.py). Each repeated input is +// timestamped independently because Batch Gateway, traffic, and dispatcher +// observations may be sampled at different times. Once this message is +// present, an absent pool_traffic or dispatcher_feedback entry means that +// source is unknown/unusable for the pool on this tick; absence must never be +// interpreted as numeric zero. Known zero traffic is carried explicitly in a +// PoolTrafficDemand value. +message BatchSchedulingData +{ + repeated BatchJobDemand job_demands = 1; + repeated PoolTrafficDemand pool_traffic = 2; + repeated BatchDispatcherFeedback dispatcher_feedback = 3; +} + +// Mirrors BatchJobDemand (core/types.py). Raw counters are retained for audit; +// remaining_requests is derived and validated by the core adapter. +message BatchJobDemand +{ + double observed_at_s = 1; // absolute wall-clock timestamp + string pool_id = 2; + string job_id = 3; + string status = 4; + int64 total_requests = 5; + int64 completed_requests = 6; + int64 failed_requests = 7; + optional double deadline_at_s = 8; // absolute wall-clock deadline; unset = none + string work_class = 9; // opaque request/workload class + int64 remaining_requests = 10; +} + +// Online (non-batch) offered load for a pool. This is deliberately separate +// from Batch Gateway state; safe capacity remains derived from existing +// Planner worker/capability inputs. +message PoolTrafficDemand +{ + double observed_at_s = 1; // absolute wall-clock timestamp + string pool_id = 2; + double online_offered_rps = 3; + // Unset means the observation source did not report frontend concurrency; + // 0 is an explicitly observed idle frontend. + optional int64 frontend_active_requests = 4; +} + +// Mirrors BatchDispatcherFeedback (core/types.py). +message BatchDispatcherFeedback +{ + double observed_at_s = 1; // absolute wall-clock timestamp + string pool_id = 2; + double observation_window_s = 3; + int64 queued_requests = 4; + int64 inflight_requests = 5; + double actual_dispatch_rps = 6; // achieved batch admissions over the window + // Unset means no applied Planner cap was reported; 0.0 is an explicit pause. + optional double applied_max_admission_rps = 7; } // Mirrors TrafficObservation (types.py) diff --git a/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.py b/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.py index c9709331726f..8f8cbe318e93 100644 --- a/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.py +++ b/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.py @@ -2,12 +2,23 @@ # SPDX-License-Identifier: Apache-2.0 # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE # source: dynamo/planner/plugins/proto/v1/plugin.proto +# Protobuf Python Version: 5.27.2 """Generated protocol buffer code.""" -from google.protobuf.internal import builder as _builder from google.protobuf import descriptor as _descriptor from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 5, + 27, + 2, + '', + 'dynamo/planner/plugins/proto/v1/plugin.proto' +) # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() @@ -15,107 +26,115 @@ -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n,dynamo/planner/plugins/proto/v1/plugin.proto\x12\x18\x64ynamo.planner.plugin.v1\"\xdc\x02\n\x0fRegisterRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x13\n\x0bplugin_type\x18\x02 \x01(\t\x12\x10\n\x08priority\x18\x03 \x01(\r\x12\x10\n\x08\x65ndpoint\x18\x04 \x01(\t\x12\x0f\n\x07version\x18\x05 \x01(\t\x12\"\n\x1a\x65xecution_interval_seconds\x18\x06 \x01(\x02\x12\x39\n\x0bhold_policy\x18\x07 \x01(\x0e\x32$.dynamo.planner.plugin.v1.HoldPolicy\x12\r\n\x05needs\x18\x08 \x03(\t\x12\x18\n\x10protocol_version\x18\t \x01(\t\x12\x12\n\nauth_token\x18\n \x01(\t\x12 \n\x18requires_produced_fields\x18\r \x03(\t\x12\"\n\x1aobservation_window_seconds\x18\x0e \x01(\x02J\x04\x08\x0b\x10\x0cJ\x04\x08\x0c\x10\r\"`\n\x10RegisterResponse\x12\x10\n\x08\x61\x63\x63\x65pted\x18\x01 \x01(\x08\x12\x15\n\rreject_reason\x18\x02 \x01(\t\x12#\n\x1bnegotiated_protocol_version\x18\x03 \x01(\t\"9\n\x10HeartbeatRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x12\n\nauth_token\x18\x02 \x01(\t\"\x1f\n\x11HeartbeatResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\"J\n\x11UnregisterRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\x12\n\nauth_token\x18\x03 \x01(\t\" \n\x12UnregisterResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\"D\n\x12ListPluginsRequest\x12\x14\n\x0cstage_filter\x18\x01 \x01(\t\x12\x18\n\x10include_disabled\x18\x02 \x01(\x08\"L\n\x13ListPluginsResponse\x12\x35\n\x07plugins\x18\x01 \x03(\x0b\x32$.dynamo.planner.plugin.v1.PluginInfo\"\xc0\x02\n\nPluginInfo\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x13\n\x0bplugin_type\x18\x02 \x01(\t\x12\x10\n\x08priority\x18\x03 \x01(\r\x12\x0f\n\x07version\x18\x04 \x01(\t\x12\x18\n\x10protocol_version\x18\x05 \x01(\t\x12\x0f\n\x07\x65nabled\x18\x06 \x01(\x08\x12\x12\n\nis_builtin\x18\x07 \x01(\x08\x12\x11\n\ttransport\x18\x08 \x01(\t\x12=\n\rcircuit_state\x18\t \x01(\x0e\x32&.dynamo.planner.plugin.v1.CircuitState\x12\x19\n\x11\x65valuations_total\x18\n \x01(\x04\x12 \n\x18last_call_at_seconds_ago\x18\x0b \x01(\x01\x12\x19\n\x11\x63\x61\x63he_age_seconds\x18\x0c \x01(\x01\"\x89\x03\n\x0fPipelineContext\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65\x63ision_id\x18\x02 \x01(\t\x12\x44\n\x0cobservations\x18\x03 \x01(\x0b\x32).dynamo.planner.plugin.v1.ObservationDataH\x00\x88\x01\x01\x12\x42\n\x0bpredictions\x18\x04 \x01(\x0b\x32(.dynamo.planner.plugin.v1.PredictionDataH\x01\x88\x01\x01\x12@\n\x08proposal\x18\x05 \x01(\x0b\x32).dynamo.planner.plugin.v1.ScalingProposalH\x02\x88\x01\x01\x12\x43\n\x0b\x63onstrained\x18\x06 \x01(\x0b\x32).dynamo.planner.plugin.v1.ScalingProposalH\x03\x88\x01\x01\x42\x0f\n\r_observationsB\x0e\n\x0c_predictionsB\x0b\n\t_proposalB\x0e\n\x0c_constrained\"\xe3\x01\n\x0fObservationData\x12>\n\x07traffic\x18\x01 \x01(\x0b\x32(.dynamo.planner.plugin.v1.TrafficMetricsH\x00\x88\x01\x01\x12\x33\n\x03\x66pm\x18\x02 \x01(\x0b\x32!.dynamo.planner.plugin.v1.FpmDataH\x01\x88\x01\x01\x12;\n\x07workers\x18\x03 \x01(\x0b\x32%.dynamo.planner.plugin.v1.WorkerStateH\x02\x88\x01\x01\x42\n\n\x08_trafficB\x06\n\x04_fpmB\n\n\x08_workers\"\xa7\x01\n\x0eTrafficMetrics\x12\x12\n\nduration_s\x18\x01 \x01(\x01\x12\x0f\n\x07num_req\x18\x02 \x01(\x01\x12\x0b\n\x03isl\x18\x03 \x01(\x01\x12\x0b\n\x03osl\x18\x04 \x01(\x01\x12\x18\n\x0bkv_hit_rate\x18\x05 \x01(\x01H\x00\x88\x01\x01\x12\x1a\n\raccept_length\x18\x06 \x01(\x01H\x01\x88\x01\x01\x42\x0e\n\x0c_kv_hit_rateB\x10\n\x0e_accept_length\"\x94\x02\n\x07\x46pmData\x12N\n\x0fprefill_engines\x18\x01 \x03(\x0b\x32\x35.dynamo.planner.plugin.v1.FpmData.PrefillEnginesEntry\x12L\n\x0e\x64\x65\x63ode_engines\x18\x02 \x03(\x0b\x32\x34.dynamo.planner.plugin.v1.FpmData.DecodeEnginesEntry\x1a\x35\n\x13PrefillEnginesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c:\x02\x38\x01\x1a\x34\n\x12\x44\x65\x63odeEnginesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c:\x02\x38\x01\"\xdf\x02\n\x0bWorkerState\x12\x1a\n\rready_prefill\x18\x01 \x01(\x05H\x00\x88\x01\x01\x12\x19\n\x0cready_decode\x18\x02 \x01(\x05H\x01\x88\x01\x01\x12\x1d\n\x10\x65xpected_prefill\x18\x03 \x01(\x05H\x02\x88\x01\x01\x12\x1c\n\x0f\x65xpected_decode\x18\x04 \x01(\x05H\x03\x88\x01\x01\x12(\n\x1bprefill_scaling_in_progress\x18\x05 \x01(\x08H\x04\x88\x01\x01\x12\'\n\x1a\x64\x65\x63ode_scaling_in_progress\x18\x06 \x01(\x08H\x05\x88\x01\x01\x42\x10\n\x0e_ready_prefillB\x0f\n\r_ready_decodeB\x13\n\x11_expected_prefillB\x12\n\x10_expected_decodeB\x1e\n\x1c_prefill_scaling_in_progressB\x1d\n\x1b_decode_scaling_in_progress\"\xb2\x02\n\x0ePredictionData\x12\x1e\n\x11predicted_num_req\x18\x01 \x01(\x01H\x00\x88\x01\x01\x12\x1a\n\rpredicted_isl\x18\x02 \x01(\x01H\x01\x88\x01\x01\x12\x1a\n\rpredicted_osl\x18\x03 \x01(\x01H\x02\x88\x01\x01\x12\x0e\n\x06source\x18\x04 \x01(\t\x12\"\n\x15predicted_kv_hit_rate\x18\x05 \x01(\x01H\x03\x88\x01\x01\x12$\n\x17predicted_accept_length\x18\x06 \x01(\x01H\x04\x88\x01\x01\x42\x14\n\x12_predicted_num_reqB\x10\n\x0e_predicted_islB\x10\n\x0e_predicted_oslB\x18\n\x16_predicted_kv_hit_rateB\x1a\n\x18_predicted_accept_length\"m\n\x0fScalingProposal\x12:\n\x07targets\x18\x01 \x03(\x0b\x32).dynamo.planner.plugin.v1.ComponentTarget\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\x0e\n\x06source\x18\x03 \x01(\t\"\x8d\x01\n\x0f\x43omponentTarget\x12\x1a\n\x12sub_component_type\x18\x01 \x01(\t\x12\x15\n\x08replicas\x18\x03 \x01(\x05H\x00\x88\x01\x01\x12\x34\n\x04type\x18\x04 \x01(\x0e\x32&.dynamo.planner.plugin.v1.OverrideTypeB\x0b\n\t_replicasJ\x04\x08\x02\x10\x03\"\\\n\x0eOverrideResult\x12:\n\x07targets\x18\x01 \x03(\x0b\x32).dynamo.planner.plugin.v1.ComponentTarget\x12\x0e\n\x06reason\x18\x02 \x01(\t\"\x0e\n\x0c\x41\x63\x63\x65ptResult\"\x1e\n\x0cRejectResult\x12\x0e\n\x06reason\x18\x01 \x01(\t\"Q\n\x13PredictStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"t\n\x14PredictStageResponse\x12=\n\x0bpredictions\x18\x01 \x01(\x0b\x32(.dynamo.planner.plugin.v1.PredictionData\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\r\n\x05\x66inal\x18\x03 \x01(\x08\"Q\n\x13ProposeStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"\xe1\x01\n\x14ProposeStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"\x8f\x01\n\x15ReconcileStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\x12:\n\tproposals\x18\x02 \x03(\x0b\x32\'.dynamo.planner.plugin.v1.ProposeResult\"\xf0\x01\n\rProposeResult\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x02 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x03 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x04 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\x10\n\x08priority\x18\x05 \x01(\rB\x08\n\x06result\"\xe3\x01\n\x16ReconcileStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"S\n\x15\x43onstrainStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"\xe3\x01\n\x16\x43onstrainStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"\x9e\x01\n\x10\x42ootstrapRequest\x12\x16\n\x0e\x62ootstrap_data\x18\x01 \x01(\x0c\x12\x44\n\x05hints\x18\x02 \x03(\x0b\x32\x35.dynamo.planner.plugin.v1.BootstrapRequest.HintsEntry\x1a,\n\nHintsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"0\n\x11\x42ootstrapResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\"\x1e\n\x0cResetRequest\x12\x0e\n\x06reason\x18\x01 \x01(\t\",\n\rResetResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t*1\n\nHoldPolicy\x12\x14\n\x10\x41\x43\x43\x45PT_WHEN_IDLE\x10\x00\x12\r\n\tHOLD_LAST\x10\x01*3\n\x0c\x43ircuitState\x12\n\n\x06\x43LOSED\x10\x00\x12\x08\n\x04OPEN\x10\x01\x12\r\n\tHALF_OPEN\x10\x02*2\n\x0cOverrideType\x12\x07\n\x03SET\x10\x00\x12\x0c\n\x08\x41T_LEAST\x10\x01\x12\x0b\n\x07\x41T_MOST\x10\x02\x32\xae\x03\n\x0ePluginRegistry\x12\x61\n\x08Register\x12).dynamo.planner.plugin.v1.RegisterRequest\x1a*.dynamo.planner.plugin.v1.RegisterResponse\x12\x64\n\tHeartbeat\x12*.dynamo.planner.plugin.v1.HeartbeatRequest\x1a+.dynamo.planner.plugin.v1.HeartbeatResponse\x12g\n\nUnregister\x12+.dynamo.planner.plugin.v1.UnregisterRequest\x1a,.dynamo.planner.plugin.v1.UnregisterResponse\x12j\n\x0bListPlugins\x12,.dynamo.planner.plugin.v1.ListPluginsRequest\x1a-.dynamo.planner.plugin.v1.ListPluginsResponse2y\n\rPredictPlugin\x12h\n\x07Predict\x12-.dynamo.planner.plugin.v1.PredictStageRequest\x1a..dynamo.planner.plugin.v1.PredictStageResponse2y\n\rProposePlugin\x12h\n\x07Propose\x12-.dynamo.planner.plugin.v1.ProposeStageRequest\x1a..dynamo.planner.plugin.v1.ProposeStageResponse2\x81\x01\n\x0fReconcilePlugin\x12n\n\tReconcile\x12/.dynamo.planner.plugin.v1.ReconcileStageRequest\x1a\x30.dynamo.planner.plugin.v1.ReconcileStageResponse2\x81\x01\n\x0f\x43onstrainPlugin\x12n\n\tConstrain\x12/.dynamo.planner.plugin.v1.ConstrainStageRequest\x1a\x30.dynamo.planner.plugin.v1.ConstrainStageResponse2\xd1\x01\n\x0fPluginLifecycle\x12\x64\n\tBootstrap\x12*.dynamo.planner.plugin.v1.BootstrapRequest\x1a+.dynamo.planner.plugin.v1.BootstrapResponse\x12X\n\x05Reset\x12&.dynamo.planner.plugin.v1.ResetRequest\x1a\'.dynamo.planner.plugin.v1.ResetResponseb\x06proto3') - -_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals()) -_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dynamo.planner.plugins.proto.v1.plugin_pb2', globals()) -if _descriptor._USE_C_DESCRIPTORS == False: +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n,dynamo/planner/plugins/proto/v1/plugin.proto\x12\x18\x64ynamo.planner.plugin.v1\"\xdc\x02\n\x0fRegisterRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x13\n\x0bplugin_type\x18\x02 \x01(\t\x12\x10\n\x08priority\x18\x03 \x01(\r\x12\x10\n\x08\x65ndpoint\x18\x04 \x01(\t\x12\x0f\n\x07version\x18\x05 \x01(\t\x12\"\n\x1a\x65xecution_interval_seconds\x18\x06 \x01(\x02\x12\x39\n\x0bhold_policy\x18\x07 \x01(\x0e\x32$.dynamo.planner.plugin.v1.HoldPolicy\x12\r\n\x05needs\x18\x08 \x03(\t\x12\x18\n\x10protocol_version\x18\t \x01(\t\x12\x12\n\nauth_token\x18\n \x01(\t\x12 \n\x18requires_produced_fields\x18\r \x03(\t\x12\"\n\x1aobservation_window_seconds\x18\x0e \x01(\x02J\x04\x08\x0b\x10\x0cJ\x04\x08\x0c\x10\r\"`\n\x10RegisterResponse\x12\x10\n\x08\x61\x63\x63\x65pted\x18\x01 \x01(\x08\x12\x15\n\rreject_reason\x18\x02 \x01(\t\x12#\n\x1bnegotiated_protocol_version\x18\x03 \x01(\t\"9\n\x10HeartbeatRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x12\n\nauth_token\x18\x02 \x01(\t\"\x1f\n\x11HeartbeatResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\"J\n\x11UnregisterRequest\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\x12\n\nauth_token\x18\x03 \x01(\t\" \n\x12UnregisterResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\"D\n\x12ListPluginsRequest\x12\x14\n\x0cstage_filter\x18\x01 \x01(\t\x12\x18\n\x10include_disabled\x18\x02 \x01(\x08\"L\n\x13ListPluginsResponse\x12\x35\n\x07plugins\x18\x01 \x03(\x0b\x32$.dynamo.planner.plugin.v1.PluginInfo\"\xc0\x02\n\nPluginInfo\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x13\n\x0bplugin_type\x18\x02 \x01(\t\x12\x10\n\x08priority\x18\x03 \x01(\r\x12\x0f\n\x07version\x18\x04 \x01(\t\x12\x18\n\x10protocol_version\x18\x05 \x01(\t\x12\x0f\n\x07\x65nabled\x18\x06 \x01(\x08\x12\x12\n\nis_builtin\x18\x07 \x01(\x08\x12\x11\n\ttransport\x18\x08 \x01(\t\x12=\n\rcircuit_state\x18\t \x01(\x0e\x32&.dynamo.planner.plugin.v1.CircuitState\x12\x19\n\x11\x65valuations_total\x18\n \x01(\x04\x12 \n\x18last_call_at_seconds_ago\x18\x0b \x01(\x01\x12\x19\n\x11\x63\x61\x63he_age_seconds\x18\x0c \x01(\x01\"\x89\x03\n\x0fPipelineContext\x12\x12\n\nrequest_id\x18\x01 \x01(\t\x12\x13\n\x0b\x64\x65\x63ision_id\x18\x02 \x01(\t\x12\x44\n\x0cobservations\x18\x03 \x01(\x0b\x32).dynamo.planner.plugin.v1.ObservationDataH\x00\x88\x01\x01\x12\x42\n\x0bpredictions\x18\x04 \x01(\x0b\x32(.dynamo.planner.plugin.v1.PredictionDataH\x01\x88\x01\x01\x12@\n\x08proposal\x18\x05 \x01(\x0b\x32).dynamo.planner.plugin.v1.ScalingProposalH\x02\x88\x01\x01\x12\x43\n\x0b\x63onstrained\x18\x06 \x01(\x0b\x32).dynamo.planner.plugin.v1.ScalingProposalH\x03\x88\x01\x01\x42\x0f\n\r_observationsB\x0e\n\x0c_predictionsB\x0b\n\t_proposalB\x0e\n\x0c_constrained\"\xb0\x02\n\x0fObservationData\x12>\n\x07traffic\x18\x01 \x01(\x0b\x32(.dynamo.planner.plugin.v1.TrafficMetricsH\x00\x88\x01\x01\x12\x33\n\x03\x66pm\x18\x02 \x01(\x0b\x32!.dynamo.planner.plugin.v1.FpmDataH\x01\x88\x01\x01\x12;\n\x07workers\x18\x03 \x01(\x0b\x32%.dynamo.planner.plugin.v1.WorkerStateH\x02\x88\x01\x01\x12\x41\n\x05\x62\x61tch\x18\x04 \x01(\x0b\x32-.dynamo.planner.plugin.v1.BatchSchedulingDataH\x03\x88\x01\x01\x42\n\n\x08_trafficB\x06\n\x04_fpmB\n\n\x08_workersB\x08\n\x06_batch\"\xe7\x01\n\x13\x42\x61tchSchedulingData\x12=\n\x0bjob_demands\x18\x01 \x03(\x0b\x32(.dynamo.planner.plugin.v1.BatchJobDemand\x12\x41\n\x0cpool_traffic\x18\x02 \x03(\x0b\x32+.dynamo.planner.plugin.v1.PoolTrafficDemand\x12N\n\x13\x64ispatcher_feedback\x18\x03 \x03(\x0b\x32\x31.dynamo.planner.plugin.v1.BatchDispatcherFeedback\"\x83\x02\n\x0e\x42\x61tchJobDemand\x12\x15\n\robserved_at_s\x18\x01 \x01(\x01\x12\x0f\n\x07pool_id\x18\x02 \x01(\t\x12\x0e\n\x06job_id\x18\x03 \x01(\t\x12\x0e\n\x06status\x18\x04 \x01(\t\x12\x16\n\x0etotal_requests\x18\x05 \x01(\x03\x12\x1a\n\x12\x63ompleted_requests\x18\x06 \x01(\x03\x12\x17\n\x0f\x66\x61iled_requests\x18\x07 \x01(\x03\x12\x1a\n\rdeadline_at_s\x18\x08 \x01(\x01H\x00\x88\x01\x01\x12\x12\n\nwork_class\x18\t \x01(\t\x12\x1a\n\x12remaining_requests\x18\n \x01(\x03\x42\x10\n\x0e_deadline_at_s\"\x9b\x01\n\x11PoolTrafficDemand\x12\x15\n\robserved_at_s\x18\x01 \x01(\x01\x12\x0f\n\x07pool_id\x18\x02 \x01(\t\x12\x1a\n\x12online_offered_rps\x18\x03 \x01(\x01\x12%\n\x18\x66rontend_active_requests\x18\x04 \x01(\x03H\x00\x88\x01\x01\x42\x1b\n\x19_frontend_active_requests\"\xf6\x01\n\x17\x42\x61tchDispatcherFeedback\x12\x15\n\robserved_at_s\x18\x01 \x01(\x01\x12\x0f\n\x07pool_id\x18\x02 \x01(\t\x12\x1c\n\x14observation_window_s\x18\x03 \x01(\x01\x12\x17\n\x0fqueued_requests\x18\x04 \x01(\x03\x12\x19\n\x11inflight_requests\x18\x05 \x01(\x03\x12\x1b\n\x13\x61\x63tual_dispatch_rps\x18\x06 \x01(\x01\x12&\n\x19\x61pplied_max_admission_rps\x18\x07 \x01(\x01H\x00\x88\x01\x01\x42\x1c\n\x1a_applied_max_admission_rps\"\xa7\x01\n\x0eTrafficMetrics\x12\x12\n\nduration_s\x18\x01 \x01(\x01\x12\x0f\n\x07num_req\x18\x02 \x01(\x01\x12\x0b\n\x03isl\x18\x03 \x01(\x01\x12\x0b\n\x03osl\x18\x04 \x01(\x01\x12\x18\n\x0bkv_hit_rate\x18\x05 \x01(\x01H\x00\x88\x01\x01\x12\x1a\n\raccept_length\x18\x06 \x01(\x01H\x01\x88\x01\x01\x42\x0e\n\x0c_kv_hit_rateB\x10\n\x0e_accept_length\"\x94\x02\n\x07\x46pmData\x12N\n\x0fprefill_engines\x18\x01 \x03(\x0b\x32\x35.dynamo.planner.plugin.v1.FpmData.PrefillEnginesEntry\x12L\n\x0e\x64\x65\x63ode_engines\x18\x02 \x03(\x0b\x32\x34.dynamo.planner.plugin.v1.FpmData.DecodeEnginesEntry\x1a\x35\n\x13PrefillEnginesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c:\x02\x38\x01\x1a\x34\n\x12\x44\x65\x63odeEnginesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\x0c:\x02\x38\x01\"\xdf\x02\n\x0bWorkerState\x12\x1a\n\rready_prefill\x18\x01 \x01(\x05H\x00\x88\x01\x01\x12\x19\n\x0cready_decode\x18\x02 \x01(\x05H\x01\x88\x01\x01\x12\x1d\n\x10\x65xpected_prefill\x18\x03 \x01(\x05H\x02\x88\x01\x01\x12\x1c\n\x0f\x65xpected_decode\x18\x04 \x01(\x05H\x03\x88\x01\x01\x12(\n\x1bprefill_scaling_in_progress\x18\x05 \x01(\x08H\x04\x88\x01\x01\x12\'\n\x1a\x64\x65\x63ode_scaling_in_progress\x18\x06 \x01(\x08H\x05\x88\x01\x01\x42\x10\n\x0e_ready_prefillB\x0f\n\r_ready_decodeB\x13\n\x11_expected_prefillB\x12\n\x10_expected_decodeB\x1e\n\x1c_prefill_scaling_in_progressB\x1d\n\x1b_decode_scaling_in_progress\"\xb2\x02\n\x0ePredictionData\x12\x1e\n\x11predicted_num_req\x18\x01 \x01(\x01H\x00\x88\x01\x01\x12\x1a\n\rpredicted_isl\x18\x02 \x01(\x01H\x01\x88\x01\x01\x12\x1a\n\rpredicted_osl\x18\x03 \x01(\x01H\x02\x88\x01\x01\x12\x0e\n\x06source\x18\x04 \x01(\t\x12\"\n\x15predicted_kv_hit_rate\x18\x05 \x01(\x01H\x03\x88\x01\x01\x12$\n\x17predicted_accept_length\x18\x06 \x01(\x01H\x04\x88\x01\x01\x42\x14\n\x12_predicted_num_reqB\x10\n\x0e_predicted_islB\x10\n\x0e_predicted_oslB\x18\n\x16_predicted_kv_hit_rateB\x1a\n\x18_predicted_accept_length\"m\n\x0fScalingProposal\x12:\n\x07targets\x18\x01 \x03(\x0b\x32).dynamo.planner.plugin.v1.ComponentTarget\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\x0e\n\x06source\x18\x03 \x01(\t\"\x8d\x01\n\x0f\x43omponentTarget\x12\x1a\n\x12sub_component_type\x18\x01 \x01(\t\x12\x15\n\x08replicas\x18\x03 \x01(\x05H\x00\x88\x01\x01\x12\x34\n\x04type\x18\x04 \x01(\x0e\x32&.dynamo.planner.plugin.v1.OverrideTypeB\x0b\n\t_replicasJ\x04\x08\x02\x10\x03\"\\\n\x0eOverrideResult\x12:\n\x07targets\x18\x01 \x03(\x0b\x32).dynamo.planner.plugin.v1.ComponentTarget\x12\x0e\n\x06reason\x18\x02 \x01(\t\"\x0e\n\x0c\x41\x63\x63\x65ptResult\"\x1e\n\x0cRejectResult\x12\x0e\n\x06reason\x18\x01 \x01(\t\"Q\n\x13PredictStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"t\n\x14PredictStageResponse\x12=\n\x0bpredictions\x18\x01 \x01(\x0b\x32(.dynamo.planner.plugin.v1.PredictionData\x12\x0e\n\x06reason\x18\x02 \x01(\t\x12\r\n\x05\x66inal\x18\x03 \x01(\x08\"Q\n\x13ProposeStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"\xe1\x01\n\x14ProposeStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"\x8f\x01\n\x15ReconcileStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\x12:\n\tproposals\x18\x02 \x03(\x0b\x32\'.dynamo.planner.plugin.v1.ProposeResult\"\xf0\x01\n\rProposeResult\x12\x11\n\tplugin_id\x18\x01 \x01(\t\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x02 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x03 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x04 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\x10\n\x08priority\x18\x05 \x01(\rB\x08\n\x06result\"\xe3\x01\n\x16ReconcileStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"S\n\x15\x43onstrainStageRequest\x12:\n\x07\x63ontext\x18\x01 \x01(\x0b\x32).dynamo.planner.plugin.v1.PipelineContext\"\xe3\x01\n\x16\x43onstrainStageResponse\x12\x38\n\x06\x61\x63\x63\x65pt\x18\x01 \x01(\x0b\x32&.dynamo.planner.plugin.v1.AcceptResultH\x00\x12<\n\x08override\x18\x02 \x01(\x0b\x32(.dynamo.planner.plugin.v1.OverrideResultH\x00\x12\x38\n\x06reject\x18\x03 \x01(\x0b\x32&.dynamo.planner.plugin.v1.RejectResultH\x00\x12\r\n\x05\x66inal\x18\x04 \x01(\x08\x42\x08\n\x06result\"\x9e\x01\n\x10\x42ootstrapRequest\x12\x16\n\x0e\x62ootstrap_data\x18\x01 \x01(\x0c\x12\x44\n\x05hints\x18\x02 \x03(\x0b\x32\x35.dynamo.planner.plugin.v1.BootstrapRequest.HintsEntry\x1a,\n\nHintsEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01\"0\n\x11\x42ootstrapResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t\"\x1e\n\x0cResetRequest\x12\x0e\n\x06reason\x18\x01 \x01(\t\",\n\rResetResponse\x12\n\n\x02ok\x18\x01 \x01(\x08\x12\x0f\n\x07message\x18\x02 \x01(\t*1\n\nHoldPolicy\x12\x14\n\x10\x41\x43\x43\x45PT_WHEN_IDLE\x10\x00\x12\r\n\tHOLD_LAST\x10\x01*3\n\x0c\x43ircuitState\x12\n\n\x06\x43LOSED\x10\x00\x12\x08\n\x04OPEN\x10\x01\x12\r\n\tHALF_OPEN\x10\x02*2\n\x0cOverrideType\x12\x07\n\x03SET\x10\x00\x12\x0c\n\x08\x41T_LEAST\x10\x01\x12\x0b\n\x07\x41T_MOST\x10\x02\x32\xae\x03\n\x0ePluginRegistry\x12\x61\n\x08Register\x12).dynamo.planner.plugin.v1.RegisterRequest\x1a*.dynamo.planner.plugin.v1.RegisterResponse\x12\x64\n\tHeartbeat\x12*.dynamo.planner.plugin.v1.HeartbeatRequest\x1a+.dynamo.planner.plugin.v1.HeartbeatResponse\x12g\n\nUnregister\x12+.dynamo.planner.plugin.v1.UnregisterRequest\x1a,.dynamo.planner.plugin.v1.UnregisterResponse\x12j\n\x0bListPlugins\x12,.dynamo.planner.plugin.v1.ListPluginsRequest\x1a-.dynamo.planner.plugin.v1.ListPluginsResponse2y\n\rPredictPlugin\x12h\n\x07Predict\x12-.dynamo.planner.plugin.v1.PredictStageRequest\x1a..dynamo.planner.plugin.v1.PredictStageResponse2y\n\rProposePlugin\x12h\n\x07Propose\x12-.dynamo.planner.plugin.v1.ProposeStageRequest\x1a..dynamo.planner.plugin.v1.ProposeStageResponse2\x81\x01\n\x0fReconcilePlugin\x12n\n\tReconcile\x12/.dynamo.planner.plugin.v1.ReconcileStageRequest\x1a\x30.dynamo.planner.plugin.v1.ReconcileStageResponse2\x81\x01\n\x0f\x43onstrainPlugin\x12n\n\tConstrain\x12/.dynamo.planner.plugin.v1.ConstrainStageRequest\x1a\x30.dynamo.planner.plugin.v1.ConstrainStageResponse2\xd1\x01\n\x0fPluginLifecycle\x12\x64\n\tBootstrap\x12*.dynamo.planner.plugin.v1.BootstrapRequest\x1a+.dynamo.planner.plugin.v1.BootstrapResponse\x12X\n\x05Reset\x12&.dynamo.planner.plugin.v1.ResetRequest\x1a\'.dynamo.planner.plugin.v1.ResetResponseb\x06proto3') - DESCRIPTOR._options = None - _FPMDATA_PREFILLENGINESENTRY._options = None - _FPMDATA_PREFILLENGINESENTRY._serialized_options = b'8\001' - _FPMDATA_DECODEENGINESENTRY._options = None - _FPMDATA_DECODEENGINESENTRY._serialized_options = b'8\001' - _BOOTSTRAPREQUEST_HINTSENTRY._options = None - _BOOTSTRAPREQUEST_HINTSENTRY._serialized_options = b'8\001' - _HOLDPOLICY._serialized_start=5066 - _HOLDPOLICY._serialized_end=5115 - _CIRCUITSTATE._serialized_start=5117 - _CIRCUITSTATE._serialized_end=5168 - _OVERRIDETYPE._serialized_start=5170 - _OVERRIDETYPE._serialized_end=5220 - _REGISTERREQUEST._serialized_start=75 - _REGISTERREQUEST._serialized_end=423 - _REGISTERRESPONSE._serialized_start=425 - _REGISTERRESPONSE._serialized_end=521 - _HEARTBEATREQUEST._serialized_start=523 - _HEARTBEATREQUEST._serialized_end=580 - _HEARTBEATRESPONSE._serialized_start=582 - _HEARTBEATRESPONSE._serialized_end=613 - _UNREGISTERREQUEST._serialized_start=615 - _UNREGISTERREQUEST._serialized_end=689 - _UNREGISTERRESPONSE._serialized_start=691 - _UNREGISTERRESPONSE._serialized_end=723 - _LISTPLUGINSREQUEST._serialized_start=725 - _LISTPLUGINSREQUEST._serialized_end=793 - _LISTPLUGINSRESPONSE._serialized_start=795 - _LISTPLUGINSRESPONSE._serialized_end=871 - _PLUGININFO._serialized_start=874 - _PLUGININFO._serialized_end=1194 - _PIPELINECONTEXT._serialized_start=1197 - _PIPELINECONTEXT._serialized_end=1590 - _OBSERVATIONDATA._serialized_start=1593 - _OBSERVATIONDATA._serialized_end=1820 - _TRAFFICMETRICS._serialized_start=1823 - _TRAFFICMETRICS._serialized_end=1990 - _FPMDATA._serialized_start=1993 - _FPMDATA._serialized_end=2269 - _FPMDATA_PREFILLENGINESENTRY._serialized_start=2162 - _FPMDATA_PREFILLENGINESENTRY._serialized_end=2215 - _FPMDATA_DECODEENGINESENTRY._serialized_start=2217 - _FPMDATA_DECODEENGINESENTRY._serialized_end=2269 - _WORKERSTATE._serialized_start=2272 - _WORKERSTATE._serialized_end=2623 - _PREDICTIONDATA._serialized_start=2626 - _PREDICTIONDATA._serialized_end=2932 - _SCALINGPROPOSAL._serialized_start=2934 - _SCALINGPROPOSAL._serialized_end=3043 - _COMPONENTTARGET._serialized_start=3046 - _COMPONENTTARGET._serialized_end=3187 - _OVERRIDERESULT._serialized_start=3189 - _OVERRIDERESULT._serialized_end=3281 - _ACCEPTRESULT._serialized_start=3283 - _ACCEPTRESULT._serialized_end=3297 - _REJECTRESULT._serialized_start=3299 - _REJECTRESULT._serialized_end=3329 - _PREDICTSTAGEREQUEST._serialized_start=3331 - _PREDICTSTAGEREQUEST._serialized_end=3412 - _PREDICTSTAGERESPONSE._serialized_start=3414 - _PREDICTSTAGERESPONSE._serialized_end=3530 - _PROPOSESTAGEREQUEST._serialized_start=3532 - _PROPOSESTAGEREQUEST._serialized_end=3613 - _PROPOSESTAGERESPONSE._serialized_start=3616 - _PROPOSESTAGERESPONSE._serialized_end=3841 - _RECONCILESTAGEREQUEST._serialized_start=3844 - _RECONCILESTAGEREQUEST._serialized_end=3987 - _PROPOSERESULT._serialized_start=3990 - _PROPOSERESULT._serialized_end=4230 - _RECONCILESTAGERESPONSE._serialized_start=4233 - _RECONCILESTAGERESPONSE._serialized_end=4460 - _CONSTRAINSTAGEREQUEST._serialized_start=4462 - _CONSTRAINSTAGEREQUEST._serialized_end=4545 - _CONSTRAINSTAGERESPONSE._serialized_start=4548 - _CONSTRAINSTAGERESPONSE._serialized_end=4775 - _BOOTSTRAPREQUEST._serialized_start=4778 - _BOOTSTRAPREQUEST._serialized_end=4936 - _BOOTSTRAPREQUEST_HINTSENTRY._serialized_start=4892 - _BOOTSTRAPREQUEST_HINTSENTRY._serialized_end=4936 - _BOOTSTRAPRESPONSE._serialized_start=4938 - _BOOTSTRAPRESPONSE._serialized_end=4986 - _RESETREQUEST._serialized_start=4988 - _RESETREQUEST._serialized_end=5018 - _RESETRESPONSE._serialized_start=5020 - _RESETRESPONSE._serialized_end=5064 - _PLUGINREGISTRY._serialized_start=5223 - _PLUGINREGISTRY._serialized_end=5653 - _PREDICTPLUGIN._serialized_start=5655 - _PREDICTPLUGIN._serialized_end=5776 - _PROPOSEPLUGIN._serialized_start=5778 - _PROPOSEPLUGIN._serialized_end=5899 - _RECONCILEPLUGIN._serialized_start=5902 - _RECONCILEPLUGIN._serialized_end=6031 - _CONSTRAINPLUGIN._serialized_start=6034 - _CONSTRAINPLUGIN._serialized_end=6163 - _PLUGINLIFECYCLE._serialized_start=6166 - _PLUGINLIFECYCLE._serialized_end=6375 +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dynamo.planner.plugins.proto.v1.plugin_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + DESCRIPTOR._loaded_options = None + _globals['_FPMDATA_PREFILLENGINESENTRY']._loaded_options = None + _globals['_FPMDATA_PREFILLENGINESENTRY']._serialized_options = b'8\001' + _globals['_FPMDATA_DECODEENGINESENTRY']._loaded_options = None + _globals['_FPMDATA_DECODEENGINESENTRY']._serialized_options = b'8\001' + _globals['_BOOTSTRAPREQUEST_HINTSENTRY']._loaded_options = None + _globals['_BOOTSTRAPREQUEST_HINTSENTRY']._serialized_options = b'8\001' + _globals['_HOLDPOLICY']._serialized_start=6046 + _globals['_HOLDPOLICY']._serialized_end=6095 + _globals['_CIRCUITSTATE']._serialized_start=6097 + _globals['_CIRCUITSTATE']._serialized_end=6148 + _globals['_OVERRIDETYPE']._serialized_start=6150 + _globals['_OVERRIDETYPE']._serialized_end=6200 + _globals['_REGISTERREQUEST']._serialized_start=75 + _globals['_REGISTERREQUEST']._serialized_end=423 + _globals['_REGISTERRESPONSE']._serialized_start=425 + _globals['_REGISTERRESPONSE']._serialized_end=521 + _globals['_HEARTBEATREQUEST']._serialized_start=523 + _globals['_HEARTBEATREQUEST']._serialized_end=580 + _globals['_HEARTBEATRESPONSE']._serialized_start=582 + _globals['_HEARTBEATRESPONSE']._serialized_end=613 + _globals['_UNREGISTERREQUEST']._serialized_start=615 + _globals['_UNREGISTERREQUEST']._serialized_end=689 + _globals['_UNREGISTERRESPONSE']._serialized_start=691 + _globals['_UNREGISTERRESPONSE']._serialized_end=723 + _globals['_LISTPLUGINSREQUEST']._serialized_start=725 + _globals['_LISTPLUGINSREQUEST']._serialized_end=793 + _globals['_LISTPLUGINSRESPONSE']._serialized_start=795 + _globals['_LISTPLUGINSRESPONSE']._serialized_end=871 + _globals['_PLUGININFO']._serialized_start=874 + _globals['_PLUGININFO']._serialized_end=1194 + _globals['_PIPELINECONTEXT']._serialized_start=1197 + _globals['_PIPELINECONTEXT']._serialized_end=1590 + _globals['_OBSERVATIONDATA']._serialized_start=1593 + _globals['_OBSERVATIONDATA']._serialized_end=1897 + _globals['_BATCHSCHEDULINGDATA']._serialized_start=1900 + _globals['_BATCHSCHEDULINGDATA']._serialized_end=2131 + _globals['_BATCHJOBDEMAND']._serialized_start=2134 + _globals['_BATCHJOBDEMAND']._serialized_end=2393 + _globals['_POOLTRAFFICDEMAND']._serialized_start=2396 + _globals['_POOLTRAFFICDEMAND']._serialized_end=2551 + _globals['_BATCHDISPATCHERFEEDBACK']._serialized_start=2554 + _globals['_BATCHDISPATCHERFEEDBACK']._serialized_end=2800 + _globals['_TRAFFICMETRICS']._serialized_start=2803 + _globals['_TRAFFICMETRICS']._serialized_end=2970 + _globals['_FPMDATA']._serialized_start=2973 + _globals['_FPMDATA']._serialized_end=3249 + _globals['_FPMDATA_PREFILLENGINESENTRY']._serialized_start=3142 + _globals['_FPMDATA_PREFILLENGINESENTRY']._serialized_end=3195 + _globals['_FPMDATA_DECODEENGINESENTRY']._serialized_start=3197 + _globals['_FPMDATA_DECODEENGINESENTRY']._serialized_end=3249 + _globals['_WORKERSTATE']._serialized_start=3252 + _globals['_WORKERSTATE']._serialized_end=3603 + _globals['_PREDICTIONDATA']._serialized_start=3606 + _globals['_PREDICTIONDATA']._serialized_end=3912 + _globals['_SCALINGPROPOSAL']._serialized_start=3914 + _globals['_SCALINGPROPOSAL']._serialized_end=4023 + _globals['_COMPONENTTARGET']._serialized_start=4026 + _globals['_COMPONENTTARGET']._serialized_end=4167 + _globals['_OVERRIDERESULT']._serialized_start=4169 + _globals['_OVERRIDERESULT']._serialized_end=4261 + _globals['_ACCEPTRESULT']._serialized_start=4263 + _globals['_ACCEPTRESULT']._serialized_end=4277 + _globals['_REJECTRESULT']._serialized_start=4279 + _globals['_REJECTRESULT']._serialized_end=4309 + _globals['_PREDICTSTAGEREQUEST']._serialized_start=4311 + _globals['_PREDICTSTAGEREQUEST']._serialized_end=4392 + _globals['_PREDICTSTAGERESPONSE']._serialized_start=4394 + _globals['_PREDICTSTAGERESPONSE']._serialized_end=4510 + _globals['_PROPOSESTAGEREQUEST']._serialized_start=4512 + _globals['_PROPOSESTAGEREQUEST']._serialized_end=4593 + _globals['_PROPOSESTAGERESPONSE']._serialized_start=4596 + _globals['_PROPOSESTAGERESPONSE']._serialized_end=4821 + _globals['_RECONCILESTAGEREQUEST']._serialized_start=4824 + _globals['_RECONCILESTAGEREQUEST']._serialized_end=4967 + _globals['_PROPOSERESULT']._serialized_start=4970 + _globals['_PROPOSERESULT']._serialized_end=5210 + _globals['_RECONCILESTAGERESPONSE']._serialized_start=5213 + _globals['_RECONCILESTAGERESPONSE']._serialized_end=5440 + _globals['_CONSTRAINSTAGEREQUEST']._serialized_start=5442 + _globals['_CONSTRAINSTAGEREQUEST']._serialized_end=5525 + _globals['_CONSTRAINSTAGERESPONSE']._serialized_start=5528 + _globals['_CONSTRAINSTAGERESPONSE']._serialized_end=5755 + _globals['_BOOTSTRAPREQUEST']._serialized_start=5758 + _globals['_BOOTSTRAPREQUEST']._serialized_end=5916 + _globals['_BOOTSTRAPREQUEST_HINTSENTRY']._serialized_start=5872 + _globals['_BOOTSTRAPREQUEST_HINTSENTRY']._serialized_end=5916 + _globals['_BOOTSTRAPRESPONSE']._serialized_start=5918 + _globals['_BOOTSTRAPRESPONSE']._serialized_end=5966 + _globals['_RESETREQUEST']._serialized_start=5968 + _globals['_RESETREQUEST']._serialized_end=5998 + _globals['_RESETRESPONSE']._serialized_start=6000 + _globals['_RESETRESPONSE']._serialized_end=6044 + _globals['_PLUGINREGISTRY']._serialized_start=6203 + _globals['_PLUGINREGISTRY']._serialized_end=6633 + _globals['_PREDICTPLUGIN']._serialized_start=6635 + _globals['_PREDICTPLUGIN']._serialized_end=6756 + _globals['_PROPOSEPLUGIN']._serialized_start=6758 + _globals['_PROPOSEPLUGIN']._serialized_end=6879 + _globals['_RECONCILEPLUGIN']._serialized_start=6882 + _globals['_RECONCILEPLUGIN']._serialized_end=7011 + _globals['_CONSTRAINPLUGIN']._serialized_start=7014 + _globals['_CONSTRAINPLUGIN']._serialized_end=7143 + _globals['_PLUGINLIFECYCLE']._serialized_start=7146 + _globals['_PLUGINLIFECYCLE']._serialized_end=7355 # @@protoc_insertion_point(module_scope) diff --git a/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.pyi b/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.pyi index a32285d93c2b..eee8156e6f56 100644 --- a/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.pyi +++ b/components/src/dynamo/planner/plugins/proto/v1/plugin_pb2.pyi @@ -161,14 +161,80 @@ class PipelineContext(_message.Message): def __init__(self, request_id: _Optional[str] = ..., decision_id: _Optional[str] = ..., observations: _Optional[_Union[ObservationData, _Mapping]] = ..., predictions: _Optional[_Union[PredictionData, _Mapping]] = ..., proposal: _Optional[_Union[ScalingProposal, _Mapping]] = ..., constrained: _Optional[_Union[ScalingProposal, _Mapping]] = ...) -> None: ... class ObservationData(_message.Message): - __slots__ = ("traffic", "fpm", "workers") + __slots__ = ("traffic", "fpm", "workers", "batch") TRAFFIC_FIELD_NUMBER: _ClassVar[int] FPM_FIELD_NUMBER: _ClassVar[int] WORKERS_FIELD_NUMBER: _ClassVar[int] + BATCH_FIELD_NUMBER: _ClassVar[int] traffic: TrafficMetrics fpm: FpmData workers: WorkerState - def __init__(self, traffic: _Optional[_Union[TrafficMetrics, _Mapping]] = ..., fpm: _Optional[_Union[FpmData, _Mapping]] = ..., workers: _Optional[_Union[WorkerState, _Mapping]] = ...) -> None: ... + batch: BatchSchedulingData + def __init__(self, traffic: _Optional[_Union[TrafficMetrics, _Mapping]] = ..., fpm: _Optional[_Union[FpmData, _Mapping]] = ..., workers: _Optional[_Union[WorkerState, _Mapping]] = ..., batch: _Optional[_Union[BatchSchedulingData, _Mapping]] = ...) -> None: ... + +class BatchSchedulingData(_message.Message): + __slots__ = ("job_demands", "pool_traffic", "dispatcher_feedback") + JOB_DEMANDS_FIELD_NUMBER: _ClassVar[int] + POOL_TRAFFIC_FIELD_NUMBER: _ClassVar[int] + DISPATCHER_FEEDBACK_FIELD_NUMBER: _ClassVar[int] + job_demands: _containers.RepeatedCompositeFieldContainer[BatchJobDemand] + pool_traffic: _containers.RepeatedCompositeFieldContainer[PoolTrafficDemand] + dispatcher_feedback: _containers.RepeatedCompositeFieldContainer[BatchDispatcherFeedback] + def __init__(self, job_demands: _Optional[_Iterable[_Union[BatchJobDemand, _Mapping]]] = ..., pool_traffic: _Optional[_Iterable[_Union[PoolTrafficDemand, _Mapping]]] = ..., dispatcher_feedback: _Optional[_Iterable[_Union[BatchDispatcherFeedback, _Mapping]]] = ...) -> None: ... + +class BatchJobDemand(_message.Message): + __slots__ = ("observed_at_s", "pool_id", "job_id", "status", "total_requests", "completed_requests", "failed_requests", "deadline_at_s", "work_class", "remaining_requests") + OBSERVED_AT_S_FIELD_NUMBER: _ClassVar[int] + POOL_ID_FIELD_NUMBER: _ClassVar[int] + JOB_ID_FIELD_NUMBER: _ClassVar[int] + STATUS_FIELD_NUMBER: _ClassVar[int] + TOTAL_REQUESTS_FIELD_NUMBER: _ClassVar[int] + COMPLETED_REQUESTS_FIELD_NUMBER: _ClassVar[int] + FAILED_REQUESTS_FIELD_NUMBER: _ClassVar[int] + DEADLINE_AT_S_FIELD_NUMBER: _ClassVar[int] + WORK_CLASS_FIELD_NUMBER: _ClassVar[int] + REMAINING_REQUESTS_FIELD_NUMBER: _ClassVar[int] + observed_at_s: float + pool_id: str + job_id: str + status: str + total_requests: int + completed_requests: int + failed_requests: int + deadline_at_s: float + work_class: str + remaining_requests: int + def __init__(self, observed_at_s: _Optional[float] = ..., pool_id: _Optional[str] = ..., job_id: _Optional[str] = ..., status: _Optional[str] = ..., total_requests: _Optional[int] = ..., completed_requests: _Optional[int] = ..., failed_requests: _Optional[int] = ..., deadline_at_s: _Optional[float] = ..., work_class: _Optional[str] = ..., remaining_requests: _Optional[int] = ...) -> None: ... + +class PoolTrafficDemand(_message.Message): + __slots__ = ("observed_at_s", "pool_id", "online_offered_rps", "frontend_active_requests") + OBSERVED_AT_S_FIELD_NUMBER: _ClassVar[int] + POOL_ID_FIELD_NUMBER: _ClassVar[int] + ONLINE_OFFERED_RPS_FIELD_NUMBER: _ClassVar[int] + FRONTEND_ACTIVE_REQUESTS_FIELD_NUMBER: _ClassVar[int] + observed_at_s: float + pool_id: str + online_offered_rps: float + frontend_active_requests: int + def __init__(self, observed_at_s: _Optional[float] = ..., pool_id: _Optional[str] = ..., online_offered_rps: _Optional[float] = ..., frontend_active_requests: _Optional[int] = ...) -> None: ... + +class BatchDispatcherFeedback(_message.Message): + __slots__ = ("observed_at_s", "pool_id", "observation_window_s", "queued_requests", "inflight_requests", "actual_dispatch_rps", "applied_max_admission_rps") + OBSERVED_AT_S_FIELD_NUMBER: _ClassVar[int] + POOL_ID_FIELD_NUMBER: _ClassVar[int] + OBSERVATION_WINDOW_S_FIELD_NUMBER: _ClassVar[int] + QUEUED_REQUESTS_FIELD_NUMBER: _ClassVar[int] + INFLIGHT_REQUESTS_FIELD_NUMBER: _ClassVar[int] + ACTUAL_DISPATCH_RPS_FIELD_NUMBER: _ClassVar[int] + APPLIED_MAX_ADMISSION_RPS_FIELD_NUMBER: _ClassVar[int] + observed_at_s: float + pool_id: str + observation_window_s: float + queued_requests: int + inflight_requests: int + actual_dispatch_rps: float + applied_max_admission_rps: float + def __init__(self, observed_at_s: _Optional[float] = ..., pool_id: _Optional[str] = ..., observation_window_s: _Optional[float] = ..., queued_requests: _Optional[int] = ..., inflight_requests: _Optional[int] = ..., actual_dispatch_rps: _Optional[float] = ..., applied_max_admission_rps: _Optional[float] = ...) -> None: ... class TrafficMetrics(_message.Message): __slots__ = ("duration_s", "num_req", "isl", "osl", "kv_hit_rate", "accept_length") diff --git a/components/src/dynamo/planner/plugins/types.py b/components/src/dynamo/planner/plugins/types.py index e728bd7c7213..9d098d1f9a95 100644 --- a/components/src/dynamo/planner/plugins/types.py +++ b/components/src/dynamo/planner/plugins/types.py @@ -30,7 +30,7 @@ from enum import IntEnum from typing import Any, Literal, Optional -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator # ---------------------------------------------------------------------------- # Enums (mirror proto enum integer values exactly) @@ -207,10 +207,77 @@ class WorkerState(_ProtoMirror): decode_scaling_in_progress: Optional[bool] = None +class BatchJobDemand(_ProtoMirror): + """Auditable batch-job demand for one inference pool. + + Both timestamps are absolute wall-clock values. ``deadline_at_s=None`` + denotes a job without an SLA deadline. ``remaining_requests`` is included + on the wire for external plugins and is derived from the raw counters by + the core adapter. + """ + + observed_at_s: float = 0.0 + pool_id: str = "" + job_id: str = "" + status: str = "" + total_requests: int = 0 + completed_requests: int = 0 + failed_requests: int = 0 + deadline_at_s: Optional[float] = None + work_class: str = "" + remaining_requests: int = 0 + + @model_validator(mode="after") + def validate_request_counters(self) -> BatchJobDemand: + if self.completed_requests + self.failed_requests > self.total_requests: + raise ValueError( + "completed_requests + failed_requests must not exceed total_requests" + ) + return self + + +class PoolTrafficDemand(_ProtoMirror): + """Online traffic and current frontend concurrency for one pool. + + ``frontend_active_requests=None`` means that the source did not report the + gauge; ``0`` is an explicitly observed idle frontend. + """ + + observed_at_s: float = 0.0 + pool_id: str = "" + online_offered_rps: float = 0.0 + frontend_active_requests: Optional[int] = Field(default=None, ge=0) + + +class BatchDispatcherFeedback(_ProtoMirror): + """Observed batch dispatcher state for one inference pool. + + ``applied_max_admission_rps=None`` means no applied Planner cap was + reported; ``0.0`` is an explicit pause. + """ + + observed_at_s: float = 0.0 + pool_id: str = "" + observation_window_s: float = 0.0 + queued_requests: int = 0 + inflight_requests: int = 0 + actual_dispatch_rps: float = 0.0 + applied_max_admission_rps: Optional[float] = None + + +class BatchSchedulingData(_ProtoMirror): + """Flat batch scheduling inputs, each carrying its source timestamp.""" + + job_demands: list[BatchJobDemand] = Field(default_factory=list) + pool_traffic: list[PoolTrafficDemand] = Field(default_factory=list) + dispatcher_feedback: list[BatchDispatcherFeedback] = Field(default_factory=list) + + class ObservationData(_ProtoMirror): traffic: Optional[TrafficMetrics] = None fpm: Optional[FpmData] = None workers: Optional[WorkerState] = None + batch: Optional[BatchSchedulingData] = None class PredictionData(_ProtoMirror): @@ -448,6 +515,10 @@ class ResetResponse(_ProtoMirror): "TrafficMetrics", "FpmData", "WorkerState", + "BatchJobDemand", + "PoolTrafficDemand", + "BatchDispatcherFeedback", + "BatchSchedulingData", "ObservationData", "PredictionData", "ComponentTarget", diff --git a/components/src/dynamo/planner/tests/plugins/orchestrator/test_engine_adapter.py b/components/src/dynamo/planner/tests/plugins/orchestrator/test_engine_adapter.py index 7269cff3e110..e1ed4e628486 100644 --- a/components/src/dynamo/planner/tests/plugins/orchestrator/test_engine_adapter.py +++ b/components/src/dynamo/planner/tests/plugins/orchestrator/test_engine_adapter.py @@ -26,8 +26,12 @@ from dynamo.planner.config.planner_config import PlannerConfig from dynamo.planner.core.types import ( + BatchDispatcherFeedback, + BatchJobDemand, + BatchSchedulingObservation, EngineCapabilities, FpmObservations, + PoolTrafficDemand, ScheduledTick, TickInput, TrafficObservation, @@ -69,6 +73,58 @@ def _agg_config_throughput_on() -> PlannerConfig: ) +def test_tick_input_to_context_maps_batch_observations(): + adapter = OrchestratorEngineAdapter(_agg_config_throughput_on(), _caps()) + tick_input = TickInput( + now_s=1_700_000_010.0, + batch=BatchSchedulingObservation( + job_demands=[ + BatchJobDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + job_id="batch-123", + status="in_progress", + total_requests=1_000, + completed_requests=275, + failed_requests=25, + deadline_at_s=1_700_003_600.0, + work_class="chat-8k", + ) + ], + pool_traffic=[ + PoolTrafficDemand( + observed_at_s=1_700_000_001.0, + pool_id="pool-a", + online_offered_rps=90.0, + ) + ], + dispatcher_feedback=[ + BatchDispatcherFeedback( + observed_at_s=1_700_000_002.0, + pool_id="pool-a", + observation_window_s=30.0, + queued_requests=700, + inflight_requests=20, + actual_dispatch_rps=9.5, + applied_max_admission_rps=10.0, + ) + ], + ), + ) + + context = adapter._tick_input_to_context(tick_input) + + assert context.observations.batch is not None + assert context.observations.batch.job_demands[0].remaining_requests == 700 + assert context.observations.batch.job_demands[0].deadline_at_s == 1_700_003_600.0 + assert context.observations.batch.pool_traffic[0].online_offered_rps == 90.0 + assert context.observations.batch.dispatcher_feedback[0].queued_requests == 700 + assert ( + context.observations.batch.dispatcher_feedback[0].applied_max_admission_rps + == 10.0 + ) + + def _agg_config_custom_intervals( *, load_interval: float, throughput_interval: float ) -> PlannerConfig: diff --git a/components/src/dynamo/planner/tests/plugins/proto/test_round_trip.py b/components/src/dynamo/planner/tests/plugins/proto/test_round_trip.py index a49e31573949..36da869e0eec 100644 --- a/components/src/dynamo/planner/tests/plugins/proto/test_round_trip.py +++ b/components/src/dynamo/planner/tests/plugins/proto/test_round_trip.py @@ -229,6 +229,128 @@ def test_pipeline_context_full(): _round_trip_wire(msg) +def test_batch_scheduling_data_full(): + msg = pyd.BatchSchedulingData( + job_demands=[ + pyd.BatchJobDemand( + observed_at_s=1_700_000_000.125, + pool_id="pool-a", + job_id="batch-123", + status="in_progress", + total_requests=1_000, + completed_requests=275, + failed_requests=25, + deadline_at_s=1_700_003_600.5, + work_class="chat-8k", + remaining_requests=700, + ) + ], + pool_traffic=[ + pyd.PoolTrafficDemand( + observed_at_s=1_700_000_001.25, + pool_id="pool-a", + online_offered_rps=90.5, + frontend_active_requests=3, + ) + ], + dispatcher_feedback=[ + pyd.BatchDispatcherFeedback( + observed_at_s=1_700_000_002.5, + pool_id="pool-a", + observation_window_s=30.0, + queued_requests=700, + inflight_requests=20, + actual_dispatch_rps=9.75, + applied_max_admission_rps=10.0, + ) + ], + ) + + _round_trip_pyd(msg) + _round_trip_wire(msg) + observation = pyd.ObservationData(batch=msg) + _round_trip_pyd(observation) + _round_trip_wire(observation) + + +def test_pool_traffic_active_requests_preserves_optional_presence(): + missing = pyd.PoolTrafficDemand(pool_id="pool-a") + missing_proto = pydantic_to_proto(missing) + assert not missing_proto.HasField("frontend_active_requests") + assert proto_to_pydantic(missing_proto) == missing + + idle = pyd.PoolTrafficDemand(pool_id="pool-a", frontend_active_requests=0) + idle_proto = pydantic_to_proto(idle) + assert idle_proto.HasField("frontend_active_requests") + assert idle_proto.frontend_active_requests == 0 + assert proto_to_pydantic(idle_proto) == idle + + busy = pyd.PoolTrafficDemand(pool_id="pool-a", frontend_active_requests=9) + assert _round_trip_pyd(busy) == busy + + +def test_batch_job_demand_allows_counters_equal_to_total(): + demand = pyd.BatchJobDemand( + total_requests=10, + completed_requests=8, + failed_requests=2, + remaining_requests=0, + ) + + assert proto_to_pydantic(pydantic_to_proto(demand)) == demand + + +def test_batch_job_demand_rejects_counter_sum_above_total(): + with pytest.raises( + ValueError, + match=r"completed_requests \+ failed_requests must not exceed total_requests", + ): + pyd.BatchJobDemand( + total_requests=10, + completed_requests=9, + failed_requests=2, + remaining_requests=-1, + ) + + +def test_batch_job_demand_proto_hydration_rejects_counter_sum_above_total(): + wire_demand = pb.BatchJobDemand( + total_requests=10, + completed_requests=9, + failed_requests=2, + remaining_requests=-1, + ) + + with pytest.raises( + ValueError, + match=r"completed_requests \+ failed_requests must not exceed total_requests", + ): + proto_to_pydantic(wire_demand) + + +def test_batch_dispatcher_applied_cap_preserves_unset_vs_zero(): + common = dict( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + observation_window_s=30.0, + queued_requests=100, + inflight_requests=5, + actual_dispatch_rps=0.0, + ) + + unreported = pydantic_to_proto(pyd.BatchDispatcherFeedback(**common)) + assert not unreported.HasField("applied_max_admission_rps") + + paused = pydantic_to_proto( + pyd.BatchDispatcherFeedback(**common, applied_max_admission_rps=0.0) + ) + assert paused.HasField("applied_max_admission_rps") + assert paused.applied_max_admission_rps == 0.0 + _round_trip_pyd( + pyd.BatchDispatcherFeedback(**common, applied_max_admission_rps=0.0) + ) + + def test_prediction_data_optional_unset_vs_zero(): """Critical invariant: optional float fields distinguish None ("no opinion") from 0.0 ("I assert zero"). Removing the diff --git a/components/src/dynamo/planner/tests/unit/test_batch_contracts.py b/components/src/dynamo/planner/tests/unit/test_batch_contracts.py new file mode 100644 index 000000000000..cc99fe5483a1 --- /dev/null +++ b/components/src/dynamo/planner/tests/unit/test_batch_contracts.py @@ -0,0 +1,226 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Focused tests for the generic batch scheduling core contracts.""" + +from __future__ import annotations + +import math + +import pytest + +from dynamo.planner.core.types import ( + BatchDispatcherFeedback, + BatchDrainLimitDecision, + BatchJobDemand, + BatchSchedulingObservation, + PlannerEffects, + PoolTrafficDemand, + TickInput, +) + +pytestmark = [ + pytest.mark.gpu_0, + pytest.mark.pre_merge, + pytest.mark.unit, + pytest.mark.planner, +] + + +def test_batch_contracts_are_compatibility_safe_and_not_shared(): + tick_input = TickInput(now_s=1_700_000_000.0) + first = PlannerEffects() + second = PlannerEffects() + + assert tick_input.batch is None + assert first.batch_drain_limits == [] + assert second.batch_drain_limits == [] + + first.batch_drain_limits.append( + BatchDrainLimitDecision( + pool_id="pool-a", + max_admission_rps=0.0, + valid_until_s=1_700_000_030.0, + decision_id="decision-1", + ) + ) + assert second.batch_drain_limits == [] + + +def test_batch_job_preserves_raw_counters_and_derives_remaining_requests(): + job = BatchJobDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + job_id="batch-123", + status="in_progress", + total_requests=1_000, + completed_requests=275, + failed_requests=25, + deadline_at_s=1_700_003_600.0, + work_class="chat-8k", + ) + + assert job.remaining_requests == 700 + + with pytest.raises(ValueError, match="must not exceed"): + BatchJobDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + job_id="batch-invalid", + status="in_progress", + total_requests=10, + completed_requests=9, + failed_requests=2, + deadline_at_s=None, + work_class="chat-8k", + ) + + +def test_batch_observation_distinguishes_unreported_cap_from_pause(): + observation = BatchSchedulingObservation( + pool_traffic=[ + PoolTrafficDemand( + observed_at_s=1_700_000_001.0, + pool_id="pool-a", + online_offered_rps=90.0, + frontend_active_requests=0, + ) + ], + dispatcher_feedback=[ + BatchDispatcherFeedback( + observed_at_s=1_700_000_002.0, + pool_id="pool-a", + observation_window_s=30.0, + queued_requests=700, + inflight_requests=20, + actual_dispatch_rps=0.0, + applied_max_admission_rps=None, + ), + BatchDispatcherFeedback( + observed_at_s=1_700_000_003.0, + pool_id="pool-b", + observation_window_s=30.0, + queued_requests=300, + inflight_requests=0, + actual_dispatch_rps=0.0, + applied_max_admission_rps=0.0, + ), + ], + ) + + assert observation.dispatcher_feedback[0].applied_max_admission_rps is None + assert observation.dispatcher_feedback[1].applied_max_admission_rps == 0.0 + assert observation.pool_traffic[0].frontend_active_requests == 0 + + +def test_pool_traffic_active_requests_is_optional_and_non_negative(): + legacy = PoolTrafficDemand( + observed_at_s=1_700_000_001.0, + pool_id="pool-a", + online_offered_rps=0.0, + ) + busy = PoolTrafficDemand( + observed_at_s=1_700_000_002.0, + pool_id="pool-a", + online_offered_rps=1.0, + frontend_active_requests=7, + ) + + assert legacy.frontend_active_requests is None + assert busy.frontend_active_requests == 7 + + for invalid in (-1, 1.5, True): + with pytest.raises(ValueError, match="non-negative integer or None"): + PoolTrafficDemand( + observed_at_s=1_700_000_003.0, + pool_id="pool-a", + online_offered_rps=1.0, + frontend_active_requests=invalid, # type: ignore[arg-type] + ) + + +def test_frontend_active_requests_distinguishes_unknown_from_idle(): + unknown = PoolTrafficDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + online_offered_rps=0.0, + ) + idle = PoolTrafficDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + online_offered_rps=0.0, + frontend_active_requests=0, + ) + + assert unknown.frontend_active_requests is None + assert idle.frontend_active_requests == 0 + + +@pytest.mark.parametrize("invalid_count", [-1, 1.5, True]) +def test_frontend_active_requests_requires_integral_count(invalid_count: object): + with pytest.raises(ValueError, match="frontend_active_requests"): + PoolTrafficDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + online_offered_rps=0.0, + frontend_active_requests=invalid_count, # type: ignore[arg-type] + ) + + +def test_batch_drain_limit_rejects_negative_rates_but_allows_pause(): + paused = BatchDrainLimitDecision( + pool_id="pool-a", + max_admission_rps=0.0, + valid_until_s=1_700_000_030.0, + decision_id="decision-pause", + ) + assert paused.max_admission_rps == 0.0 + + with pytest.raises(ValueError, match="non-negative"): + BatchDrainLimitDecision( + pool_id="pool-a", + max_admission_rps=-1.0, + valid_until_s=1_700_000_030.0, + decision_id="decision-invalid", + ) + + +def test_batch_contracts_reject_nonfinite_rates(): + invalid_rate = math.nan + with pytest.raises(ValueError, match="finite and non-negative"): + PoolTrafficDemand( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + online_offered_rps=invalid_rate, + ) + with pytest.raises(ValueError, match="finite and non-negative"): + BatchDrainLimitDecision( + pool_id="pool-a", + max_admission_rps=invalid_rate, + valid_until_s=1_700_000_030.0, + decision_id="decision-invalid", + ) + + +@pytest.mark.parametrize("invalid_identifier", ["", None]) +def test_batch_contracts_require_string_identifiers(invalid_identifier: object): + with pytest.raises(ValueError, match="non-empty string"): + BatchDrainLimitDecision( + pool_id=invalid_identifier, # type: ignore[arg-type] + max_admission_rps=1.0, + valid_until_s=1_700_000_030.0, + decision_id="decision-invalid", + ) + + +@pytest.mark.parametrize("invalid_count", [-1, 1.5, True]) +def test_batch_contracts_require_integral_counts(invalid_count: object): + with pytest.raises(ValueError, match="non-negative integers"): + BatchDispatcherFeedback( + observed_at_s=1_700_000_000.0, + pool_id="pool-a", + observation_window_s=30.0, + queued_requests=invalid_count, # type: ignore[arg-type] + inflight_requests=0, + actual_dispatch_rps=0.0, + )