Skip to content

Commit cd109ab

Browse files
MicheleCampiMicheleCampi
committed
fix: remove mid-stream TPOT predictions from predicted-latency producer
The mid-stream prediction path in processTokenForLatencyPrediction was observability-only and low quality: computed from the scheduling-time metrics snapshot, averaged together with the scheduling-time TPOT prediction, and the only remaining caller of buildPredictionRequest using the scraped RunningRequestsSize gauge (train/serve inconsistency). avgPredictedTPOT now reduces to the scheduling-time prediction, which is the meaningful predicted-vs-actual comparison. The decodeTokenSampler becomes dead code with this removal (its only remaining consumer was the removed prediction path) and is deleted, along with the write-only tpotObservations field. The samplingMean and maxDecodeTokenSamplesForPrediction config parameters are kept in Config for compatibility with existing configs (the strict decoder would otherwise reject them) but are ignored, with a deprecation log when set. Fixes #2016 Signed-off-by: Michele Campi <215741962+MicheleCampi@users.noreply.github.com>
1 parent 1a297c4 commit cd109ab

6 files changed

Lines changed: 16 additions & 294 deletions

File tree

‎pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/decode_token_sampler.go‎

Lines changed: 0 additions & 111 deletions
This file was deleted.

‎pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/decode_token_sampler_test.go‎

Lines changed: 0 additions & 96 deletions
This file was deleted.

‎pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go‎

Lines changed: 7 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,10 @@ func (pl *PredictedLatency) decrementEndpointCounter(m *sync.Map, key string, de
117117
}
118118

