Documentation
¶
Index ¶
- Constants
- func ConstOpenGate() pipeline.Gate
- type BinaryMetricDispatchGate
- type CachedMetricSource
- type CascadeMetricSource
- type CompositeGate
- type DispatchGateFunc
- type GateFactory
- type GatingMode
- type LocalConcurrencyGate
- type MetricDispatchGate
- func NewBudgetDispatchGate(source MetricSource, baseline float64, fallback float64) *MetricDispatchGate
- func NewMetricDispatchGate(source MetricSource, threshold float64, fallback float64) *MetricDispatchGate
- func NewSaturationDispatchGate(source MetricSource, threshold float64, fallback float64) *MetricDispatchGate
- func (g *MetricDispatchGate) Apply(ctx context.Context, msg *api.InternalRequest, ...) (pipeline.Verdict, error)
- func (g *MetricDispatchGate) Budget(ctx context.Context) float64
- func (g *MetricDispatchGate) WithInferencePool(pool string) *MetricDispatchGate
- func (g *MetricDispatchGate) WithOwner(owner pipeline.GateOwner) *MetricDispatchGate
- type MetricSource
- type PromQLMetricSource
- func NewFlowControlQueueSizePromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, ...) (*PromQLMetricSource, error)
- func NewGMPPromQLMetricSource(projectID string, expr string) (*PromQLMetricSource, error)
- func NewPoolQueueSizePromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, ...) (*PromQLMetricSource, error)
- func NewPromQLMetricSource(clientConfig api.Config, expr string) (*PromQLMetricSource, error)
- func NewPromQLMetricSourceFromLabels(promConfig promapi.Config, metricName string, labels map[string]string) (*PromQLMetricSource, error)
- func NewSaturationPromQLSourceFromConfig(promConfig promapi.Config, params map[string]any) (*PromQLMetricSource, error)
- func NewVLLMSaturationPromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, ...) (*PromQLMetricSource, error)
- type Sample
- type ScrapeConfig
- type ScrapeMetricSource
- type TierPriorityAdmissionGate
- type WaitOnRefuseGate
Constants ¶
const DefaultCacheTTL = 5 * time.Second
DefaultCacheTTL is the default TTL for cached Prometheus metric sources.
Variables ¶
This section is empty.
Functions ¶
func ConstOpenGate ¶
Types ¶
type BinaryMetricDispatchGate ¶
type BinaryMetricDispatchGate struct {
// contains filtered or unexported fields
}
BinaryMetricDispatchGate implements DispatchGate using a MetricSource. It returns 0.0 (no capacity) if the metric value is non-zero, and 1.0 (full capacity) if the metric value is zero.
func AverageQueueSizeGate ¶
func AverageQueueSizeGate() *BinaryMetricDispatchGate
AverageQueueSizeGate creates a BinaryMetricDispatchGate from command-line flags.
func NewBinaryMetricDispatchGateWithSource ¶
func NewBinaryMetricDispatchGateWithSource(source MetricSource) *BinaryMetricDispatchGate
NewBinaryMetricDispatchGateWithSource creates a new gate using the provided MetricSource.
func (*BinaryMetricDispatchGate) Apply ¶
func (g *BinaryMetricDispatchGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
Apply implements pipeline.Gate.
type CachedMetricSource ¶
type CachedMetricSource struct {
// contains filtered or unexported fields
}
CachedMetricSource wraps a MetricSource with a TTL cache so that repeated queries within the TTL return the cached result instead of hitting the backend (e.g. Prometheus) on every call.
func NewCachedMetricSource ¶
func NewCachedMetricSource(source MetricSource, ttl time.Duration) *CachedMetricSource
NewCachedMetricSource wraps the given source with a cache that holds results for the specified TTL duration.
type CascadeMetricSource ¶
type CascadeMetricSource struct {
// contains filtered or unexported fields
}
CascadeMetricSource tries each source in order and returns the first successful result. If a source returns an error or no samples, the next source is tried. Log messages are emitted only on transitions (entering/leaving fallback, or switching fallback source) to avoid noise during sustained outages.
func NewCascadeMetricSource ¶
func NewCascadeMetricSource(sources ...MetricSource) *CascadeMetricSource
NewCascadeMetricSource creates a CascadeMetricSource from the given sources, tried in order. At least two sources are required; panics otherwise.
type CompositeGate ¶
type CompositeGate struct {
// contains filtered or unexported fields
}
CompositeGate combines multiple pipeline.Gates. It returns the minimum budget across all inner Gates. It applies all inner gates (all or nothing) to incoming requests.
func NewCompositeGate ¶
func NewCompositeGate(gates ...pipeline.Gate) *CompositeGate
NewCompositeGate creates a CompositeGate with the given inner gates.
func (*CompositeGate) Apply ¶
func (c *CompositeGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
type DispatchGateFunc ¶
DispatchGateFunc is a function type that implements Gate. This allows any function with the signature func(context.Context) float64 to be used as a Gate.
func (DispatchGateFunc) Apply ¶
func (f DispatchGateFunc) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
Apply implements Gate.
type GateFactory ¶
type GateFactory struct {
// contains filtered or unexported fields
}
GateFactory creates DispatchGate instances based on configuration.
func NewGateFactory ¶
func NewGateFactory(prometheusURL string) *GateFactory
NewGateFactory creates a new GateFactory with an optional Prometheus URL. If prometheusURL is empty, Prometheus gates will fail at creation time. Prometheus metric sources are cached with DefaultCacheTTL; use NewGateFactoryWithCacheTTL to override.
func NewGateFactoryWithCacheTTL ¶
func NewGateFactoryWithCacheTTL(prometheusURL string, cacheTTL time.Duration) *GateFactory
NewGateFactoryWithCacheTTL creates a GateFactory with a custom cache TTL for Prometheus metric sources. A TTL of 0 disables caching.
func (*GateFactory) Close ¶
func (f *GateFactory) Close() error
Close closes all Redis clients created by this factory.
func (*GateFactory) CreateGate ¶
func (f *GateFactory) CreateGate(cfg pipeline.GateConfig) (pipeline.Gate, error)
CreateGate creates a DispatchGate based on the gate type and parameters. Supported gate types:
- "constant": Always returns budget 1.0 (fully open)
- "redis": Queries Redis for dispatch budget
- "prometheus-saturation": Queries Prometheus for pool saturation metric. Params: pool (required), threshold (default 0.8), fallback (default 0.0)
- "composite": Combines multiple gates. Params: gates (JSON array of gate configurations)
- "prometheus-budget": Cascades three Prometheus metric sources to compute dispatch budget D, using the first that returns a sample. [0] D = 1 − (queue_size / max_SYS) via inference_extension_flow_control_queue_size. Requires llm-d's flow control plugin, which the llm-d router does not enable. [1] D = 1 − (mean per-pod queue depth / max_concurrency) via inference_pool_per_pod_queue_size. Part of EPP's base metric set, so this is the source a stock install lands on. [2] D = 1 − (vllm_running / max_SYS). Filters vLLM metrics by inference_pool label, which vLLM does not emit natively — model server pods must carry this label and Prometheus must be configured with metric relabeling to propagate it (see docs/guides/e2e-deploy.md). Sources [0] and [2] compute max_SYS = ready_pods × max_concurrency dynamically; [1] averages over pods, in which the ready_pods factor cancels. Gate closes when D ≤ B (baseline); returns D − B when open, so callers compute N = max_SYS × (D − B). Params: pool (required), max_concurrency (default 100), baseline (default 0.05), fallback (default 0.0) max_concurrency is per ready pod, not for the pool as a whole: the gate closes once the observed load reaches max_concurrency × (1 − baseline) per ready pod. Set it too high for the pool's real capacity and the gate never closes; the resolved closing point is logged at gate creation so this is visible.
- "prometheus-query": Evaluates an arbitrary user-supplied PromQL expression as the dispatch budget. The expression must resolve to an instant vector with a single sample whose value is in [0, 1]. Unlike prometheus-saturation and prometheus-budget, this gate does not construct queries internally — the user provides the complete PromQL expression. Params: query (required), fallback (default 0.0)
For unsupported or unknown gate types, returns ConstOpenGate as a safe default.
func (*GateFactory) WithLogger ¶ added in v0.9.0
func (f *GateFactory) WithLogger(logger logr.Logger) *GateFactory
WithLogger sets the logger the factory uses to report the PromQL each Prometheus gate resolved to. Gates build their queries from gate_params rather than taking them verbatim, so without this the only way to find out what is actually being asked of Prometheus is to read the source.
type GatingMode ¶
type GatingMode string
const ( GatingModeBlocking GatingMode = "blocking" GatingModeClassifying GatingMode = "classifying" )
type LocalConcurrencyGate ¶
type LocalConcurrencyGate struct {
// contains filtered or unexported fields
}
LocalConcurrencyGate limits the number of concurrent in-flight requests processed from a single queue locally.
func NewLocalConcurrencyGate ¶
func NewLocalConcurrencyGate(limit int) *LocalConcurrencyGate
NewLocalConcurrencyGate creates a new LocalConcurrencyGate with the specified limit.
func (*LocalConcurrencyGate) Apply ¶
func (g *LocalConcurrencyGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
Apply implements pipeline.Gate. Returns VerdictContinue if request fits in budget, VerdictRefuse with redeliver otherwise.
func (*LocalConcurrencyGate) Budget ¶
func (g *LocalConcurrencyGate) Budget(ctx context.Context) float64
Budget implements pipeline.Gate. Returns the fraction of available capacity in [0.0, 1.0].
func (*LocalConcurrencyGate) WithGatingMode ¶
func (g *LocalConcurrencyGate) WithGatingMode(mode GatingMode) *LocalConcurrencyGate
WithGatingMode configures the gating mode (blocking or classifying).
type MetricDispatchGate ¶
type MetricDispatchGate struct {
// contains filtered or unexported fields
}
MetricDispatchGate implements DispatchGate by querying a MetricSource for a budget value D and returning D − threshold, clamped to [0.0, 1.0].
The gate closes (returns 0.0) when D ≤ threshold. This implements the doc formula N = max_SYS × (D − B) when threshold is set to the reserved baseline B. On error or missing/invalid data, the gate returns the configured fallback budget.
func NewBudgetDispatchGate ¶
func NewBudgetDispatchGate(source MetricSource, baseline float64, fallback float64) *MetricDispatchGate
NewBudgetDispatchGate creates a MetricDispatchGate for the prometheus-budget use case. The source should return the dispatch budget D (e.g. D = 1 − F, where F is EPP fullness, or D = 1 − S, where S is inference pool saturation). baseline is the reserved baseline B ∈ [0, 1]: the gate closes when D ≤ B and returns D − B when open, so the caller computes N = max_SYS × (D − B). fallback is returned on error, clamped to [0.0, 1.0].
func NewMetricDispatchGate ¶
func NewMetricDispatchGate(source MetricSource, threshold float64, fallback float64) *MetricDispatchGate
NewMetricDispatchGate creates a MetricDispatchGate with the given source, threshold, and fallback budget value. The fallback is clamped to [0.0, 1.0].
func NewSaturationDispatchGate ¶
func NewSaturationDispatchGate(source MetricSource, threshold float64, fallback float64) *MetricDispatchGate
NewSaturationDispatchGate creates a MetricDispatchGate for the saturation use case. The source should return a budget value D (e.g. 1 − saturation). The threshold and fallback are given in saturation space and converted to budget space via 1 − value.
func (*MetricDispatchGate) Apply ¶
func (g *MetricDispatchGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
Apply implements pipeline.Gate.
func (*MetricDispatchGate) Budget ¶
func (g *MetricDispatchGate) Budget(ctx context.Context) float64
Budget implements DispatchGate. Returns 0.0 when D ≤ threshold (gate closed), otherwise D − threshold clamped to [0.0, 1.0]. On error or missing data the gate returns the configured fallback budget.
func (*MetricDispatchGate) WithInferencePool ¶ added in v0.9.0
func (g *MetricDispatchGate) WithInferencePool(pool string) *MetricDispatchGate
WithInferencePool sets the InferencePool this gate queries, exposed as the inference_pool label on the gauges above. It is what the gate measures, not who it throttles — pool_name is the latter.
func (*MetricDispatchGate) WithOwner ¶ added in v0.9.0
func (g *MetricDispatchGate) WithOwner(owner pipeline.GateOwner) *MetricDispatchGate
WithOwner sets the queue or worker pool this gate belongs to. Its queue_id/queue_name/pool_name label the async_gate_metric_value and async_gate_metric_threshold gauges, so they join with the owner's other series (async_dispatch_budget, async_gate_decisions_total, ...).
type MetricSource ¶
type MetricSource interface {
// Query returns the current samples for the preconfigured query.
Query(ctx context.Context) ([]Sample, error)
}
MetricSource queries a metrics backend for time-series data. The query configuration is baked into the implementation at construction time; callers simply invoke Query to retrieve the current samples.
type PromQLMetricSource ¶
type PromQLMetricSource struct {
// contains filtered or unexported fields
}
PromQLMetricSource implements MetricSource by executing a PromQL expression against a Prometheus-compatible API.
func NewFlowControlQueueSizePromQL ¶
func NewFlowControlQueueSizePromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, namespace string) (*PromQLMetricSource, error)
NewFlowControlQueueSizePromQL builds a PromQLMetricSource that returns the EPP queue depth as a dispatch budget D = 1 − (queue_size / (ready_pods × maxConcurrency)), where queue_size is inference_extension_flow_control_queue_size and max_SYS = ready_pods × maxConcurrency is computed dynamically from the inference_pool_ready_pods metric. inferencePool and maxConcurrency are required.
func NewGMPPromQLMetricSource ¶
func NewGMPPromQLMetricSource(projectID string, expr string) (*PromQLMetricSource, error)
NewGMPPromQLMetricSource creates a PromQL MetricSource for Google Managed Prometheus.
func NewPoolQueueSizePromQL ¶ added in v0.9.0
func NewPoolQueueSizePromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, namespace string) (*PromQLMetricSource, error)
NewPoolQueueSizePromQL builds a PromQLMetricSource that returns the model server queue depth EPP reports per pod as a dispatch budget D = 1 − (mean per-pod queue depth / maxConcurrency).
inference_extension_flow_control_queue_size is only recorded when EPP runs the flow control plugin, which the llm-d router does not enable. inference_pool_per_pod_queue_size is part of EPP's base metric set, so this source resolves on a stock install.
Averaging over pods is what makes max_SYS = ready_pods × maxConcurrency reduce to maxConcurrency, so this needs no inference_pool_ready_pods join. That matters for more than brevity: EPP's metrics refresh returns early when the pool has no pods, freezing inference_pool_ready_pods (and inference_pool_average_queue_size, which is why that metric is not used here) at their last values. A drained pool would then read as idle capacity. inference_pool_per_pod_queue_size is emitted by a scrape-time collector instead, so it simply stops reporting — the query yields no samples and the cascade moves on rather than opening the gate onto a pool with nothing behind it.
avg is sum/count over the per-pod series, so the result does not change when several EPP replicas each report the same pods. EPP labels its inference_pool_* series with "name", not "inference_pool" (the same label inference_pool_ready_pods uses above). inferencePool and maxConcurrency are required.
func NewPromQLMetricSource ¶
func NewPromQLMetricSource(clientConfig api.Config, expr string) (*PromQLMetricSource, error)
NewPromQLMetricSource creates a MetricSource that executes the given PromQL expression.
func NewPromQLMetricSourceFromLabels ¶
func NewPromQLMetricSourceFromLabels(promConfig promapi.Config, metricName string, labels map[string]string) (*PromQLMetricSource, error)
NewPromQLMetricSourceFromLabels constructs a PromQL instant vector selector from a metric name and label matchers, and returns a PromQLMetricSource for it.
func NewSaturationPromQLSourceFromConfig ¶
func NewSaturationPromQLSourceFromConfig(promConfig promapi.Config, params map[string]any) (*PromQLMetricSource, error)
NewSaturationPromQLSourceFromConfig builds a PromQLMetricSource for the saturation use case. It returns a budget value (1 - saturation) by constructing a PromQL query of the form "1 - inference_extension_flow_control_pool_saturation{...}", filtered by the "pool" param (required).
func NewVLLMSaturationPromQL ¶
func NewVLLMSaturationPromQL(promConfig promapi.Config, inferencePool string, maxConcurrency float64, namespace string) (*PromQLMetricSource, error)
NewVLLMSaturationPromQL builds a PromQLMetricSource that estimates inference pool saturation from vLLM and pool metrics, returning D = 1 − (running_requests / (ready_pods × maxConcurrency)). This serves as a fallback when EPP flow control metrics are unavailable. inferencePool and maxConcurrency are required.
func (*PromQLMetricSource) Expr ¶
func (s *PromQLMetricSource) Expr() string
type ScrapeConfig ¶
type ScrapeConfig struct {
URL string
MetricName string
Labels map[string]string
MaxCountPerPod float64
PodsURL string
PodsMetric string
PodsLabels map[string]string
}
ScrapeConfig holds configuration for NewScrapeMetricSource.
type ScrapeMetricSource ¶
type ScrapeMetricSource struct {
// contains filtered or unexported fields
}
ScrapeMetricSource implements MetricSource by scraping raw Prometheus /metrics endpoints. It reads a metric value, optionally computes saturation using a max capacity (static or dynamic from a pods metric), and returns budget in [0, 1].
Two modes for max capacity:
- Static: maxCountPerPod is used directly as the total max count (single pod or precomputed).
- Dynamic: when podsURL/podsMetric are set, ready pods are scraped from a second endpoint (e.g., EPP) and max_count = ready_pods * maxCountPerPod.
When maxCountPerPod == 0, the metric value is assumed to already be saturation in [0, 1]. Output value = 1 - saturation (available capacity / budget).
func NewScrapeMetricSource ¶
func NewScrapeMetricSource(cfg ScrapeConfig) *ScrapeMetricSource
NewScrapeMetricSource creates a MetricSource that scrapes Prometheus text-format /metrics endpoints and returns budget values in [0, 1].
type TierPriorityAdmissionGate ¶
type TierPriorityAdmissionGate struct {
// contains filtered or unexported fields
}
func NewTierPriorityAdmissionGate ¶
func NewTierPriorityAdmissionGate(saturationGate pipeline.Gate, tierLabel string) *TierPriorityAdmissionGate
func (*TierPriorityAdmissionGate) Apply ¶
func (g *TierPriorityAdmissionGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
type WaitOnRefuseGate ¶
type WaitOnRefuseGate struct {
// contains filtered or unexported fields
}
WaitOnRefuseGate wraps a single inner gate and converts any ActionRefuse verdict from the inner gate into ActionWait.
func NewWaitOnRefuseGate ¶
func NewWaitOnRefuseGate(inner pipeline.Gate) *WaitOnRefuseGate
NewWaitOnRefuseGate creates a WaitOnRefuseGate with the given inner gate.
func (*WaitOnRefuseGate) Apply ¶
func (w *WaitOnRefuseGate) Apply(ctx context.Context, msg *api.InternalRequest, releases *[]pipeline.GateReleaseFunc) (pipeline.Verdict, error)
Apply implements pipeline.Gate. Calls Apply on the inner gate and overrides ActionRefuse to ActionWait.
Source Files