test(e2e): add cache-affinity test for multimodal routing (llm-d#1562)

* test(e2e): add cache-affinity test for multimodal routing

Adds a Ginkgo case under "Running multimodal cache-affinity configuration"
that spins 2 decode pods, wires mm-embeddings-cache-producer +
mm-embeddings-cache-scorer, and asserts two identical multimodal requests
land on the same decode pod via cache-affinity scoring. Iterates over
image, audio, and video in a single It block.

Introduces swapToSimRenderSidecar so the EPP's loopback render endpoint
runs the inference sim instead of text-only Qwen2.5-1.5B; the sim emits
deterministic mm_features for all three modalities while real text-only
vLLM does not. Detection is config-content based: any EPP config
containing mm-embeddings-cache-producer triggers the sidecar swap.

Bumps the pinned sim image to v0.9.2, which includes the audio and video
mm_features stub from llm-d-inference-sim PR llm-d#539.

Fixes llm-d#1541

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

* test(e2e): extend mm cache-affinity test with discrimination + metrics

The original same-content loop passes if the scorer always picks pod 1.
Three additional assertion blocks close that gap:

- A/B image symmetry: each content key independently routes back to its
  first-served pod, proving scoring is keyed on content hash rather than
  "first request wins forever".
- Mixed image+audio: per-item match accounting must drive cache affinity
  on a multi-modality request; probes the empty-hash short-circuit in
  ExtractMMItems.
- Metric assertions on llm_d_router_epp_encoder_cache_{hits,queries}_total
  with the {modality} label, wrapped in Eventually because PreRequest's
  LRU write runs in a wg.Go goroutine.

Helpers added:
- runChatCompletionWithImageAndAudio for the mixed-modality body.
- getMetricValue scrapes the EPP /metrics port-forward and sums matching
  series; sufficient for the small label set this test asserts on.

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

* test(e2e): address review on multimodal cache-affinity test

Rename swapToSimRenderSidecar params (eppManifests/yamlDocs), extend the
mixed request to image+audio+video, and note hits<queries in the metrics
assertion.

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

* test(e2e): add estimate token-producer per-modality metric test

Runs the mm cache-affinity config with the estimate backend and asserts
encoder-cache metrics are labelled per modality (image/audio/video). Fails
until the estimate modality fix (llm-d#1618) lands.

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

---------

Signed-off-by: Sam Batschelet <sbatsche@redhat.com>

Author: hexfusion

test/e2e/configs_test.go

@@ -242,6 +242,57 @@ schedulingProfiles:
     weight: 10
 `
 
+// EPP configuration for multimodal cache-affinity routing. Pairs the
+// mm-embeddings-cache producer+scorer with token-producer at loopback; the sim
+// sidecar (swapped in by usesSimRender) provides mm_features.
+const mmCacheAffinityConfig = `apiVersion: llm-d.ai/v1alpha1
+kind: EndpointPickerConfig
+plugins:
+- type: token-producer
+  parameters:
+    modelName: food-review
+    vllm:
+      url: http://localhost:8000
+- type: mm-embeddings-cache-producer
+  parameters:
+    cacheSizeInMBPerServer: 64
+- type: mm-embeddings-cache-scorer
+- type: decode-filter
+- type: max-score-picker
+- type: single-profile-handler
+schedulingProfiles:
+- name: default
+  plugins:
+  - pluginRef: decode-filter
+  - pluginRef: mm-embeddings-cache-scorer
+    weight: 10
+  - pluginRef: max-score-picker
+`
+
+// Same as mmCacheAffinityConfig but the token-producer uses the estimate
+// backend, which derives mm_features in-EPP and never calls the render endpoint.
+const mmCacheAffinityEstimateConfig = `apiVersion: llm-d.ai/v1alpha1
+kind: EndpointPickerConfig
+plugins:
+- type: token-producer
+  parameters:
+    estimate: {}
+- type: mm-embeddings-cache-producer
+  parameters:
+    cacheSizeInMBPerServer: 64
+- type: mm-embeddings-cache-scorer
+- type: decode-filter
+- type: max-score-picker
+- type: single-profile-handler
+schedulingProfiles:
+- name: default
+  plugins:
+  - pluginRef: decode-filter
+  - pluginRef: mm-embeddings-cache-scorer
+    weight: 10
+  - pluginRef: max-score-picker
+`
+
 // EPP configuration for running scale model server test
 const scaleConfig = `apiVersion: llm-d.ai/v1alpha1
 kind: EndpointPickerConfig

test/e2e/e2e_test.go

@@ -770,6 +770,99 @@ var _ = ginkgo.Describe("Run end to end tests", ginkgo.Ordered, func() {
 		})
 	})
 
