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>

Author: MicheleCampi

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

@@ -225,7 +225,10 @@ func (pl *PredictedLatency) readInFlightLoad(endpoint fwksched.Endpoint) inFligh
 }
 
 type Config struct {
-	SamplingMean                       float64       `json:"samplingMean,omitempty"`
+	// Deprecated: no longer used. Mid-stream TPOT predictions were removed;
+	// the value is accepted for config compatibility and ignored.
+	SamplingMean float64 `json:"samplingMean,omitempty"`
+	// Deprecated: no longer used. Accepted for config compatibility and ignored.
 	MaxDecodeTokenSamplesForPrediction int           `json:"maxDecodeTokenSamplesForPrediction,omitempty"`
 	SLOBufferFactor                    float64       `json:"sloBufferFactor,omitempty"`
 	ContextTTL                         time.Duration `json:"contextTTL,omitempty"`
@@ -267,6 +270,9 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
 	if handle == nil {
 		return nil, errors.New("plugin handle is required")
 	}
+	if parameters.SamplingMean != DefaultConfig.SamplingMean || parameters.MaxDecodeTokenSamplesForPrediction != DefaultConfig.MaxDecodeTokenSamplesForPrediction {
+		log.FromContext(handle.Context()).Info("Deprecated: samplingMean and maxDecodeTokenSamplesForPrediction are ignored; mid-stream TPOT predictions were removed")
+	}
 	if err := registerMetrics(handle.Metrics()); err != nil {
 		return nil, err
 	}
@@ -282,14 +288,6 @@ func PredictedLatencyFactory(name string, rawParameters *json.Decoder, handle pl
 func (c *Config) validate() error {
 	var errs []error
 
-	if c.SamplingMean <= 0 {
-		errs = append(errs, fmt.Errorf("samplingMean must be > 0, got %f", c.SamplingMean))
-	}
-
-	if c.MaxDecodeTokenSamplesForPrediction < 0 {
-		errs = append(errs, fmt.Errorf("maxDecodeTokenSamplesForPrediction must be >= 0, got %d", c.MaxDecodeTokenSamplesForPrediction))
-	}
-
 	if c.SLOBufferFactor <= 0 {
 		errs = append(errs, fmt.Errorf("sloBufferFactor must be > 0, got %f", c.SLOBufferFactor))
 	}
@@ -370,8 +368,6 @@ type predictedLatencyCtx struct {
 	predictedTTFT             float64
 	avgTPOT                   float64
 	avgPredictedTPOT          float64
-	decodeTokenSampler        *decodeTokenSampler
-	tpotObservations          []float64
 	predictedTPOTObservations []float64
 
 	inputTokenCount int

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

@@ -391,16 +391,10 @@ func TestPredictedLatencyFactory(t *testing.T) {
 			expectErr:  false,
 		},
 		{
-			name:       "invalid samplingMean <= 0",
-			pluginName: "bad-sampling-mean",
-			jsonParams: `{"samplingMean": -1.0}`,
-			expectErr:  true,
-		},
-		{
-			name:       "invalid maxSampledTokens < 0",
-			pluginName: "bad-max-tokens",
-			jsonParams: `{"maxDecodeTokenSamplesForPrediction": -1}`,
-			expectErr:  true,
+			name:       "deprecated sampling params are accepted and ignored",
+			pluginName: "deprecated-sampling-params",
+			jsonParams: `{"samplingMean": -1.0, "maxDecodeTokenSamplesForPrediction": -1}`,
+			expectErr:  false,
 		},
 		{
 			name:       "invalid sloBufferFactor <= 0",

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

@@ -149,15 +149,15 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched.
 
 	if predictedLatencyCtx.ttft == 0 {
 		if pl.config.StreamingMode && !response.EndOfStream {
-			processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
+			processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
 		}
 	} else {
-		processTokenForLatencyPrediction(ctx, pl.typedName.Name, pl.typedName.Type, pl.latencypredictor, pl.config.EndpointRoleLabel, predictedLatencyCtx, targetMetadata, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
+		processTokenForLatencyPrediction(ctx, predictedLatencyCtx, now)
 	}
 
 	if response.EndOfStream {
 		if !pl.config.StreamingMode {
-			processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now, pl.config.SamplingMean, pl.config.MaxDecodeTokenSamplesForPrediction)
+			processFirstTokenForLatencyPrediction(ctx, pl.latencypredictor, pl.config.StreamingMode, pl.config.EndpointRoleLabel, predictedLatencyCtx, now)
 		}
 
 		if predictedLatencyCtx.ttft > 0 {
@@ -247,12 +247,9 @@ func processFirstTokenForLatencyPrediction(
 	endpointRoleLabel string,
 	predictedLatencyCtx *predictedLatencyCtx,
 	now time.Time,
-	samplingMean float64,
-	maxDecodeTokenSamplesForPrediction int,
 ) {
 	logger := log.FromContext(ctx)
 
-	initializeSampler(ctx, predictedLatencyCtx, samplingMean, maxDecodeTokenSamplesForPrediction)
 	predictedLatencyCtx.ttft = float64(now.Sub(predictedLatencyCtx.requestReceivedTimestamp).Milliseconds())
 	predictedLatencyCtx.generatedTokenCount = 1
 
@@ -286,15 +283,6 @@ func processFirstTokenForLatencyPrediction(
 	refreshLastSeenMetrics(ctx, predictedLatencyCtx)
 }
 
-func initializeSampler(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx, samplingMean float64, maxDecodeTokenSamplesForPrediction int) {
-	if predictedLatencyCtx.decodeTokenSampler == nil {
-		logger := log.FromContext(ctx)
-		requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
-		predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
-		logger.V(logutil.DEBUG).Info("Initialized token sampler for first token", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
-	}
-}
-
 func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx) {
 	logger := log.FromContext(ctx)
 	targetName := predictedLatencyCtx.targetMetadata.NamespacedName.Name
@@ -309,67 +297,20 @@ func predictFirstTPOT(ctx context.Context, predictedLatencyCtx *predictedLatency
 	}
 }
 
-// processTokenForLatencyPrediction records actual inter-token latency, sampled predictions, and advances timestamp.
+// processTokenForLatencyPrediction records the actual TPOT for the token and advances the timestamp.
 func processTokenForLatencyPrediction(
 	ctx context.Context,
-	pluginName, pluginType string,
-	predictor latencypredictor.PredictorInterface,
-	endpointRoleLabel string,
 	predictedLatencyCtx *predictedLatencyCtx,
-	targetEndpointMetadata *fwkdl.EndpointMetadata,
 	now time.Time,
-	samplingMean float64,
-	maxDecodeTokenSamplesForPrediction int,
 ) {
 	logger := log.FromContext(ctx)
 
-	if predictedLatencyCtx.decodeTokenSampler == nil {
-		requestID := predictedLatencyCtx.schedulingRequest.Headers[reqcommon.RequestIDHeaderKey]
-		predictedLatencyCtx.decodeTokenSampler = newDecodeTokenSampler(requestID, samplingMean, maxDecodeTokenSamplesForPrediction)
-		logger.V(logutil.DEBUG).Info("Initialized token sampler for subsequent tokens", "request_id", requestID, "next_prediction_token", predictedLatencyCtx.decodeTokenSampler.getNextSampleToken())
-	}
-
 	latencyMs := float64(now.Sub(predictedLatencyCtx.lastTokenTimestamp).Milliseconds())
 	predictedLatencyCtx.generatedTokenCount++
 
-	if predictedLatencyCtx.generatedTokenCount == 2 || predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
-		predictedLatencyCtx.tpotObservations = append(predictedLatencyCtx.tpotObservations, latencyMs)
-	}
 	if predictedLatencyCtx.generatedTokenCount == 2 {
 		logger.V(logutil.DEBUG).Info("First inter-token latency observed",
-			"actual_tpot_ms", latencyMs,
-			"predicted_tpot_ms", predictedLatencyCtx.avgPredictedTPOT)
-	}
-
-	m, err := getLatestMetricsForProfile(predictedLatencyCtx, "")
-	if err != nil {
-		logger.V(logutil.DEBUG).Info("Skipping TPOT prediction due to missing metrics or schedulingResult", "error", err)
-		return
-	}
-
-	if predictedLatencyCtx.decodeTokenSampler.shouldPredict(predictedLatencyCtx.generatedTokenCount) {
-		in := buildPredictionRequest(
-			endpointRoleLabel,
-			targetEndpointMetadata,
-			m,
-			predictedLatencyCtx.inputTokenCount,
-			predictedLatencyCtx.generatedTokenCount,
-			0,
-		)
-		start := time.Now()
-		p, err := predictor.Predict(ctx, in)
-		dur := time.Since(start)
-		if err != nil || p == nil {
-			logger.V(logutil.DEBUG).Error(err, "TPOT predict failed", "duration_ms", dur.Milliseconds())
-			predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, 0)
-			predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, 0, len(predictedLatencyCtx.predictedTPOTObservations))
-		} else {
-			logger.V(logutil.DEBUG).Info("TPOT predict succeeded", "value_ms", p.TPOT, "duration_ms", dur.Milliseconds())
-			predictedLatencyCtx.predictedTPOTObservations = append(predictedLatencyCtx.predictedTPOTObservations, p.TPOT)
-			predictedLatencyCtx.avgPredictedTPOT = calculateRunningAverage(predictedLatencyCtx.avgPredictedTPOT, p.TPOT, len(predictedLatencyCtx.predictedTPOTObservations))
-		}
-		recordRequestTPOTPredictionDuration(ctx, pluginName, pluginType, predictedLatencyCtx.schedulingRequest.TargetModel, predictedLatencyCtx.incomingModelName, dur.Seconds())
-		predictedLatencyCtx.decodeTokenSampler.recordPrediction(predictedLatencyCtx.generatedTokenCount)
+			"actual_tpot_ms", latencyMs)
 	}
 
 	predictedLatencyCtx.lastTokenTimestamp = now