diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/README.md b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/README.md index a17c25286e..a530425a8c 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/README.md +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/README.md @@ -16,6 +16,8 @@ DataProducer, PreRequest, ResponseHeader, ResponseBody, ProducerPlugin, Consumer - TPOT training data collection at EOS (streaming mode) - Per-endpoint running request queue tracking (TPOT SLO priority queue) - Prefix cache score forwarding from `PrefixCacheMatchInfo` attributes +- Multimodal encoder-cache size forwarding from `EncoderCacheMatchInfo` attributes + (opt-in via `useEncoderCacheFeatures`) - TPOT neutralization for prefill endpoints in disaggregated serving - E2E latency metrics when `streamingMode=false` @@ -30,6 +32,8 @@ DataProducer, PreRequest, ResponseHeader, ResponseBody, ProducerPlugin, Consumer | `streamingMode` | `false` | Record TTFT on first chunk (true) vs EOS (false) | | `endpointRoleLabel` | `""` | Label key for disaggregated serving roles | | `predictInProduce` | `true` | Enable/disable bulk predictions. Set false for training-only mode | +| `useEncoderCacheFeatures` | `false` | Feed multimodal encoder-cache sizes (`encoder_input_size`, `encoder_matched_size`) to the predictor. Requires (and auto-creates) a multimodal encoder-cache producer | +| `encoderCacheMatchInfoProducerName` | `""` | Multimodal encoder-cache producer to read match data from. Empty defaults to the auto-created producer | ## Default Behavior (`streamingMode: false`) diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks.go index f1c556de2f..f6aadf34e9 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks.go @@ -28,6 +28,7 @@ import ( fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" attrconcurrency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/concurrency" attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency" + attrmm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/multimodal" attrprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/prefix" tokenproducer "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/tokenizer" ) @@ -59,6 +60,10 @@ func (pl *PredictedLatency) Produce(ctx context.Context, request *fwksched.Infer } predictedLatencyCtx.prefixCacheScoresForEndpoints[endpoint.GetMetadata().NamespacedName.Name] = prefixCacheScore + if pl.config.UseEncoderCacheFeatures { + pl.captureEncoderCacheSizes(ctx, predictedLatencyCtx, endpoint) + } + // Capture the in-flight load here, in the DAG-ordered Produce hook, and // reuse it for both the prediction features and the dispatch-time training // features. Re-reading the live attribute in PreRequest would make the @@ -114,6 +119,48 @@ func (pl *PredictedLatency) Produce(ctx context.Context, request *fwksched.Infer return nil } +// captureEncoderCacheSizes stores the request's total multimodal encoder item +// size and the endpoint's matched portion in the request context. The match +// data is attached by the multimodal encoder-cache producer, which the DAG +// orders before this plugin; absence means a text-only request and leaves the +// sizes at 0. +func (pl *PredictedLatency) captureEncoderCacheSizes(ctx context.Context, predictedLatencyCtx *predictedLatencyCtx, endpoint fwksched.Endpoint) { + logger := log.FromContext(ctx) + raw, ok := endpoint.Get(pl.encoderCacheDataKey.String()) + if !ok { + return + } + matchInfo, ok := raw.(*attrmm.EncoderCacheMatchInfo) + if !ok || matchInfo == nil { + return + } + inputSize := sumItemSizes(matchInfo.RequestItems()) + matchedSize := sumItemSizes(matchInfo.MatchedItems()) + // The predictor rejects matched > input, so clamp inconsistent match data + // rather than dropping the whole prediction or training entry. + if matchedSize > inputSize { + logger.V(logutil.DEBUG).Info("Encoder cache matched size exceeds input size, clamping", + "pod", endpoint.GetMetadata().String(), + "encoderInputSize", inputSize, + "encoderMatchedSize", matchedSize) + matchedSize = inputSize + } + predictedLatencyCtx.encoderInputSize = inputSize + predictedLatencyCtx.encoderMatchedSizeForEndpoints[endpoint.GetMetadata().NamespacedName.Name] = matchedSize + logger.V(logutil.DEBUG).Info("Encoder cache sizes for pod", + "pod", endpoint.GetMetadata().String(), + "encoderInputSize", inputSize, + "encoderMatchedSize", matchedSize) +} + +func sumItemSizes(items []attrmm.MatchItem) int { + total := 0 + for _, item := range items { + total += item.Size + } + return total +} + func (pl *PredictedLatency) Produces() map[plugin.DataKey]any { return map[plugin.DataKey]any{ pl.latencyPredictionInfoDataKey: attrlatency.LatencyPredictionInfo{}, @@ -121,11 +168,15 @@ func (pl *PredictedLatency) Produces() map[plugin.DataKey]any { } func (pl *PredictedLatency) Consumes() plugin.DataDependencies { - return plugin.DataDependencies{ - Required: map[plugin.DataKey]any{ - pl.prefixMatchDataKey: attrprefix.PrefixCacheMatchInfo{}, - pl.inFlightLoadDataKey: attrconcurrency.InFlightLoad{}, - tokenproducer.TokenizedPromptDataKey: fwksched.TokenizedPrompt{}, - }, + required := map[plugin.DataKey]any{ + pl.prefixMatchDataKey: attrprefix.PrefixCacheMatchInfo{}, + pl.inFlightLoadDataKey: attrconcurrency.InFlightLoad{}, + tokenproducer.TokenizedPromptDataKey: fwksched.TokenizedPrompt{}, + } + // Required (not Optional) because only Required dependencies create DAG + // ordering edges; the encoder-cache producer must run before this plugin. + if pl.config.UseEncoderCacheFeatures { + required[pl.encoderCacheDataKey] = attrmm.EncoderCacheMatchInfo{} } + return plugin.DataDependencies{Required: required} } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks_test.go index 20e716c478..9b3d684857 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks_test.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/dataproducer_hooks_test.go @@ -21,9 +21,11 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency" + attrmm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/multimodal" attrprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/prefix" ) @@ -36,6 +38,88 @@ func TestProducesConsumes(t *testing.T) { consumes := pl.Consumes() assert.Contains(t, consumes.Required, attrprefix.PrefixCacheMatchInfoDataKey) + assert.NotContains(t, consumes.Required, attrmm.EncoderCacheMatchInfoKey, + "encoder-cache match data must not be consumed when the feature is disabled") +} + +func TestConsumes_EncoderCacheFeatureEnabled(t *testing.T) { + cfg := DefaultConfig + cfg.UseEncoderCacheFeatures = true + pl := NewPredictedLatency(LatencyDataProviderPluginType, cfg, nil) + + consumes := pl.Consumes() + assert.Contains(t, consumes.Required, attrmm.EncoderCacheMatchInfoKey) +} + +// TestProduce_CapturesEncoderCacheSizes verifies that Produce reads the +// multimodal encoder-cache match data attached to endpoints and captures the +// request's input size plus the per-endpoint matched size, leaving endpoints +// without match data (text-only requests) at 0. +func TestProduce_CapturesEncoderCacheSizes(t *testing.T) { + cfg := DefaultConfig + cfg.PredictInProduce = false + cfg.UseEncoderCacheFeatures = true + pl := NewPredictedLatency(LatencyDataProviderPluginType, cfg, nil) + + request := createTestInferenceRequest("encoder-test", 0, 0) + matched := createTestEndpoint("pod-matched", 0.1, 0, 0) + unmatched := createTestEndpoint("pod-unmatched", 0.1, 0, 0) + + items := []attrmm.MatchItem{{Hash: "img-a", Size: 1}, {Hash: "img-b", Size: 1}} + matched.Put(pl.encoderCacheDataKey.String(), attrmm.NewEncoderCacheMatchInfo(items[:1], items)) + unmatched.Put(pl.encoderCacheDataKey.String(), attrmm.NewEncoderCacheMatchInfo(nil, items)) + + require.NoError(t, pl.Produce(context.Background(), request, []fwksched.Endpoint{matched, unmatched})) + + plCtx, err := pl.getPredictedLatencyContextForRequest(request) + require.NoError(t, err) + assert.Equal(t, 2, plCtx.encoderInputSize) + assert.Equal(t, 1, plCtx.encoderMatchedSizeForEndpoints["pod-matched"]) + assert.Equal(t, 0, plCtx.encoderMatchedSizeForEndpoints["pod-unmatched"]) +} + +// TestProduce_ClampsInconsistentEncoderMatchData verifies that match data +// whose matched size exceeds the input size is clamped so downstream +// predictor validation (matched <= input) cannot reject the request. +func TestProduce_ClampsInconsistentEncoderMatchData(t *testing.T) { + cfg := DefaultConfig + cfg.PredictInProduce = false + cfg.UseEncoderCacheFeatures = true + pl := NewPredictedLatency(LatencyDataProviderPluginType, cfg, nil) + + request := createTestInferenceRequest("encoder-clamp-test", 0, 0) + endpoint := createTestEndpoint("pod-a", 0.1, 0, 0) + + matched := []attrmm.MatchItem{{Hash: "img-a", Size: 3}} + requestItems := []attrmm.MatchItem{{Hash: "img-b", Size: 1}} + endpoint.Put(pl.encoderCacheDataKey.String(), attrmm.NewEncoderCacheMatchInfo(matched, requestItems)) + + require.NoError(t, pl.Produce(context.Background(), request, []fwksched.Endpoint{endpoint})) + + plCtx, err := pl.getPredictedLatencyContextForRequest(request) + require.NoError(t, err) + assert.Equal(t, 1, plCtx.encoderInputSize) + assert.Equal(t, 1, plCtx.encoderMatchedSizeForEndpoints["pod-a"]) +} + +// TestProduce_EncoderCacheFeatureDisabledIgnoresMatchData is the negative +// control: with the feature off, attached match data is not read. +func TestProduce_EncoderCacheFeatureDisabledIgnoresMatchData(t *testing.T) { + cfg := DefaultConfig + cfg.PredictInProduce = false + pl := NewPredictedLatency(LatencyDataProviderPluginType, cfg, nil) + + request := createTestInferenceRequest("encoder-disabled-test", 0, 0) + endpoint := createTestEndpoint("pod-a", 0.1, 0, 0) + items := []attrmm.MatchItem{{Hash: "img-a", Size: 1}} + endpoint.Put(pl.encoderCacheDataKey.String(), attrmm.NewEncoderCacheMatchInfo(items, items)) + + require.NoError(t, pl.Produce(context.Background(), request, []fwksched.Endpoint{endpoint})) + + plCtx, err := pl.getPredictedLatencyContextForRequest(request) + require.NoError(t, err) + assert.Equal(t, 0, plCtx.encoderInputSize) + assert.Empty(t, plCtx.encoderMatchedSizeForEndpoints) } // TestProduce_CancelledContextDoesNotPublish verifies that when the diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction.go index 04f2a775d1..6f41c13a2e 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction.go @@ -167,13 +167,16 @@ func (p *Predictor) predictBayesianRidge(req PredictionRequest, mr *MetricsRespo } c := mr.Coefficients - // Updated linear combination for TTFT to include prefix_cache_score + // Linear combination for TTFT; encoder-cache coefficients are absent from + // models trained without multimodal features and contribute 0 in that case. ttft := c.TTFTIntercept + c.TTFTCoeffs["kv_cache_percentage"]*req.KVCachePercentage + c.TTFTCoeffs["input_token_length"]*float64(req.InputTokenLength) + c.TTFTCoeffs["num_request_waiting"]*float64(req.NumRequestWaiting) + c.TTFTCoeffs["num_request_running"]*float64(req.NumRequestRunning) + - c.TTFTCoeffs["prefix_cache_score"]*req.PrefixCacheScore + c.TTFTCoeffs["prefix_cache_score"]*req.PrefixCacheScore + + c.TTFTCoeffs["encoder_input_size"]*float64(req.EncoderInputSize) + + c.TTFTCoeffs["encoder_matched_size"]*float64(req.EncoderMatchedSize) // Linear combination for TPOT (remains unchanged - no prefix cache effect) tpot := c.TPOTIntercept + @@ -252,6 +255,12 @@ func (p *Predictor) ValidatePredictionRequest(req PredictionRequest) error { if req.PrefixCacheScore < 0.0 || req.PrefixCacheScore > 1.0 { return fmt.Errorf("prefix_cache_score must be between 0.0 and 1.0, got %f", req.PrefixCacheScore) } + if req.EncoderInputSize < 0 { + return fmt.Errorf("encoder_input_size must be non-negative, got %d", req.EncoderInputSize) + } + if req.EncoderMatchedSize < 0 || req.EncoderMatchedSize > req.EncoderInputSize { + return fmt.Errorf("encoder_matched_size must be between 0 and encoder_input_size (%d), got %d", req.EncoderInputSize, req.EncoderMatchedSize) + } return nil } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction_test.go new file mode 100644 index 0000000000..4dbae6402b --- /dev/null +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/prediction_test.go @@ -0,0 +1,118 @@ +/* +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 latencypredictorclient + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func validEncoderPredictionRequest() PredictionRequest { + return PredictionRequest{ + KVCachePercentage: 0.5, + InputTokenLength: 100, + NumRequestWaiting: 1, + NumRequestRunning: 1, + NumTokensGenerated: 10, + } +} + +func TestValidatePredictionRequest_EncoderSizes(t *testing.T) { + predictor := &Predictor{} + + tests := []struct { + name string + inputSize int + matchedSize int + wantErr bool + }{ + {name: "both zero", inputSize: 0, matchedSize: 0, wantErr: false}, + {name: "partial match", inputSize: 5, matchedSize: 3, wantErr: false}, + {name: "full match", inputSize: 5, matchedSize: 5, wantErr: false}, + {name: "negative input size", inputSize: -1, matchedSize: 0, wantErr: true}, + {name: "negative matched size", inputSize: 5, matchedSize: -1, wantErr: true}, + {name: "matched exceeds input", inputSize: 5, matchedSize: 6, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + req := validEncoderPredictionRequest() + req.EncoderInputSize = tt.inputSize + req.EncoderMatchedSize = tt.matchedSize + err := predictor.ValidatePredictionRequest(req) + if tt.wantErr { + assert.Error(t, err) + } else { + assert.NoError(t, err) + } + }) + } +} + +func TestValidateTrainingEntry_EncoderSizes(t *testing.T) { + predictor := &Predictor{} + entry := TrainingEntry{ + KVCachePercentage: 0.5, + InputTokenLength: 100, + NumRequestWaiting: 1, + NumRequestRunning: 1, + NumTokensGenerated: 10, + ActualTTFT: 50.0, + ActualTPOT: 5.0, + } + + entry.EncoderInputSize, entry.EncoderMatchedSize = 4, 2 + assert.NoError(t, predictor.ValidateTrainingEntry(entry)) + + entry.EncoderInputSize, entry.EncoderMatchedSize = -1, 0 + assert.Error(t, predictor.ValidateTrainingEntry(entry)) + + entry.EncoderInputSize, entry.EncoderMatchedSize = 2, 3 + assert.Error(t, predictor.ValidateTrainingEntry(entry)) +} + +// TestPredictBayesianRidge_EncoderCoefficients verifies the TTFT linear model +// applies encoder-cache coefficients when present and degrades to a zero +// contribution when the model was trained without them. +func TestPredictBayesianRidge_EncoderCoefficients(t *testing.T) { + predictor := &Predictor{} + req := validEncoderPredictionRequest() + req.EncoderInputSize = 10 + req.EncoderMatchedSize = 4 + + withoutEncoder := &MetricsResponse{Coefficients: &ModelCoefficients{ + TTFTIntercept: 100, + TTFTCoeffs: map[string]float64{}, + TPOTCoeffs: map[string]float64{}, + }} + resp, err := predictor.predictBayesianRidge(req, withoutEncoder, 0.9, ObjectiveMean) + require.NoError(t, err) + assert.Equal(t, 100.0, resp.TTFT) + + withEncoder := &MetricsResponse{Coefficients: &ModelCoefficients{ + TTFTIntercept: 100, + TTFTCoeffs: map[string]float64{ + "encoder_input_size": 2, + "encoder_matched_size": -1, + }, + TPOTCoeffs: map[string]float64{}, + }} + resp, err = predictor.predictBayesianRidge(req, withEncoder, 0.9, ObjectiveMean) + require.NoError(t, err) + assert.Equal(t, 100.0+2*10-1*4, resp.TTFT) +} diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/training.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/training.go index e5779c3a15..8379f7756b 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/training.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/training.go @@ -221,6 +221,12 @@ func (p *Predictor) ValidateTrainingEntry(entry TrainingEntry) error { if entry.PrefixCacheScore < 0.0 || entry.PrefixCacheScore > 1.0 { return fmt.Errorf("prefix_cache_score must be between 0.0 and 1.0, got %f", entry.PrefixCacheScore) } + if entry.EncoderInputSize < 0 { + return fmt.Errorf("encoder_input_size must be non-negative, got %d", entry.EncoderInputSize) + } + if entry.EncoderMatchedSize < 0 || entry.EncoderMatchedSize > entry.EncoderInputSize { + return fmt.Errorf("encoder_matched_size must be between 0 and encoder_input_size (%d), got %d", entry.EncoderInputSize, entry.EncoderMatchedSize) + } return nil } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/types.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/types.go index ee5f6b6c7f..f12d1b09d6 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/types.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient/types.go @@ -162,14 +162,19 @@ type PredictorInterface interface { // --- Data Models --- type TrainingEntry struct { - KVCachePercentage float64 `json:"kv_cache_percentage"` - InputTokenLength int `json:"input_token_length"` - NumRequestWaiting int `json:"num_request_waiting"` - NumRequestRunning int `json:"num_request_running"` - NumTokensGenerated int `json:"num_tokens_generated"` - ActualTTFT float64 `json:"actual_ttft_ms"` - ActualTPOT float64 `json:"actual_tpot_ms"` - PrefixCacheScore float64 `json:"prefix_cache_score"` + KVCachePercentage float64 `json:"kv_cache_percentage"` + InputTokenLength int `json:"input_token_length"` + NumRequestWaiting int `json:"num_request_waiting"` + NumRequestRunning int `json:"num_request_running"` + NumTokensGenerated int `json:"num_tokens_generated"` + ActualTTFT float64 `json:"actual_ttft_ms"` + ActualTPOT float64 `json:"actual_tpot_ms"` + PrefixCacheScore float64 `json:"prefix_cache_score"` + // EncoderInputSize is the total size of the request's multimodal encoder + // items; EncoderMatchedSize is the portion likely present in the target + // pod's encoder cache. Both are 0 for text-only requests. + EncoderInputSize int `json:"encoder_input_size"` + EncoderMatchedSize int `json:"encoder_matched_size"` PodType string `json:"pod_type,omitempty"` // "prefill", "decode", or "" for monolithic PrefillTokensInFlight int64 `json:"prefill_tokens_in_flight"` DecodeTokensInFlight int64 `json:"decode_tokens_in_flight"` @@ -181,15 +186,20 @@ type BulkTrainingRequest struct { } type PredictionRequest struct { - KVCachePercentage float64 `json:"kv_cache_percentage"` - InputTokenLength int `json:"input_token_length"` - NumRequestWaiting int `json:"num_request_waiting"` - NumRequestRunning int `json:"num_request_running"` - NumTokensGenerated int `json:"num_tokens_generated"` - PrefixCacheScore float64 `json:"prefix_cache_score"` - PodType string `json:"pod_type,omitempty"` // "prefill", "decode", or "" for monolithic - PrefillTokensInFlight int64 `json:"prefill_tokens_in_flight"` - DecodeTokensInFlight int64 `json:"decode_tokens_in_flight"` + KVCachePercentage float64 `json:"kv_cache_percentage"` + InputTokenLength int `json:"input_token_length"` + NumRequestWaiting int `json:"num_request_waiting"` + NumRequestRunning int `json:"num_request_running"` + NumTokensGenerated int `json:"num_tokens_generated"` + PrefixCacheScore float64 `json:"prefix_cache_score"` + // EncoderInputSize is the total size of the request's multimodal encoder + // items; EncoderMatchedSize is the portion likely present in the target + // pod's encoder cache. Both are 0 for text-only requests. + EncoderInputSize int `json:"encoder_input_size"` + EncoderMatchedSize int `json:"encoder_matched_size"` + PodType string `json:"pod_type,omitempty"` // "prefill", "decode", or "" for monolithic + PrefillTokensInFlight int64 `json:"prefill_tokens_in_flight"` + DecodeTokensInFlight int64 `json:"decode_tokens_in_flight"` } type PredictionResponse struct { diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go index e957209bdf..d7eab0db38 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/plugin.go @@ -41,6 +41,7 @@ import ( fwksched "github.com/llm-d/llm-d-router/pkg/epp/framework/interface/scheduling" attrconcurrency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/concurrency" attrlatency "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/latency" + attrmm "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/multimodal" attrprefix "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/datalayer/attribute/prefix" latencyproducerconstants "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/constants" "github.com/llm-d/llm-d-router/pkg/epp/metadata" @@ -76,6 +77,7 @@ type PredictedLatency struct { config Config prefixMatchDataKey plugin.DataKey inFlightLoadDataKey plugin.DataKey + encoderCacheDataKey plugin.DataKey latencyPredictionInfoDataKey plugin.DataKey } @@ -241,6 +243,16 @@ type Config struct { // load to read for the prefill-tokens-in-flight and active-request-count // features. Empty defaults to the auto-created producer. InFlightLoadProducerName string `json:"inFlightLoadProducerName,omitempty"` + // UseEncoderCacheFeatures enables the multimodal encoder-cache features + // (encoder_input_size, encoder_matched_size). When true, the multimodal + // encoder-cache match data is consumed as a required dependency, so a + // producer is auto-created when none is configured. When false, both + // features are always 0. Default: false. + UseEncoderCacheFeatures bool `json:"useEncoderCacheFeatures,omitempty"` + // EncoderCacheMatchInfoProducerName selects which multimodal encoder-cache + // producer's match data to read. Empty defaults to the auto-created producer. + // Ignored unless UseEncoderCacheFeatures is true. + EncoderCacheMatchInfoProducerName string `json:"encoderCacheMatchInfoProducerName,omitempty"` } var DefaultConfig = Config{ @@ -307,6 +319,7 @@ func NewPredictedLatency(name string, config Config, predictor latencypredictor. config: config, prefixMatchDataKey: attrprefix.PrefixCacheMatchInfoDataKey.WithNonEmptyProducerName(config.PrefixMatchInfoProducerName), inFlightLoadDataKey: attrconcurrency.InFlightLoadDataKey.WithNonEmptyProducerName(config.InFlightLoadProducerName), + encoderCacheDataKey: attrmm.EncoderCacheMatchInfoKey.WithNonEmptyProducerName(config.EncoderCacheMatchInfoProducerName), latencyPredictionInfoDataKey: attrlatency.LatencyPredictionInfoDataKey.WithNonEmptyProducerName(name), } @@ -378,6 +391,13 @@ type predictedLatencyCtx struct { prefixCacheScoresForEndpoints map[string]float64 + // encoderInputSize is the request's total multimodal encoder item size; + // encoderMatchedSizeForEndpoints is the per-endpoint portion likely present + // in that endpoint's encoder cache, keyed by endpoint name. Both stay 0 + // when encoder-cache features are disabled or the request is text-only. + encoderInputSize int + encoderMatchedSizeForEndpoints map[string]int + // inFlightLoadForEndpoints holds the in-flight load captured for every // candidate endpoint during Produce, keyed by NamespacedName.String(). // Produce is DAG-ordered, so capturing here (rather than re-reading the live @@ -406,12 +426,13 @@ func newPredictedLatencyContext(request *fwksched.InferenceRequest) *predictedLa } } return &predictedLatencyCtx{ - schedulingRequest: *request, - inputTokenCount: inputTokenCount, - lastSeenMetrics: make(map[string]*fwkdl.Metrics), - prefixCacheScoresForEndpoints: make(map[string]float64), - inFlightLoadForEndpoints: make(map[string]inFlightLoadSnapshot), - predictionsForScheduling: make(map[string]endpointPredictionResult), + schedulingRequest: *request, + inputTokenCount: inputTokenCount, + lastSeenMetrics: make(map[string]*fwkdl.Metrics), + prefixCacheScoresForEndpoints: make(map[string]float64), + encoderMatchedSizeForEndpoints: make(map[string]int), + inFlightLoadForEndpoints: make(map[string]inFlightLoadSnapshot), + predictionsForScheduling: make(map[string]endpointPredictionResult), } } diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/prediction.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/prediction.go index adb6f19444..b6fbd42108 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/prediction.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/prediction.go @@ -54,6 +54,8 @@ func (pl *PredictedLatency) generatePredictions(ctx context.Context, predictedLa prefixCacheScores := make([]float64, len(candidateEndpoints)) prefillTokensInFlights := make([]int64, len(candidateEndpoints)) numRequestRunnings := make([]int, len(candidateEndpoints)) + encoderInputSizes := make([]int, len(candidateEndpoints)) + encoderMatchedSizes := make([]int, len(candidateEndpoints)) for i, endpoint := range candidateEndpoints { logger.V(logutil.TRACE).Info("Candidate pod for scheduling", "endpoint", endpoint.GetMetadata().String(), "metrics", endpoint.GetMetrics().String()) @@ -68,6 +70,8 @@ func (pl *PredictedLatency) generatePredictions(ctx context.Context, predictedLa inputTokenLengths[i] = predictedLatencyCtx.inputTokenCount generatedTokenCounts[i] = 1 prefixCacheScores[i] = prefixCacheScore + encoderInputSizes[i] = predictedLatencyCtx.encoderInputSize + encoderMatchedSizes[i] = predictedLatencyCtx.encoderMatchedSizeForEndpoints[endpoint.GetMetadata().NamespacedName.Name] // Reuse the in-flight load captured for this endpoint earlier in Produce, // so the prediction features are identical to the dispatch-time training @@ -81,7 +85,7 @@ func (pl *PredictedLatency) generatePredictions(ctx context.Context, predictedLa } // Bulk predict - bulkPredictions, err := bulkPredictWithMetrics(ctx, pl.typedName.Name, pl.typedName.Type, predictedLatencyCtx, pl.latencypredictor, metricsStates, pl.config.EndpointRoleLabel, targetEndpointsMetadatas, inputTokenLengths, generatedTokenCounts, prefixCacheScores, prefillTokensInFlights, numRequestRunnings) + bulkPredictions, err := bulkPredictWithMetrics(ctx, pl.typedName.Name, pl.typedName.Type, predictedLatencyCtx, pl.latencypredictor, metricsStates, pl.config.EndpointRoleLabel, targetEndpointsMetadatas, inputTokenLengths, generatedTokenCounts, prefixCacheScores, prefillTokensInFlights, numRequestRunnings, encoderInputSizes, encoderMatchedSizes) if err != nil { logger.V(logutil.DEBUG).Error(err, "Bulk prediction failed") return nil, err diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go index d96e1e2cfa..0a4383f791 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/requestcontrol_hooks.go @@ -194,6 +194,8 @@ func (pl *PredictedLatency) ResponseBody(ctx context.Context, request *fwksched. now, 0, 0, + 0, + 0, ) entry.PrefillTokensInFlight = predictedLatencyCtx.prefillTokensAtDispatch entry.DecodeTokensInFlight = predictedLatencyCtx.decodeTokensAtDispatch @@ -260,11 +262,12 @@ func processFirstTokenForLatencyPrediction( prefillMetrics, err := getLatestMetricsForProfile(predictedLatencyCtx, ExperimentalDefaultPrefillProfile) if err == nil { prefillPrefixCacheScore := predictedLatencyCtx.prefixCacheScoresForEndpoints[prefillTargetMetadata.NamespacedName.Name] + prefillEncoderMatchedSize := predictedLatencyCtx.encoderMatchedSizeForEndpoints[prefillTargetMetadata.NamespacedName.Name] logger.V(logutil.DEBUG).Info("Recording prefill TTFT training data", "ttft_ms", predictedLatencyCtx.ttft, "prefillPod", prefillTargetMetadata.NamespacedName.Name, "prefixCacheScore", prefillPrefixCacheScore) - recordTTFTTrainingData(ctx, predictor, endpointRoleLabel, predictedLatencyCtx, prefillMetrics, prefillTargetMetadata, now, prefillPrefixCacheScore) + recordTTFTTrainingData(ctx, predictor, endpointRoleLabel, predictedLatencyCtx, prefillMetrics, prefillTargetMetadata, now, prefillPrefixCacheScore, prefillEncoderMatchedSize) } } else { m, err := getLatestMetricsForProfile(predictedLatencyCtx, "") @@ -274,8 +277,9 @@ func processFirstTokenForLatencyPrediction( } targetEndpointMetadata := predictedLatencyCtx.targetMetadata prefixCacheScore := predictedLatencyCtx.prefixCacheScoresForEndpoints[targetEndpointMetadata.NamespacedName.Name] + encoderMatchedSize := predictedLatencyCtx.encoderMatchedSizeForEndpoints[targetEndpointMetadata.NamespacedName.Name] logger.V(logutil.DEBUG).Info("Recording TTFT training data", "ttft_ms", predictedLatencyCtx.ttft, "predicted_ttft_ms", predictedLatencyCtx.predictedTTFT, "prefixCacheScore", prefixCacheScore) - recordTTFTTrainingData(ctx, predictor, endpointRoleLabel, predictedLatencyCtx, m, targetEndpointMetadata, now, prefixCacheScore) + recordTTFTTrainingData(ctx, predictor, endpointRoleLabel, predictedLatencyCtx, m, targetEndpointMetadata, now, prefixCacheScore, encoderMatchedSize) } if streamingMode { @@ -355,6 +359,8 @@ func processTokenForLatencyPrediction( predictedLatencyCtx.inputTokenCount, predictedLatencyCtx.generatedTokenCount, 0, + 0, + 0, ) start := time.Now() p, err := predictor.Predict(ctx, in) diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training.go index e381c73dcd..1d22327fab 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training.go @@ -37,6 +37,8 @@ func buildPredictionRequest( inputTokenLength int, generatedTokens int, prefixCacheScore float64, + encoderInputSize int, + encoderMatchedSize int, ) latencypredictor.PredictionRequest { podType := "" if endpointRoleLabel != "" && targetEndpointMetadata != nil && targetEndpointMetadata.Labels != nil { @@ -50,6 +52,8 @@ func buildPredictionRequest( NumRequestRunning: metrics.RunningRequestsSize, NumTokensGenerated: generatedTokens, PrefixCacheScore: prefixCacheScore, + EncoderInputSize: encoderInputSize, + EncoderMatchedSize: encoderMatchedSize, PodType: podType, } } @@ -65,6 +69,8 @@ func buildTrainingEntry( timestamp time.Time, generatedTokens int, prefixCacheScore float64, + encoderInputSize int, + encoderMatchedSize int, ) latencypredictor.TrainingEntry { podType := "" if endpointRoleLabel != "" && targetEndpointMetadata != nil && targetEndpointMetadata.Labels != nil { @@ -81,6 +87,8 @@ func buildTrainingEntry( NumRequestRunning: m.RunningRequestsSize, NumTokensGenerated: generatedTokens, PrefixCacheScore: prefixCacheScore, + EncoderInputSize: encoderInputSize, + EncoderMatchedSize: encoderMatchedSize, PodType: podType, } } @@ -95,6 +103,7 @@ func recordTTFTTrainingData( targetEndpointMetadata *fwkdl.EndpointMetadata, now time.Time, prefixCacheScore float64, + encoderMatchedSize int, ) { logger := log.FromContext(ctx) entry := buildTrainingEntry( @@ -107,6 +116,8 @@ func recordTTFTTrainingData( now, 0, prefixCacheScore, + predictedLatencyCtx.encoderInputSize, + encoderMatchedSize, ) // In disaggregated serving TTFT is incurred on the prefill endpoint, so the // in-flight features are snapshotted from that endpoint; otherwise the decode @@ -168,6 +179,8 @@ func bulkPredictWithMetrics( prefixCacheScores []float64, prefillTokensInFlights []int64, numRequestRunnings []int, + encoderInputSizes []int, + encoderMatchedSizes []int, ) ([]*latencypredictor.PredictionResponse, error) { logger := log.FromContext(ctx) @@ -201,6 +214,8 @@ func bulkPredictWithMetrics( inputTokenLengths[i], generatedTokenCounts[i], prefixCacheScores[i], + 0, + 0, ) if i < len(prefillTokensInFlights) { bulkRequests[i].PrefillTokensInFlight = prefillTokensInFlights[i] @@ -208,6 +223,14 @@ func bulkPredictWithMetrics( if i < len(numRequestRunnings) { bulkRequests[i].NumRequestRunning = numRequestRunnings[i] } + if i < len(encoderInputSizes) { + bulkRequests[i].EncoderInputSize = encoderInputSizes[i] + } + if i < len(encoderMatchedSizes) { + // The predictor rejects matched > input, so clamp instead of + // failing the whole bulk request when the slices are inconsistent. + bulkRequests[i].EncoderMatchedSize = min(encoderMatchedSizes[i], bulkRequests[i].EncoderInputSize) + } } start := time.Now() diff --git a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training_test.go b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training_test.go index 61eb01da33..beb1ea9df8 100644 --- a/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training_test.go +++ b/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/training_test.go @@ -21,6 +21,7 @@ import ( "errors" "strings" "testing" + "time" latencypredictor "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/predictedlatency/latencypredictorclient" "github.com/stretchr/testify/assert" @@ -55,7 +56,7 @@ func TestBulkPredictWithMetrics(t *testing.T) { generatedTokenCounts := []int{1, 1} prefixCacheScores := []float64{0.0, 0.0} - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil, nil, nil) assert.NoError(t, err) assert.Len(t, results, 2) @@ -94,7 +95,7 @@ func TestBulkPredictWithMetrics_PropagatesInFlightOverrides(t *testing.T) { _, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, - prefillTokensInFlights, numRequestRunnings) + prefillTokensInFlights, numRequestRunnings, nil, nil) require.NoError(t, err) require.Len(t, mockPredictor.capturedBulkStrictRequests, 2) @@ -104,6 +105,81 @@ func TestBulkPredictWithMetrics_PropagatesInFlightOverrides(t *testing.T) { assert.Equal(t, int64(200), mockPredictor.capturedBulkStrictRequests[1].PrefillTokensInFlight) } +// TestBulkPredictWithMetrics_PropagatesEncoderSizes verifies that the +// per-endpoint encoder-cache size slices are written onto the outgoing +// PredictionRequests. +func TestBulkPredictWithMetrics_PropagatesEncoderSizes(t *testing.T) { + mockPredictor := &mockPredictor{ + predictions: map[string]*latencypredictor.PredictionResponse{ + "0.5": {TTFT: 0.5, TPOT: 0.03}, + "0.6": {TTFT: 0.6, TPOT: 0.04}, + }, + } + + metricsStates := []*fwkdl.Metrics{ + {KVCacheUsagePercent: 0.5}, + {KVCacheUsagePercent: 0.6}, + } + pods := []*fwkdl.EndpointMetadata{ + {NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"}}, + {NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod2"}}, + } + inputTokenLengths := []int{1, 1} + generatedTokenCounts := []int{1, 1} + prefixCacheScores := []float64{0.0, 0.0} + encoderInputSizes := []int{3, 3} + encoderMatchedSizes := []int{2, 0} + + _, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, + metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, + nil, nil, encoderInputSizes, encoderMatchedSizes) + require.NoError(t, err) + + require.Len(t, mockPredictor.capturedBulkStrictRequests, 2) + assert.Equal(t, 3, mockPredictor.capturedBulkStrictRequests[0].EncoderInputSize) + assert.Equal(t, 2, mockPredictor.capturedBulkStrictRequests[0].EncoderMatchedSize) + assert.Equal(t, 3, mockPredictor.capturedBulkStrictRequests[1].EncoderInputSize) + assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[1].EncoderMatchedSize) +} + +// TestBulkPredictWithMetrics_ClampsMatchedWithoutInputSizes verifies that a +// matched-size slice without a corresponding input-size slice is clamped to +// the input size (0) instead of producing requests that fail validation. +func TestBulkPredictWithMetrics_ClampsMatchedWithoutInputSizes(t *testing.T) { + mockPredictor := &mockPredictor{ + predictions: map[string]*latencypredictor.PredictionResponse{ + "0.5": {TTFT: 0.5, TPOT: 0.03}, + }, + } + + metricsStates := []*fwkdl.Metrics{{KVCacheUsagePercent: 0.5}} + pods := []*fwkdl.EndpointMetadata{ + {NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"}}, + } + + _, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, + metricsStates, "", pods, []int{1}, []int{1}, []float64{0.0}, + nil, nil, nil, []int{5}) + require.NoError(t, err) + + require.Len(t, mockPredictor.capturedBulkStrictRequests, 1) + assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[0].EncoderInputSize) + assert.Equal(t, 0, mockPredictor.capturedBulkStrictRequests[0].EncoderMatchedSize) +} + +func TestBuildPredictionRequestAndTrainingEntry_EncoderSizes(t *testing.T) { + m := &fwkdl.Metrics{KVCacheUsagePercent: 0.5} + pod := &fwkdl.EndpointMetadata{NamespacedName: types.NamespacedName{Namespace: "default", Name: "pod1"}} + + req := buildPredictionRequest("", pod, m, 10, 1, 0.0, 5, 4) + assert.Equal(t, 5, req.EncoderInputSize) + assert.Equal(t, 4, req.EncoderMatchedSize) + + entry := buildTrainingEntry("", pod, m, 10, 100, 0, time.Now(), 0, 0.0, 5, 4) + assert.Equal(t, 5, entry.EncoderInputSize) + assert.Equal(t, 4, entry.EncoderMatchedSize) +} + func TestBulkPredictWithMetrics_Error(t *testing.T) { mockPredictor := &mockPredictor{ err: errors.New("prediction failed"), @@ -121,7 +197,7 @@ func TestBulkPredictWithMetrics_Error(t *testing.T) { generatedTokenCounts := []int{1} prefixCacheScores := []float64{0.0} - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil, nil, nil) assert.Error(t, err) assert.Nil(t, results) @@ -139,7 +215,7 @@ func TestBulkPredictWithMetrics_InputMismatch(t *testing.T) { generatedTokenCounts := []int{1} prefixCacheScores := []float64{0.0} - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil, nil, nil) assert.Error(t, err) assert.Nil(t, results) @@ -172,7 +248,7 @@ func TestBulkPredictWithMetrics_WithPredictedLatencyCtx(t *testing.T) { incomingModelName: "incoming-model", } - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", plCtx, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", plCtx, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil, nil, nil) assert.NoError(t, err) assert.Len(t, results, 1) @@ -196,7 +272,7 @@ func TestBulkPredictWithMetrics_ChatCompletionsInputTokenLength(t *testing.T) { generatedTokenCounts := []int{1} prefixCacheScores := []float64{0.0} - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mp, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, []int64{0}, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mp, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, []int64{0}, nil, nil, nil) assert.NoError(t, err) assert.Len(t, results, 1) @@ -215,7 +291,7 @@ func TestBulkPredictWithMetrics_NilMetricsState(t *testing.T) { generatedTokenCounts := []int{1} prefixCacheScores := []float64{0.0} - results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil) + results, err := bulkPredictWithMetrics(context.Background(), "test-plugin", "test-type", nil, mockPredictor, metricsStates, "", pods, inputTokenLengths, generatedTokenCounts, prefixCacheScores, nil, nil, nil, nil) assert.Error(t, err) assert.Nil(t, results)