Skip to content

Commit 5e7e43d

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> # Conflicts: # pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go
1 parent d632664 commit 5e7e43d

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
@@ -225,7 +225,10 @@ func (pl *PredictedLatency) readInFlightLoad(endpoint fwksched.Endpoint) inFligh
225225
}
226226

227227
type Config struct {
228-
SamplingMean float64 `json:"samplingMean,omitempty"`
228+
// Deprecated: no longer used. Mid-stream TPOT predictions were removed;
229+
// the value is accepted for config compatibility and ignored.
230+
SamplingMean float64 `json:"samplingMean,omitempty"`
231+
// Deprecated: no longer used. Accepted for config compatibility and ignored.
229232
MaxDecodeTokenSamplesForPrediction int `json:"maxDecodeTokenSamplesForPrediction,omitempty"`
230233
SLOBufferFactor float64 `json:"sloBufferFactor,omitempty"`
231234
ContextTTL time.Duration `json:"contextTTL,omitempty"`
@@ -267,6 +270,9 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
267270
if handle == nil {
268271
return nil, errors.New("plugin handle is required")
269272
}
273+
if parameters.SamplingMean != DefaultConfig.SamplingMean || parameters.MaxDecodeTokenSamplesForPrediction != DefaultConfig.MaxDecodeTokenSamplesForPrediction {
274+
log.FromContext(handle.Context()).Info("Deprecated: samplingMean and maxDecodeTokenSamplesForPrediction are ignored; mid-stream TPOT predictions were removed")
275+
}
270276
if err := registerMetrics(handle.Metrics()); err != nil {
271277
return nil, err
272278
}
@@ -282,14 +288,6 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
282288
func (c *Config) validate() error {
283289
var errs []error
284290

285-
if c.SamplingMean <= 0 {
286-
errs = append(errs, fmt.Errorf("samplingMean must be > 0, got %f", c.SamplingMean))
287-
}
288-
289-
if c.MaxDecodeTokenSamplesForPrediction < 0 {
290-
errs = append(errs, fmt.Errorf("maxDecodeTokenSamplesForPrediction must be >= 0, got %d", c.MaxDecodeTokenSamplesForPrediction))
291-
}
292-
293291
if c.SLOBufferFactor <= 0 {
294292
errs = append(errs, fmt.Errorf("sloBufferFactor must be > 0, got %f", c.SLOBufferFactor))
295293
}
@@ -370,8 +368,6 @@ type predictedLatencyCtx struct {
370368
predictedTTFT float64
371369
avgTPOT float64
372370
avgPredictedTPOT float64
373-
decodeTokenSampler *decodeTokenSampler
374-
tpotObservations []float64
375371
predictedTPOTObservations []float64
376372

377373
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
@@ -391,16 +391,10 @@ func TestPredictedLatencyFactory(t *testing.T) {
391391
expectErr: false,
392392
},
393393
{
394-
name: "invalid samplingMean <= 0",
395-
pluginName: "bad-sampling-mean",
396-
jsonParams: `{"samplingMean": -1.0}`,
397-
expectErr: true,
398-
},
399-
{
400-
name: "invalid maxSampledTokens < 0",
401-
pluginName: "bad-max-tokens",
402-
jsonParams: `{"maxDecodeTokenSamplesForPrediction": -1}`,
403-
expectErr: true,
394+
name: "deprecated sampling params are accepted and ignored",
395+
pluginName: "deprecated-sampling-params",
396+
jsonParams: `{"samplingMean": -1.0, "maxDecodeTokenSamplesForPrediction": -1}`,
397+
expectErr: false,
404398
},
405399
{
406400
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
@@ -149,15 +149,15 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched.
149149

150150
if predictedLatencyCtx.ttft == 0 {
151151
if pl.config.StreamingMode && !response.EndOfStream {
152-
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
152+
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
153153
}
154154
} else {
155-
processTokenForLatencyPrediction(ctx, pl.typedName.Name, pl.typedName.Type, pl.latencypredictor, pl.config.EndpointRoleLabel, predictedLatencyCtx, targetMetadata, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
155+
processTokenForLatencyPrediction(ctx, predictedLatencyCtx, now)
156156
}
157157

158158
if response.EndOfStream {
159159
if !pl.config.StreamingMode {
160-
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
160+
processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
161161
}
162162

163163
if predictedLatencyCtx.ttft > 0 {
@@ -247,12 +247,9 @@ func processFirstTokenForLatencyPrediction(
247247
endpointRoleLabel string,
248248
predictedLatencyCtx *predictedLatencyCtx,
249249
now time.Time,
250-
samplingMean float64,
251-
maxDecodeTokenSamplesForPrediction int,
252250
) {
253251
logger := log.FromContext(ctx)
254252

255-
initializeSampler(ctx, predictedLatencyCtx, samplingMean, maxDecodeTokenSamplesForPrediction)
256253
predictedLatencyCtx.ttft = float64(now.Sub(predictedLatencyCtx.requestReceivedTimestamp).Milliseconds())
257254
predictedLatencyCtx.generatedTokenCount = 1
258255

@@ -286,15 +283,6 @@ func processFirstTokenForLatencyPrediction(
286283
refreshLastSeenMetrics(ctx, predictedLatencyCtx)
287284
}
288285

289-
func initializeSampler(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx, samplingMean float64, maxDecodeTokenSamplesForPrediction int) {
290-
if predictedLatencyCtx.decodeTokenSampler == nil {
291-
logger := log.FromContext(ctx)
292-
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
293-
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
294-
logger.V(logutil.DEBUG).Info("Initialized token sampler for first token", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
295-
}
296-
}
297-
298286
func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx) {
299287
logger := log.FromContext(ctx)
300288
targetName := predictedLatencyCtx.targetMetadata.NamespacedName.Name
@@ -309,67 +297,20 @@ func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatency
309297
}
310298
}
311299

312-
// processTokenForLatencyPrediction records actual inter-token latency, sampled predictions, and advances timestamp.
300+
// processTokenForLatencyPrediction records actual inter-token latency and advances the timestamp.
313301
func processTokenForLatencyPrediction(
314302
ctx context.Context,
315-
pluginName, pluginType string,
316-
predictor latencypredictor.PredictorInterface,
317-
endpointRoleLabel string,
318303
predictedLatencyCtx *predictedLatencyCtx,
319-
targetEndpointMetadata *fwkdl.EndpointMetadata,
320304
now time.Time,
321-
samplingMean float64,
322-
maxDecodeTokenSamplesForPrediction int,
323305
) {
324306
logger := log.FromContext(ctx)
325307

326-
if predictedLatencyCtx.decodeTokenSampler == nil {
327-
requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
328-
predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
329-
logger.V(logutil.DEBUG).Info("Initialized token sampler for subsequent tokens", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
330-
}
331-
332308
latencyMs := float64(now.Sub(predictedLatencyCtx.lastTokenTimestamp).Milliseconds())
333309
predictedLatencyCtx.generatedTokenCount++
334310

335-
if predictedLatencyCtx.generatedTokenCount == 2 || predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
336-
predictedLatencyCtx.tpotObservations = append(predictedLatencyCtx.tpotObservations, latencyMs)
337-
}
338311
if predictedLatencyCtx.generatedTokenCount == 2 {
339312
logger.V(logutil.DEBUG).Info("First inter-token latency observed",
340-
"actual_tpot_ms", latencyMs,
341-
"predicted_tpot_ms", predictedLatencyCtx.avgPredictedTPOT)
342-
}
343-
344-
m, err := getLatestMetricsForProfile(predictedLatencyCtx, "")
345-
if err != nil {
346-
logger.V(logutil.DEBUG).Info("Skipping TPOT prediction due to missing metrics or schedulingResult", "error", err)
347-
return
348-
}
349-
350-
if predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
351-
in := buildPredictionRequest(
352-
endpointRoleLabel,
353-
targetEndpointMetadata,
354-
m,
355-
predictedLatencyCtx.inputTokenCount,
356-
predictedLatencyCtx.generatedTokenCount,
357-
0,
358-
)
359-
start := time.Now()
360-
p, err := predictor.Predict(ctx, in)
361-
dur := time.Since(start)
362-
if err != nil || p == nil {
363-
logger.V(logutil.DEBUG).Error(err, "TPOT predict failed", "duration_ms", dur.Milliseconds())
364-
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, 0)
365-
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, 0, len(predictedLatencyCtx.predictedTPOTObservations))
366-
} else {
367-
logger.V(logutil.DEBUG).Info("TPOT predict succeeded", "value_ms", p.TPOT, "duration_ms", dur.Milliseconds())
368-
predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, p.TPOT)
369-
predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, p.TPOT, len(predictedLatencyCtx.predictedTPOTObservations))
370-
}
371-
recordRequestTPOTPredictionDuration(ctx, pluginName, pluginType, predictedLatencyCtx.schedulingRequest.TargetModel, predictedLatencyCtx.incomingModelName, dur.Seconds())
372-
predictedLatencyCtx.decodeTokenSampler.recordPrediction(predictedLatencyCtx.generatedTokenCount)
313+
"actual_tpot_ms", latencyMs)
373314
}
374315

375316
predictedLatencyCtx.lastTokenTimestamp = now

0 commit comments

Comments
 (0)