diff --git a/cmd/runner/runner.go b/cmd/runner/runner.go index 5cf777de..6f6c89e8 100644 --- a/cmd/runner/runner.go +++ b/cmd/runner/runner.go @@ -55,6 +55,7 @@ import ( "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/picker/weightedrandom" inflightrequestsscorer "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/inflightrequests" sessionaffinity "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/sessionaffinity" + ttftawarescorer "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/modelselector/scorer/ttftaware" "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/basemodelextractor" "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/bodyfieldtoheader" modelselectorplugin "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/requesthandling/modelselector" @@ -301,6 +302,7 @@ func (r *Runner) registerInTreePlugins() { plugin.Register(weightedrandom.WeightedRandomPickerType, weightedrandom.WeightedRandomPickerFactory) plugin.Register(modelselectorplugin.ModelSelectorPluginType, modelselectorplugin.ModelSelectorPluginFactory) plugin.Register(inflightrequestsscorer.PluginType, inflightrequestsscorer.ScorerFactory) + plugin.Register(ttftawarescorer.PluginType, ttftawarescorer.ScorerFactory) plugin.Register(sessionaffinity.PluginType, sessionaffinity.ScorerFactory) plugin.Register(modelnametoheader.PluginType, modelnametoheader.PluginFactory) } diff --git a/pkg/framework/plugins/modelselector/scorer/ttftaware/README.md b/pkg/framework/plugins/modelselector/scorer/ttftaware/README.md new file mode 100644 index 00000000..5479ed0a --- /dev/null +++ b/pkg/framework/plugins/modelselector/scorer/ttftaware/README.md @@ -0,0 +1,119 @@ +# TTFT-Aware Scorer + +Routes each request to the model with the lowest predicted TTFT under current load. + +It consumes the per-model snapshot published by the +[TTFT percentile extractor](../../../datalayer/ttftpercentile/README.md) — the service floor +`P10Low`, the operating points `LowTTFT` / `HighTTFT` (default P25 / P50, configured on the +extractor) and their banded inflight averages `InflightAtLow` / `InflightAtHigh` — and turns them +into a prediction and a score. + +## Prediction — `predictedTTFT` + +Every pool has a latency curve: how long a new request waits as a function of how many requests are +already in flight. We measure three points on it and interpolate. + +| point | coordinates | meaning | +|---|---|---| +| **A** | `(0, P10Low)` | queue-free service time | +| **B** | `(InflightAtLow, LowTTFT)` | low operating point (default P25) | +| **C** | `(InflightAtHigh, HighTTFT)` | high operating point (default P50) | + +**A** and **C** always define the curve. **B** is inserted between them only when +[admissible](#when-the-low-point-is-used), splitting it into two segments. Any single prediction +reads one segment, so it uses two of the three points. Every point is something the extractor observed. + +``` +if B admissible: + if inflight < InflightAtLow: # segment A->B + predictedTTFT = P10Low + inflight * (LowTTFT - P10Low) / InflightAtLow + else: # segment B->C, extended past C + predictedTTFT = LowTTFT + (inflight - InflightAtLow) * + (HighTTFT - LowTTFT) / (InflightAtHigh - InflightAtLow) +else: # segment A->C, extended past C + predictedTTFT = P10Low + inflight * (HighTTFT - P10Low) / InflightAtHigh +``` + +The curve is continuous (both branches give `LowTTFT` at `InflightAtLow`), equals `P10Low` at zero +load, and is monotone non-decreasing given `P10Low < LowTTFT < HighTTFT` — which the admissibility +checks enforce. `predictedTTFT` is clamped to `>= P10Low` as a defensive guard. + +**Why three points and not two.** TTFT rises *faster than linearly* with inflight as a pool +approaches saturation. A single chord from the floor to **C** cuts across that convex curve and +under-predicts in between; **B** lets the curve bend so the loaded segment follows the local slope +where the pool is actually operating. + +**Why A→B is a segment of its own.** The B→C slope is measured where queueing dominates, so it is +steep. Running it backwards below **B** makes TTFT cross below the floor within a handful of +requests — a draining-but-still-loaded pool would be predicted at its idle latency and win every +decision it appeared in. Interpolating from `(0, floor)` matches how TTFT flattens as the queue +drains. + +### When the low point is used + +**B** is admissible only when all of these hold. They are conditions on a measured point, not +tuning knobs: + +| check | condition | why | +|---|---|---| +| separated in load | `InflightAtHigh - InflightAtLow >= minInflightGap` | the B→C slope is `ΔTTFT / Δinflight`; if both points sit at the same load that denominator is noise and the slope is meaningless | +| ordered in latency | `HighTTFT > LowTTFT` | TTFT must rise with the percentile. If noise inverts them the slope goes negative and the *most* loaded pool scores best, feeding a saturated pool | +| above the floor | `LowTTFT > P10Low` | `P10Low` is a long-window statistic and `LowTTFT` a recent one, so after a drain the recent P25 can fall below it — which would tilt the A→B segment downwards | +| positive inflight | `InflightAtLow > 0` | keeps the A→B divisor safe; the gap check alone does not imply it | + +When **B** is dropped the curve is the single floor chord **A → C**, which is well defined at any +load. + + +### Model states + +- **cold** — `Floor() == 0` (never observed, or fewer than `minRequests` observations so the floor + is not yet trustworthy). Seeded optimistically at the best observed TTFT. +- **seed** — has a floor but is not calibrated (`RecentN < minRequests`, or no inflight operating + point). Predicts at the floor. +- **trusted** — `RecentN >= minRequests`, `InflightAtHigh > 0`, `HighTTFT > floor`. Uses the curve + above. + +## Score + +``` +score = (maxTTFT - predictedTTFT) / (maxTTFT - minTTFT) +``` + +Lowest predicted TTFT scores highest. Cold models seed at `minTTFT`; if every model is cold, all +score 1.0. + +### Exploration + +An under-observed pool can be starved: competing against a calibrated pool it may never win the +traffic it needs to calibrate. `explorationRate` breaks that loop — each under-observed pool is +flipped independently per request, and with probability `explorationRate` its final score is forced +to `1.0` so the picker sends it a probe; otherwise it is suppressed to `0`, but only when a +calibrated pool exists to take the traffic. The override applies to the **final score only**, so a +probe never distorts the trusted pools' normalisation. + +Together with the extractor's floor sample guard this reproduces what a manual warmup would do: a +new pool reads as cold → receives probes → crosses `minRequests` → competes on its true latency. + +## Parameters + +| Parameter | Default | Description | +|---|---|---| +| `explorationRate` | 0.0 | Per-pool probability that an under-observed pool is probed on a given request. 0 = all traffic to the trusted winner. | +| `minInflightGap` | 2.0 | Minimum inflight separation between the operating points for the low one to be used as an anchor. Must be > 0. | +| `roundTTFTStep` | 0.0 | Quantize each prediction to a multiple of this many seconds before ranking (e.g. `0.01` = 10 ms). Pools landing in the same bucket tie and the picker splits them, instead of one winning on a difference too small to be meaningful. `0` = disabled. Must be >= 0. | + +`minRequests` is read from the extractor's published metrics and configured +[there](../../../datalayer/ttftpercentile/README.md), not here. + +## Configuration + +```yaml +- type: ttft-aware-scorer + parameters: + explorationRate: 0.1 + minInflightGap: 2.0 +``` + +The scorer requires `ttft-percentile-extractor` in `datalayer.extractors`, a model list (e.g. via +`model-config-datasource`) so model selection has candidates, and a picker. diff --git a/pkg/framework/plugins/modelselector/scorer/ttftaware/plugin.go b/pkg/framework/plugins/modelselector/scorer/ttftaware/plugin.go new file mode 100644 index 00000000..42365c7e --- /dev/null +++ b/pkg/framework/plugins/modelselector/scorer/ttftaware/plugin.go @@ -0,0 +1,247 @@ +/* +Copyright 2026 The llm-d Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package ttftaware routes each request to the model with the lowest predicted TTFT under current +// load. The prediction is a piecewise-linear curve through the model's measured points, evaluated +// at its current inflight count. See README.md for the model, the equations and the rationale. +package ttftaware + +import ( + "context" + "encoding/json" + "fmt" + "math" + "math/rand" + + "sigs.k8s.io/controller-runtime/pkg/log" + + logutil "github.com/llm-d/llm-d-inference-payload-processor/pkg/common/observability/logging" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/datalayer" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/modelselector" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/plugin" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/interface/requesthandling" + "github.com/llm-d/llm-d-inference-payload-processor/pkg/framework/plugins/datalayer/ttftpercentile" +) + +const ( + PluginType = "ttft-aware-scorer" + + defaultExplorationRate = 0.0 // disabled; 0.1 probes each under-observed model on ~10% of requests + defaultMinInflightGap = 2.0 // inflight separation the operating points need to define a slope + defaultRoundTTFTStep = 0.0 // disabled; e.g. 0.01 quantizes predictions to 10ms +) + +var _ modelselector.Scorer = &TTFTAwareScorer{} + +// TTFTAwareScorerConfig holds optional parameters for the scorer plugin. See README.md. +type TTFTAwareScorerConfig struct { + // ExplorationRate is the per-model probability that an under-observed model is forced to the + // top score, so it gets the traffic it needs to calibrate. Range [0, 1]. Default 0 (disabled). + ExplorationRate float64 `json:"explorationRate,omitempty"` + // MinInflightGap is the minimum inflight separation between the two operating points for the + // low one to be usable as a curve anchor. Below it their slope is noise. Must be > 0. + MinInflightGap float64 `json:"minInflightGap,omitempty"` + // RoundTTFTStep quantizes each prediction to a multiple of this many seconds before ranking + // (e.g. 0.01 = 10ms), so models closer together than the step tie and the picker splits them + // instead of one winning on a difference too small to be meaningful. Must be >= 0. Default 0. + RoundTTFTStep float64 `json:"roundTTFTStep,omitempty"` +} + +type TTFTAwareScorer struct { + typedName plugin.TypedName + explorationRate float64 + minInflightGap float64 + roundTTFTStep float64 +} + +func ScorerFactory(name string, parameters json.RawMessage, _ plugin.Handle) (plugin.Plugin, error) { + cfg := TTFTAwareScorerConfig{ + ExplorationRate: defaultExplorationRate, + MinInflightGap: defaultMinInflightGap, + RoundTTFTStep: defaultRoundTTFTStep, + } + if len(parameters) > 0 { + if err := json.Unmarshal(parameters, &cfg); err != nil { + return nil, fmt.Errorf("failed to parse parameters for plugin %q: %w", name, err) + } + } + if cfg.ExplorationRate < 0 || cfg.ExplorationRate > 1 { + return nil, fmt.Errorf("explorationRate must be in [0, 1] for plugin %q", name) + } + if cfg.MinInflightGap <= 0 { + return nil, fmt.Errorf("minInflightGap must be > 0 for plugin %q", name) + } + if cfg.RoundTTFTStep < 0 { + return nil, fmt.Errorf("roundTTFTStep must be >= 0 for plugin %q", name) + } + return NewTTFTAwareScorer(). + WithName(name). + WithExplorationRate(cfg.ExplorationRate). + WithMinInflightGap(cfg.MinInflightGap). + WithRoundTTFTStep(cfg.RoundTTFTStep), nil +} + +func NewTTFTAwareScorer() *TTFTAwareScorer { + return &TTFTAwareScorer{ + typedName: plugin.TypedName{Type: PluginType, Name: PluginType}, + explorationRate: defaultExplorationRate, + minInflightGap: defaultMinInflightGap, + roundTTFTStep: defaultRoundTTFTStep, + } +} + +func (s *TTFTAwareScorer) TypedName() plugin.TypedName { return s.typedName } + +func (s *TTFTAwareScorer) WithName(name string) *TTFTAwareScorer { + s.typedName.Name = name + return s +} + +func (s *TTFTAwareScorer) WithExplorationRate(r float64) *TTFTAwareScorer { + s.explorationRate = r + return s +} + +func (s *TTFTAwareScorer) WithMinInflightGap(g float64) *TTFTAwareScorer { + s.minInflightGap = g + return s +} + +func (s *TTFTAwareScorer) WithRoundTTFTStep(step float64) *TTFTAwareScorer { + s.roundTTFTStep = step + return s +} + +// modelEval is the scorer's per-model working state for one Score call. +type modelEval struct { + pred float64 + trusted bool // calibrated operating point + observed bool // has a service floor (not truly cold) +} + +// Score ranks models by predicted TTFT: score = (maxTTFT - predictedTTFT) / (maxTTFT - minTTFT). +// Cold models are seeded at the best observed TTFT; if every model is cold, all score 1.0. +func (s *TTFTAwareScorer) Score(ctx context.Context, _ *plugin.CycleState, _ *requesthandling.InferenceRequest, models []datalayer.Model) map[datalayer.Model]float64 { + evals := make([]modelEval, len(models)) + minPred := math.MaxFloat64 + maxPred := 0.0 + anyObserved := false + anyTrusted := false + + for i, model := range models { + pred, trusted, observed := s.predict(metricsFor(model)) + if s.roundTTFTStep > 0 { + // Quantize before ranking so predictions closer together than the step tie. + pred = math.Round(pred/s.roundTTFTStep) * s.roundTTFTStep + } + evals[i] = modelEval{pred: pred, trusted: trusted, observed: observed} + if observed { + anyObserved = true + minPred = min(minPred, pred) + maxPred = max(maxPred, pred) // cold models seed to minPred, so they can never be the max + } + if trusted { + anyTrusted = true + } + } + + scores := make(map[datalayer.Model]float64, len(models)) + + // No model has a floor yet -> nothing to rank; explore all equally. + if !anyObserved { + for _, model := range models { + scores[model] = 1.0 + } + return scores + } + + for i, model := range models { + e := &evals[i] + if !e.observed { + e.pred = minPred // seed cold models at the best observed TTFT (optimistic) + } + if maxPred == minPred { + scores[model] = 1.0 + } else { + scores[model] = (maxPred - e.pred) / (maxPred - minPred) + } + // Independent coin per under-observed model: heads forces a calibration probe, tails + // suppresses it so calibrated models take the traffic. Final score only — never feeds the + // normalization above, so a probe cannot distort the trusted models' scores. + if s.explorationRate > 0 && !e.trusted { + if rand.Float64() < s.explorationRate { + scores[model] = 1.0 + } else if anyTrusted { + scores[model] = 0 + } + } + } + + if dl := log.FromContext(ctx).V(logutil.DEBUG); dl.Enabled() { + for i, model := range models { + dl.Info("ttft-aware score", "model", model.GetName(), + "predictedTTFT", evals[i].pred, "score", scores[model], "trusted", evals[i].trusted) + } + } + + return scores +} + +// predict returns the model's predicted TTFT at its current inflight, and whether it is trusted +// (calibrated) and observed (has a floor). Cold -> (0, false, false); observed but uncalibrated -> +// the floor as an optimistic seed. See README.md for the curve and the admissibility checks. +func (s *TTFTAwareScorer) predict(m ttftpercentile.TTFTPercentileMetrics) (pred float64, trusted, observed bool) { + floor := m.Floor() + if floor == 0 { + return 0, false, false + } + if !(m.RecentN >= m.MinRequests && m.InflightAtHigh > 0 && m.HighTTFT > floor) { + return floor, false, true + } + + // The low point is admissible only if the two operating points are separated in load, ordered + // in latency, and above the floor — otherwise its segment slopes on noise or downwards. + useLow := m.InflightAtHigh-m.InflightAtLow >= s.minInflightGap && + m.HighTTFT > m.LowTTFT && + m.LowTTFT > floor && + m.InflightAtLow > 0 + + cur := float64(m.Requests) + switch { + case useLow && cur < m.InflightAtLow: + // (0, floor) -> low point. Extending the steeper high-load slope backwards instead would + // predict a draining-but-loaded pool at its idle latency. + pred = floor + cur*(m.LowTTFT-floor)/m.InflightAtLow + case useLow: + // low -> high point, extended beyond it. + pred = m.LowTTFT + (cur-m.InflightAtLow)*(m.HighTTFT-m.LowTTFT)/(m.InflightAtHigh-m.InflightAtLow) + default: + // Low point inadmissible: floor chord (0, floor) -> high point, extended beyond it. + pred = floor + cur*(m.HighTTFT-floor)/m.InflightAtHigh + } + return max(pred, floor), true, true +} + +// metricsFor reads the TTFT percentile metrics an extractor published for the model. +// A missing or malformed attribute yields the zero value, which reads as truly cold. +func metricsFor(model datalayer.Model) ttftpercentile.TTFTPercentileMetrics { + if val, err := datalayer.ReadAttributeKey[ttftpercentile.TTFTPercentileMetrics]( + model.GetAttributes(), ttftpercentile.AttributeKey, + ); err == nil { + return val + } + return ttftpercentile.TTFTPercentileMetrics{} +}