119119
type Config struct {
120-
SamplingMean float64 `json:"samplingMean,omitempty"`
120+
// Deprecated: no longer used. Mid-stream TPOT predictions were removed;
121+
// the value is accepted for config compatibility and ignored.
122+
SamplingMean float64 `json:"samplingMean,omitempty"`
123+
// Deprecated: no longer used. Accepted for config compatibility and ignored.
121124
MaxDecodeTokenSamplesForPrediction int `json:"maxDecodeTokenSamplesForPrediction,omitempty"`
122125
SLOBufferFactor float64 `json:"sloBufferFactor,omitempty"`
123126
ContextTTL time.Duration `json:"contextTTL,omitempty"`
@@ -155,6 +158,9 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
155158
if handle == nil {
156159
return nil, errors.New("plugin handle is required")
157160
}
161+
if parameters.SamplingMean != DefaultConfig.SamplingMean || parameters.MaxDecodeTokenSamplesForPrediction != DefaultConfig.MaxDecodeTokenSamplesForPrediction {
162+
log.FromContext(handle.Context()).Info("Deprecated: samplingMean and maxDecodeTokenSamplesForPrediction are ignored; mid-stream TPOT predictions were removed")
163+
}
158164
if err := registerMetrics(handle.Metrics()); err != nil {
159165
return nil, err
160166
}
@@ -170,14 +176,6 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
170176
func (c *Config) validate() error {
171177
var errs []error
172178

173-
if c.SamplingMean <= 0 {
174-
errs = append(errs, fmt.Errorf("samplingMean must be > 0, got %f", c.SamplingMean))
175-
}
176-
177-
if c.MaxDecodeTokenSamplesForPrediction < 0 {
178-
errs = append(errs, fmt.Errorf("maxDecodeTokenSamplesForPrediction must be >= 0, got %d", c.MaxDecodeTokenSamplesForPrediction))
179-
}
180-
181179
if c.SLOBufferFactor <= 0 {
182180
errs = append(errs, fmt.Errorf("sloBufferFactor must be > 0, got %f", c.SLOBufferFactor))
183181
}
@@ -265,8 +263,6 @@ type predictedLatencyCtx struct {
265263
predictedTTFT float64
266264
avgTPOT float64
267265
avgPredictedTPOT float64
268-
decodeTokenSampler *decodeTokenSampler
269-
tpotObservations []float64
270266
predictedTPOTObservations []float64
271267

272268
inputTokenCount int

‎pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin_test.go‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -365,16 +365,10 @@ func TestPredictedLatencyFactory(t *testing.T) {
365365
expectErr: false,
366366
},
367367
{
368-
name: "invalid samplingMean <= 0",
369-
pluginName: "bad-sampling-mean",
370-
jsonParams: `{"samplingMean": -1.0}`,
371-
expectErr: true,
372-
},
373-
{
374-
name: "invalid maxSampledTokens < 0",
375-
pluginName: "bad-max-tokens",
376-
jsonParams: `{"maxDecodeTokenSamplesForPrediction": -1}`,
377-
expectErr: true,
368+
name: "deprecated sampling params are accepted and ignored",
369+
pluginName: "deprecated-sampling-params",
370+
jsonParams: `{"samplingMean": -1.0, "maxDecodeTokenSamplesForPrediction": -1}`,
371+
expectErr: false,
378372
},
379373
{
380374
name: "invalid sloBufferFactor <= 0",

‎pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go‎

Lines changed: 5 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched.
138138

139139
if predictedLatencyCtx.ttft == 0 {
140140
if pl.config.StreamingMode && !response.EndOfStream {
141-
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
141+
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
142142

143143
// Only decrement if PreRequest actually incremented the prefill pod counter.
144144
// If Produce timed out, PreRequest may have skipped incrementing, and
@@ -149,13 +149,13 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched.
149149
}
150150
}
151151
} else {
152-
processTokenForLatencyPrediction(ctx, pl.typedName.Name, pl.typedName.Type, pl.latencypredictor, pl.config.EndpointRoleLabel, predictedLatencyCtx, targetMetadata, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
152+
processTokenForLatencyPrediction(ctx, predictedLatencyCtx, now)
153153
}
154154

155155
if response.EndOfStream {
156156
ttftNotYetRecorded := predictedLatencyCtx.ttft == 0
157157
if !pl.config.StreamingMode {
158-
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
158+
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
159159
}
160160

161161
if predictedLatencyCtx.ttft > 0 {
@@ -257,12 +257,9 @@ func processFirstTokenForLatencyPrediction(
257257
endpointRoleLabel string,
258258
predictedLatencyCtx *predictedLatencyCtx,
259259
now time.Time,
260-
samplingMean float64,
261-
maxDecodeTokenSamplesForPrediction int,
262260
) {
263261
logger := log.FromContext(ctx)
264262

265-
initializeSampler(ctx, predictedLatencyCtx, samplingMean, maxDecodeTokenSamplesForPrediction)
266263
predictedLatencyCtx.ttft = float64(now.Sub(predictedLatencyCtx.requestReceivedTimestamp).Milliseconds())
267264
predictedLatencyCtx.generatedTokenCount = 1
268265

@@ -296,15 +293,6 @@ func processFirstTokenForLatencyPrediction(
296293
refreshLastSeenMetrics(ctx, predictedLatencyCtx)
297294
}
298295

299-
func initializeSampler(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx, samplingMean float64, maxDecodeTokenSamplesForPrediction int) {
300-
if predictedLatencyCtx.decodeTokenSampler == nil {
301-
logger := log.FromContext(ctx)
302-
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
303-
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
304-
logger.V(logutil.DEBUG).Info("Initialized token sampler for first token", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
305-
}
306-
}
307-
308296
func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx) {
309297
logger := log.FromContext(ctx)
310298
targetName := predictedLatencyCtx.targetMetadata.NamespacedName.Name
@@ -319,67 +307,20 @@ func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatency
319307
}
320308
}
321309

322-
// processTokenForLatencyPrediction records actual inter-token latency, sampled predictions, and advances timestamp.
310+
// processTokenForLatencyPrediction records actual inter-token latency and advances the timestamp.
323311
func processTokenForLatencyPrediction(
324312
ctx context.Context,
325-
pluginName, pluginType string,
326-
predictor latencypredictor.PredictorInterface,
327-
endpointRoleLabel string,
328313
predictedLatencyCtx *predictedLatencyCtx,
329-
targetEndpointMetadata *fwkdl.EndpointMetadata,
330314
now time.Time,
331-
samplingMean float64,
332-
maxDecodeTokenSamplesForPrediction int,
333315
) {
334316
logger := log.FromContext(ctx)
335317

336-
if predictedLatencyCtx.decodeTokenSampler == nil {
337-
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
338-
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
339-
logger.V(logutil.DEBUG).Info("Initialized token sampler for subsequent tokens", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
340-
}
341-
342318
latencyMs := float64(now.Sub(predictedLatencyCtx.lastTokenTimestamp).Milliseconds())
343319
predictedLatencyCtx.generatedTokenCount++
344320

345-
if predictedLatencyCtx.generatedTokenCount == 2 || predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
346-
predictedLatencyCtx.tpotObservations = append(predictedLatencyCtx.tpotObservations, latencyMs)
347-
}
348321
if predictedLatencyCtx.generatedTokenCount == 2 {
349322
logger.V(logutil.DEBUG).Info("First inter-token latency observed",
350-
"actual_tpot_ms", latencyMs,
351-
"predicted_tpot_ms", predictedLatencyCtx.avgPredictedTPOT)
352-
}
353-
354-
m, err := getLatestMetricsForProfile(predictedLatencyCtx, "")
355-
if err != nil {
356-
logger.V(logutil.DEBUG).Info("Skipping TPOT prediction due to missing metrics or schedulingResult", "error", err)
357-
return
358-
}
359-
360-
if predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
361-
in := buildPredictionRequest(
362-
endpointRoleLabel,
363-
targetEndpointMetadata,
364-
m,
365-
predictedLatencyCtx.inputTokenCount,
366-
predictedLatencyCtx.generatedTokenCount,
367-
0,
368-
)
369-
start := time.Now()
370-
p, err := predictor.Predict(ctx, in)
371-
dur := time.Since(start)
372-
if err != nil || p == nil {
373-
logger.V(logutil.DEBUG).Error(err, "TPOT predict failed", "duration_ms", dur.Milliseconds())
374-
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, 0)
375-
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, 0, len(predictedLatencyCtx.predictedTPOTObservations))
376-
} else {
377-
logger.V(logutil.DEBUG).Info("TPOT predict succeeded", "value_ms", p.TPOT, "duration_ms", dur.Milliseconds())
378-
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, p.TPOT)
379-
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, p.TPOT, len(predictedLatencyCtx.predictedTPOTObservations))
380-
}
381-
recordRequestTPOTPredictionDuration(ctx, pluginName, pluginType, predictedLatencyCtx.schedulingRequest.TargetModel, predictedLatencyCtx.incomingModelName, dur.Seconds())
382-
predictedLatencyCtx.decodeTokenSampler.recordPrediction(predictedLatencyCtx.generatedTokenCount)
323+
"actual_tpot_ms", latencyMs)
383324
}
384325

385326
predictedLatencyCtx.lastTokenTimestamp = now

0 commit comments

Comments
 (0)