+	ginkgo.When("Running multimodal cache-affinity configuration", ginkgo.Label(extendedTestLabel), func() {
+		ginkgo.It("should route identical multimodal content to the same decode pod", func() {
+			infPoolObjects = createInferencePool(1, true)
+
+			decodeReplicas := 2
+			modelServers := createModelServersDecode(decodeReplicas)
+			epp := createEndPointPicker(mmCacheAffinityConfig)
+
+			_, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector)
+			gomega.Expect(decodePods).Should(gomega.HaveLen(decodeReplicas))
+
+			for _, tc := range []struct {
+				name string
+				send func() (string, string)
+			}{
+				{"image", func() (string, string) { return runChatCompletionWithImages(testImageURL) }},
+				{"audio", runChatCompletionWithAudio},
+				{"video", runChatCompletionWithVideo},
+			} {
+				ginkgo.By("modality=" + tc.name)
+
+				ns1, pod1 := tc.send()
+				gomega.Expect(ns1).Should(gomega.Equal(nsName))
+				gomega.Expect(pod1).Should(gomega.BeElementOf(decodePods))
+
+				ns2, pod2 := tc.send()
+				gomega.Expect(ns2).Should(gomega.Equal(nsName))
+				gomega.Expect(pod2).Should(gomega.Equal(pod1))
+			}
+
+			// A/B symmetry: each content key independently routes back to its
+			// first-served pod. Catches "first request wins forever" bugs that
+			// the same-content loop above can't distinguish.
+			ginkgo.By("A/B image symmetry")
+			_, podA := runChatCompletionWithImages(testImageURL)
+			_, podB := runChatCompletionWithImages(testImageURL2)
+			_, podARe := runChatCompletionWithImages(testImageURL)
+			gomega.Expect(podARe).Should(gomega.Equal(podA))
+			_, podBRe := runChatCompletionWithImages(testImageURL2)
+			gomega.Expect(podBRe).Should(gomega.Equal(podB))
+
+			// Mixed image + audio + video in one request. Probes per-item match
+			// accounting and the empty-hash short-circuit in ExtractMMItems.
+			ginkgo.By("mixed image+audio+video cache affinity")
+			_, podMix := runChatCompletionWithImageAudioVideo(testImageURL, testAudioData, testVideoURL)
+			_, podMixRe := runChatCompletionWithImageAudioVideo(testImageURL, testAudioData, testVideoURL)
+			gomega.Expect(podMixRe).Should(gomega.Equal(podMix))
+
+			// hits is the matched subset of queries, so hits < queries (never equal).
+			// PreRequest writes the LRU async (wg.Go) → Eventually.
+			ginkgo.By("metrics: hits_total + queries_total")
+			gomega.Eventually(func() float64 {
+				return getMetricValue("llm_d_router_epp_encoder_cache_queries_total",
+					map[string]string{"modality": "image"})
+			}, 5*time.Second, 250*time.Millisecond).Should(gomega.BeNumerically(">=", 1))
+			gomega.Eventually(func() float64 {
+				return getMetricValue("llm_d_router_epp_encoder_cache_hits_total",
+					map[string]string{"modality": "image"})
+			}, 5*time.Second, 250*time.Millisecond).Should(gomega.BeNumerically(">=", 1))
+
+			testutils.DeleteObjects(testConfig, epp)
+			testutils.DeleteObjects(testConfig, modelServers)
+		})
+
+		// Estimate token-producer: encoder-cache metrics must be labelled per
+		// modality, not folded into "image" (#1617). Depends on the fix in #1618.
+		ginkgo.It("should label per-modality metrics with the estimate token-producer", func() {
+			infPoolObjects = createInferencePool(1, true)
+
+			decodeReplicas := 2
+			modelServers := createModelServersDecode(decodeReplicas)
+			epp := createEndPointPicker(mmCacheAffinityEstimateConfig)
+
+			_, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector)
+			gomega.Expect(decodePods).Should(gomega.HaveLen(decodeReplicas))
+
+			// One request per modality so each is recorded by the producer.
+			runChatCompletionWithImages(testImageURL)
+			runChatCompletionWithAudio()
+			runChatCompletionWithVideo()
+
+			for _, modality := range []string{"image", "audio", "video"} {
+				gomega.Eventually(func() float64 {
+					return getMetricValue("llm_d_router_epp_encoder_cache_queries_total",
+						map[string]string{"modality": modality})
+				}, 5*time.Second, 250*time.Millisecond).Should(gomega.BeNumerically(">=", 1), "modality=%s", modality)
+			}
+
+			testutils.DeleteObjects(testConfig, epp)
+			testutils.DeleteObjects(testConfig, modelServers)
+		})
+	})
+
 	ginkgo.When("Running simple non-PD KV enabled configuration", ginkgo.Label(extendedTestLabel), func() {
 		ginkgo.It("should run successfully", func() {
 			infPoolObjects = createInferencePool(1, true)

test/e2e/requests_test.go

@@ -216,6 +216,15 @@ func runChatCompletionWithVideo() (string, string) {
 	return runRawChatCompletion(body)
 }
 
+// runChatCompletionWithImageAudioVideo sends one request with image_url,
+// input_audio, and video_url blocks to probe per-item accounting across modalities.
+func runChatCompletionWithImageAudioVideo(imageURL, audioData, videoURL string) (string, string) {
+	ginkgo.By("Sending Multimodal Chat Completion Request with image + audio + video")
+	body := fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":[{"type":"image_url","image_url":{"url":%q},"uuid":"image-mixed-0"},{"type":"input_audio","input_audio":{"data":%q,"format":"wav"}},{"type":"video_url","video_url":{"url":%q}},{"type":"text","text":"Describe what you see and hear."}]}],"max_tokens":150}`,
+		simModelName, imageURL, audioData, videoURL)
+	return runRawChatCompletion(body)
+}
+
 // runChatCompletionWithImageEmbeds sends a chat completion request with an image_embeds content block
 // carrying a pre-encoded tensor. image_embeds is not a recognised multimodal type for encode
 // disaggregation, so the request routes like a text request (decode-only or prefill-decode).

test/e2e/setup_test.go

@@ -185,6 +185,8 @@ func createEndPointPickerHelper(eppConfig string, replicas int, isLeaderElection
 		})
 	if !usesTokenProducer(eppConfig) {
 		eppYamls = removeRenderSidecar(eppYamls)
+	} else if usesSimRender(eppConfig) {
+		eppYamls = swapToSimRenderSidecar(eppYamls, vllmSimImage, simModelName)
 	}
 	eppYamls = appendEppArgs(eppYamls, eppExtraArgs)
 
@@ -205,3 +207,16 @@ func usesTokenProducer(eppConfig string) bool {
 	}
 	return false
 }
+
+// usesSimRender reports whether the config needs multimodal /render output;
+// such configs require the sim sidecar (text-only vLLM emits no mm_features).
+func usesSimRender(eppConfig string) bool {
+	cfg, _, err := configloader.LoadRawConfig([]byte(eppConfig), ginkgo.GinkgoLogr)
+	gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+	for _, plugin := range cfg.Plugins {
+		if plugin.Type == "mm-embeddings-cache-producer" {
+			return true
+		}
+	}
+	return false
+}

test/e2e/utils_test.go

@@ -29,6 +29,46 @@ const (
 	deploymentKind = "deployment"
 )
 
+// getMetricValue scrapes the EPP /metrics endpoint and returns the sum of all
+// series for metricName whose label set matches every entry in filters.
+// Returns 0 on scrape error or no match.
+func getMetricValue(metricName string, filters map[string]string) float64 {
+	resp, err := http.Get(fmt.Sprintf("http://localhost:%s/metrics", metricsPort))
+	if err != nil {
+		return 0
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return 0
+	}
+	var total float64
+	for _, line := range strings.Split(string(body), "\n") {
+		if !strings.HasPrefix(line, metricName+"{") && !strings.HasPrefix(line, metricName+" ") {
+			continue
+		}
+		match := true
+		for k, v := range filters {
+			if !strings.Contains(line, fmt.Sprintf("%s=%q", k, v)) {
+				match = false
+				break
+			}
+		}
+		if !match {
+			continue
+		}
+		parts := strings.Fields(line)
+		if len(parts) < 2 {
+			continue
+		}
+		f, err := strconv.ParseFloat(parts[len(parts)-1], 64)
+		if err == nil {
+			total += f
+		}
+	}
+	return total
+}
+
 func scaleDeployment(objects []string, increment int) {
 	direction := "up"
 	absIncrement := increment
@@ -247,6 +287,56 @@ func filterDocument(doc string) string {
 	return strings.TrimRight(string(out), "\n")
 }
 
+// swapToSimRenderSidecar rewrites the vllm-render sidecar to run the inference
+// sim, which emits mm_features that text-only vLLM does not.
+func swapToSimRenderSidecar(eppManifests []string, simImage, modelName string) []string {
+	outputs := make([]string, len(eppManifests))
+	for idx, manifest := range eppManifests {
+		yamlDocs := strings.Split(manifest, "\n---")
+		rendered := make([]string, 0, len(yamlDocs))
+		for _, yamlDoc := range yamlDocs {
+			if strings.TrimSpace(yamlDoc) == "" {
+				continue
+			}
+			rendered = append(rendered, swapRenderSidecarInDoc(yamlDoc, simImage, modelName))
+		}
+		outputs[idx] = strings.Join(rendered, "\n---\n")
+	}
+	return outputs
+}
+
+func swapRenderSidecarInDoc(doc, simImage, modelName string) string {
+	obj := &unstructured.Unstructured{}
+	err := yaml.Unmarshal([]byte(doc), &obj.Object)
+	gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+	if len(obj.Object) == 0 || obj.GetKind() != "Deployment" {
+		return doc
+	}
+	path := []string{"spec", "template", "spec", "containers"}
+	containers, found, err := unstructured.NestedSlice(obj.Object, path...)
+	if err != nil || !found {
+		return doc
+	}
+	for i, raw := range containers {
+		c, ok := raw.(map[string]any)
+		if !ok || c["name"] != "vllm-render" {
+			continue
+		}
+		c["image"] = simImage
+		delete(c, "command")
+		// --log-http: surface inbound URL/method so CI failures are diagnosable.
+		c["args"] = []any{"--port=8000", "--model=" + modelName, "--log-http"}
+		delete(c, "volumeMounts")
+		containers[i] = c
+	}
+	err = unstructured.SetNestedSlice(obj.Object, containers, path...)
+	gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+	removePodSpecListItem(obj, "volumes", "model-cache")
+	out, err := yaml.Marshal(obj.Object)
+	gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+	return strings.TrimRight(string(out), "\n")
+}
+
 func removePodSpecListItem(obj *unstructured.Unstructured, fieldName, itemName string) {
 	path := []string{"spec", "template", "spec", fieldName}
 	items, found, err := unstructured.NestedSlice(obj.Object, path...)