From e9e3b443d4757b24edf532d7dace9523c36bca3e Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:41:06 +0300 Subject: [PATCH 01/10] Added the ability to set the number of replicas and to enable leader selection Signed-off-by: Shmuel Kallner --- deploy/components/inference-gateway/deployment.yaml | 3 ++- deploy/environments/dev/base-kind-istio/patch-deployments.yaml | 1 + 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/deploy/components/inference-gateway/deployment.yaml b/deploy/components/inference-gateway/deployment.yaml index 389b48d8f9..0e34342e4f 100644 --- a/deploy/components/inference-gateway/deployment.yaml +++ b/deploy/components/inference-gateway/deployment.yaml @@ -5,7 +5,7 @@ metadata: labels: app: ${EPP_NAME} spec: - replicas: 1 + replicas: ${EPP_REPLICA_COUNT} selector: matchLabels: app: ${EPP_NAME} @@ -36,6 +36,7 @@ spec: - --config-file - "/etc/epp/epp-config.yaml" - --metrics-endpoint-auth=${METRICS_ENDPOINT_AUTH} + - --ha-enable-leader-election=${ENABLE_LEADER_ELECTION} ports: - containerPort: 5557 - containerPort: 9002 diff --git a/deploy/environments/dev/base-kind-istio/patch-deployments.yaml b/deploy/environments/dev/base-kind-istio/patch-deployments.yaml index 4dc0638cb9..a60bd69c32 100644 --- a/deploy/environments/dev/base-kind-istio/patch-deployments.yaml +++ b/deploy/environments/dev/base-kind-istio/patch-deployments.yaml @@ -24,3 +24,4 @@ spec: - --config-file - "/etc/epp/epp-config.yaml" - --metrics-endpoint-auth=false + - --ha-enable-leader-election=${ENABLE_LEADER_ELECTION} From 476b4354c358e40d86e3d61cbd32275d4c87c8c1 Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:41:31 +0300 Subject: [PATCH 02/10] Added RBAC for leader selection Signed-off-by: Shmuel Kallner --- deploy/components/inference-gateway/rbac.yaml | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/deploy/components/inference-gateway/rbac.yaml b/deploy/components/inference-gateway/rbac.yaml index 75ac704905..7a7990fb9e 100644 --- a/deploy/components/inference-gateway/rbac.yaml +++ b/deploy/components/inference-gateway/rbac.yaml @@ -62,3 +62,27 @@ roleRef: apiGroup: rbac.authorization.k8s.io kind: Role name: ${EPP_NAME} +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: ${EPP_NAME}-leader-election +rules: +- apiGroups: [ "coordination.k8s.io" ] + resources: [ "leases" ] + verbs: [ "get", "list", "watch", "create", "update", "patch", "delete" ] +- apiGroups: [ "" ] + resources: [ "events" ] + verbs: [ "create", "patch" ] +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: ${EPP_NAME}-leader-election-binding +subjects: +- kind: ServiceAccount + name: ${EPP_NAME} +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: ${EPP_NAME}-leader-election \ No newline at end of file From c309e3474389814a818995b3c9c281f8a497755c Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:42:17 +0300 Subject: [PATCH 03/10] Set extra variables for development runs Signed-off-by: Shmuel Kallner --- scripts/kind-dev-env.sh | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/scripts/kind-dev-env.sh b/scripts/kind-dev-env.sh index 57a94d96c5..866b38cd65 100755 --- a/scripts/kind-dev-env.sh +++ b/scripts/kind-dev-env.sh @@ -382,11 +382,17 @@ kubectl --context ${KUBE_CONTEXT} delete configmap epp-config --ignore-not-found envsubst '$MODEL_NAME' < ${EPP_CONFIG} > ${TEMP_FILE} kubectl --context ${KUBE_CONTEXT} create configmap epp-config --from-file=epp-config.yaml=${TEMP_FILE} +# The replica count is changed in some end to end tests +export EPP_REPLICA_COUNT=1 +# Some end to end tests enable leader election +export ENABLE_LEADER_ELECTION=false + # Deploy Istio base (shared infrastructure) kubectl kustomize --enable-helm deploy/environments/dev/base-kind-istio \ | envsubst '${POOL_NAME} ${MODEL_NAME} ${MODEL_NAME_SAFE} ${EPP_NAME} ${EPP_IMAGE} ${VLLM_IMAGE} \ ${SIDECAR_IMAGE} ${VLLM_RENDER_IMAGE} ${TARGET_PORTS} ${NAMESPACE} ${METRICS_ENDPOINT_AUTH} \ -${VLLM_REPLICA_COUNT_E} ${VLLM_REPLICA_COUNT_P} ${VLLM_REPLICA_COUNT_D} ${VLLM_DATA_PARALLEL_SIZE}' \ + ${EPP_REPLICA_COUNT} ${VLLM_REPLICA_COUNT_E} ${VLLM_REPLICA_COUNT_P} ${VLLM_REPLICA_COUNT_D} \ + ${VLLM_DATA_PARALLEL_SIZE} ${ENABLE_LEADER_ELECTION}' \ | kubectl --context ${KUBE_CONTEXT} apply -f - # Deploy scenario-specific vLLM components From b69b5fa336c144f5f6c4dd20927638066829750b Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:43:17 +0300 Subject: [PATCH 04/10] Use the same version of kind in both e2e tests Signed-off-by: Shmuel Kallner --- test/scripts/test-e2e-gaie.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/scripts/test-e2e-gaie.sh b/test/scripts/test-e2e-gaie.sh index d9720b4114..2dfced8f9b 100755 --- a/test/scripts/test-e2e-gaie.sh +++ b/test/scripts/test-e2e-gaie.sh @@ -25,7 +25,7 @@ EPP_IMAGE="${EPP_IMAGE:-ghcr.io/llm-d/llm-d-router-endpoint-picker:dev}" SIM_IMAGE="${VLLM_IMAGE:-ghcr.io/llm-d/llm-d-inference-sim:v0.9.2}" MANIFEST_PATH="${MANIFEST_PATH:-${DIR}/../testdata/sim-deployment.yaml}" USE_KIND="${USE_KIND:-true}" -KIND_NODE_IMAGE="${KIND_NODE_IMAGE:-mirror.gcr.io/kindest/node:v1.32.2}" +KIND_NODE_IMAGE="${KIND_NODE_IMAGE:-kindest/node:v1.31.12}" KIND_CLUSTER_NAME="inference-e2e" From e23f03748407a7a859b7c328cec9b2f854a12637 Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:44:09 +0300 Subject: [PATCH 05/10] Refactored functions for more flexibility in tests Signed-off-by: Shmuel Kallner --- test/utils/igw/utils.go | 39 +++++++++++++++++++++++++++------------ 1 file changed, 27 insertions(+), 12 deletions(-) diff --git a/test/utils/igw/utils.go b/test/utils/igw/utils.go index 621fa5edac..cc175e34c7 100644 --- a/test/utils/igw/utils.go +++ b/test/utils/igw/utils.go @@ -395,14 +395,30 @@ func EventuallyExists(testConfig *TestConfig, getResource func() error) { } func CreateAndVerifyObjs(testConfig *TestConfig, objs []*unstructured.Unstructured) []string { - objNames := make([]string, 0, len(objs)) - for _, unstrObj := range objs { + objNames := CreateObjsWithVerifier(testConfig, objs, + func(kind string, clientObj client.Object) { + switch kind { + case "CustomResourceDefinition": + CRDEstablished(testConfig, clientObj.(*apiextv1.CustomResourceDefinition)) + case "Deployment": + DeploymentAvailable(testConfig, clientObj.(*appsv1.Deployment)) + case "Pod": + PodReady(testConfig, clientObj.(*corev1.Pod)) + } + }) + + return objNames +} + +func CreateObjsWithVerifier(testConfig *TestConfig, objs []*unstructured.Unstructured, verifier func(kind string, clientObj client.Object)) []string { + objNames := make([]string, len(objs)) + for idx, unstrObj := range objs { ginkgo.By(fmt.Sprintf("Processing GVK: %s", unstrObj.GroupVersionKind())) unstrObj.SetNamespace(testConfig.NsName) kind := unstrObj.GetKind() name := unstrObj.GetName() - objNames = append(objNames, kind+"/"+name) + objNames[idx] = kind + "/" + name err := testConfig.K8sClient.Create(testConfig.Context, unstrObj, &client.CreateOptions{}) if apierrors.IsAlreadyExists(err) { @@ -418,14 +434,7 @@ func CreateAndVerifyObjs(testConfig *TestConfig, objs []*unstructured.Unstructur types.NamespacedName{Namespace: testConfig.NsName, Name: name}, clientObj) }) - switch kind { - case "CustomResourceDefinition": - CRDEstablished(testConfig, clientObj.(*apiextv1.CustomResourceDefinition)) - case "Deployment": - DeploymentAvailable(testConfig, clientObj.(*appsv1.Deployment)) - case "Pod": - PodReady(testConfig, clientObj.(*corev1.Pod)) - } + verifier(kind, clientObj) } return objNames } @@ -477,6 +486,12 @@ func ProcessKustomize(testConfig *TestConfig, kustomizePath string, action func( // CreateObjsFromYaml creates K8S objects from yaml and waits for them to be instantiated func CreateObjsFromYaml(testConfig *TestConfig, docs []string) []string { + objs := CreateUnstructuredObjs(testConfig, docs) + return CreateAndVerifyObjs(testConfig, objs) +} + +// CreateUnstructuredObjs creates K8S UnstructuredObject structs from an array of YAMLs +func CreateUnstructuredObjs(testConfig *TestConfig, docs []string) []*unstructured.Unstructured { objs := make([]*unstructured.Unstructured, 0, len(docs)) decoder := serializer.NewCodecFactory(testConfig.Scheme).UniversalDeserializer() @@ -502,7 +517,7 @@ func CreateObjsFromYaml(testConfig *TestConfig, docs []string) []string { } objs = append(objs, unstrObj) } - return CreateAndVerifyObjs(testConfig, objs) + return objs } // ApplyYAMLFile reads a file containing YAML (possibly multiple docs) From b7574a65379b2f272b8fcf7687213c6a18323b5c Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:44:44 +0300 Subject: [PATCH 06/10] Refactored metrics fetching Signed-off-by: Shmuel Kallner --- test/e2e/utils_test.go | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/test/e2e/utils_test.go b/test/e2e/utils_test.go index 4f0ce6d2a3..129938116b 100644 --- a/test/e2e/utils_test.go +++ b/test/e2e/utils_test.go @@ -273,11 +273,9 @@ func substituteMany(inputs []string, substitutions map[string]string) []string { return outputs } -// getCounterMetric fetches the current value of a Prometheus counter metric from the given metrics URL. +// getMetrics fetches the current Prometheus metrics from the given metrics URL. // Retries on transient connection errors (e.g. the previous EPP pod is still terminating). -// -//nolint:unparam // metricName may vary in future test cases -func getCounterMetric(metricsURL, metricName, labelMatch string) int { +func getMetrics(metricsURL string) []string { var body []byte gomega.Eventually(func() error { resp, err := http.Get(metricsURL) @@ -292,8 +290,15 @@ func getCounterMetric(metricsURL, metricName, labelMatch string) int { return err }, 10*time.Second, 1*time.Second).Should(gomega.Succeed()) - metricsText := string(body) - for _, line := range strings.Split(metricsText, "\n") { + return strings.Split(string(body), "\n") +} + +// getCounterMetric fetches the current value of a Prometheus counter metric from the given metrics URL. +// Retries on transient connection errors (e.g. the previous EPP pod is still terminating). +// +//nolint:unparam // metricName may vary in future test cases +func getCounterMetric(metricsURL, metricName, labelMatch string) int { + for _, line := range getMetrics(metricsURL) { if strings.HasPrefix(line, metricName) && strings.Contains(line, labelMatch) { fields := strings.Fields(line) if len(fields) >= 2 { From be86ef14798b9919cd05668929809ccb149090d8 Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:45:10 +0300 Subject: [PATCH 07/10] Aliased additional IGW test utilities Signed-off-by: Shmuel Kallner --- test/utils/k8s_objects.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/test/utils/k8s_objects.go b/test/utils/k8s_objects.go index 9af344f5b6..9c660ca6e5 100644 --- a/test/utils/k8s_objects.go +++ b/test/utils/k8s_objects.go @@ -38,10 +38,13 @@ import ( type TestConfig = igwtestutils.TestConfig var ( - NewTestConfig = igwtestutils.NewTestConfig - ApplyYAMLFile = igwtestutils.ApplyYAMLFile - CreateObjsFromYaml = igwtestutils.CreateObjsFromYaml - ReadYaml = igwtestutils.ReadYaml + NewTestConfig = igwtestutils.NewTestConfig + ApplyYAMLFile = igwtestutils.ApplyYAMLFile + CreateObjsFromYaml = igwtestutils.CreateObjsFromYaml + CreateObjsWithVerifier = igwtestutils.CreateObjsWithVerifier + CreateUnstructuredObjs = igwtestutils.CreateUnstructuredObjs + ReadYaml = igwtestutils.ReadYaml + EventuallyExists = igwtestutils.EventuallyExists ) // DeleteObjects deletes a set of Kubernetes objects in the form of kind/name. From 2c8cc558dc67e57dad4768e5af49f6a3517ae003 Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:45:43 +0300 Subject: [PATCH 08/10] Refcatored EPP creation for leadership tests Signed-off-by: Shmuel Kallner --- test/e2e/setup_test.go | 58 +++++++++++++++++++++++++----------------- 1 file changed, 35 insertions(+), 23 deletions(-) diff --git a/test/e2e/setup_test.go b/test/e2e/setup_test.go index ff9173df09..f9956e21e2 100644 --- a/test/e2e/setup_test.go +++ b/test/e2e/setup_test.go @@ -11,6 +11,7 @@ import ( "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" configloader "github.com/llm-d/llm-d-router/pkg/epp/config/loader" "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/requestcontrol/dataproducer/tokenizer" @@ -128,6 +129,27 @@ func createModelServersEPDUnified(replicas int) []string { } func createEndPointPicker(eppConfig string) []string { + objects := createEndPointPickerHelper(eppConfig, 1, false, true) + podsInDeploymentsReady(objects) + + // Envoy registers the EPP as a healthy ext_proc upstream asynchronously. + // "no healthy upstream" returns HTTP 500 with empty body; any non-empty + // response (200 or 500-with-body) means EPP is reachable from Envoy. + ginkgo.By("Waiting for gateway to be ready") + gomega.Eventually(func() bool { + resp, err := http.Get(fmt.Sprintf("http://localhost:%s/v1/models", port)) + if err != nil { + return false + } + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + return resp.StatusCode == http.StatusOK || len(body) > 0 + }, readyTimeout, 2*time.Second).Should(gomega.BeTrue(), "gateway should be ready within the ready timeout") + + return objects +} + +func createEndPointPickerHelper(eppConfig string, replicas int, isLeaderElectionEnabled bool, waitForReady bool) []string { configMap := &corev1.ConfigMap{ TypeMeta: metav1.TypeMeta{ APIVersion: "v1", @@ -148,38 +170,28 @@ func createEndPointPicker(eppConfig string) []string { eppYamls := testutils.ReadYaml(eppManifest) eppYamls = substituteMany(eppYamls, map[string]string{ - "${EPP_NAME}": "e2e-epp", + "${EPP_NAME}": eppName, "${EPP_IMAGE}": eppImage, "${VLLM_RENDER_IMAGE}": vllmRenderImage, // The render sidecar needs a real, fetchable model. Sim tests // don't query it; the cost is paying weights-load on every EPP. - "${MODEL_NAME}": kvModelName, - "${NAMESPACE}": nsName, - "${POOL_NAME}": simModelName + "-inference-pool", - "${METRICS_ENDPOINT_AUTH}": "false", + "${MODEL_NAME}": kvModelName, + "${NAMESPACE}": nsName, + "${POOL_NAME}": simModelName + "-inference-pool", + "${METRICS_ENDPOINT_AUTH}": "false", + "${EPP_REPLICA_COUNT}": strconv.Itoa(replicas), + "${ENABLE_LEADER_ELECTION}": strconv.FormatBool(isLeaderElectionEnabled), }) if !usesTokenProducer(eppConfig) { eppYamls = removeRenderSidecar(eppYamls) } - objects = append(objects, testutils.CreateObjsFromYaml(testConfig, eppYamls)...) - podsInDeploymentsReady(objects) - - // Envoy registers the EPP as a healthy ext_proc upstream asynchronously. - // "no healthy upstream" returns HTTP 500 with empty body; any non-empty - // response (200 or 500-with-body) means EPP is reachable from Envoy. - ginkgo.By("Waiting for gateway to be ready") - gomega.Eventually(func() bool { - resp, err := http.Get(fmt.Sprintf("http://localhost:%s/v1/models", port)) - if err != nil { - return false - } - body, _ := io.ReadAll(resp.Body) - _ = resp.Body.Close() - return resp.StatusCode == http.StatusOK || len(body) > 0 - }, readyTimeout, 2*time.Second).Should(gomega.BeTrue(), "gateway should be ready within the ready timeout") - - return objects + if waitForReady { + return append(objects, testutils.CreateObjsFromYaml(testConfig, eppYamls)...) + } else { + objs := testutils.CreateUnstructuredObjs(testConfig, eppYamls) + return append(objects, testutils.CreateObjsWithVerifier(testConfig, objs, func(kind string, clientObj client.Object) {})...) + } } func usesTokenProducer(eppConfig string) bool { From 18545f6a3861657078c3328fc882db060ed0671d Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Sun, 21 Jun 2026 18:47:04 +0300 Subject: [PATCH 09/10] Refactored tests, added encoding and leadership tests Signed-off-by: Shmuel Kallner --- test/e2e/e2e_suite_test.go | 9 +++ test/e2e/e2e_test.go | 114 +++++++++++++++++++++++++--- test/e2e/requests_test.go | 150 +++++++++++++++++++++++++++++++++++++ 3 files changed, 264 insertions(+), 9 deletions(-) diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go index 0428829be9..16d3f80849 100644 --- a/test/e2e/e2e_suite_test.go +++ b/test/e2e/e2e_suite_test.go @@ -15,6 +15,7 @@ import ( apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "sigs.k8s.io/controller-runtime/pkg/client/config" k8slog "sigs.k8s.io/controller-runtime/pkg/log" @@ -28,6 +29,8 @@ import ( const ( // kindClusterName is the name of the Kind cluster created for e2e tests. kindClusterName = "e2e-tests" + // eppName is the value of the app label on the EPP pods + eppName = "e2e-epp" // defaultReadyTimeout is the default timeout for a resource to report a ready state. defaultReadyTimeout = 3 * time.Minute // defaultInterval is the default interval to check if a resource exists or ready conditions. @@ -281,6 +284,12 @@ func setupNameSpace() { _, err = testConfig.KubeCli.CoreV1().Namespaces().Create(testConfig.Context, namespace, metav1.CreateOptions{}) gomega.Expect(err).NotTo(gomega.HaveOccurred()) createdNameSpace = true + + ginkgo.By("Ensuring namespace exists: " + nsName) + testutils.EventuallyExists(testConfig, func() error { + return testConfig.K8sClient.Get(testConfig.Context, + types.NamespacedName{Name: nsName}, &corev1.Namespace{}) + }) } // createCRDs creates the Inference Extension CRDs used for testing. diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go index ce1b8892ce..93a66d34b3 100644 --- a/test/e2e/e2e_test.go +++ b/test/e2e/e2e_test.go @@ -2,10 +2,13 @@ package e2e import ( "fmt" + "strings" "time" "github.com/onsi/ginkgo/v2" "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + "sigs.k8s.io/controller-runtime/pkg/client" "github.com/llm-d/llm-d-router/pkg/epp/framework/plugins/scheduling/profilehandler/disagg" testutils "github.com/llm-d/llm-d-router/test/utils" @@ -49,6 +52,9 @@ var ( prefillDecodeSelector = map[string]string{"llm-d.ai/role": "prefill-decode"} encodeSelector = map[string]string{"llm-d.ai/role": "encode"} epdSingleSelector = map[string]string{"llm-d.ai/role": "encode-prefill-decode"} + + singleEmbedding = []string{"The food was delicious and the service was great."} + doubleEmbedding = []string{"First sentence to embed.", "Second sentence to embed."} ) var _ = ginkgo.Describe("Run end to end tests", ginkgo.Ordered, func() { @@ -60,23 +66,89 @@ var _ = ginkgo.Describe("Run end to end tests", ginkgo.Ordered, func() { epp := createEndPointPicker(simpleConfig) - prefillPods, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector) - gomega.Expect(prefillPods).Should(gomega.BeEmpty()) - gomega.Expect(decodePods).Should(gomega.HaveLen(1)) + generateAndCheckLoad(5) - nsHdr, podHdr, _ := runCompletion(simplePrompt, simModelName) - gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) - gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + testutils.DeleteObjects(testConfig, epp) + testutils.DeleteObjects(testConfig, modelServers) + }) - nsHdr, podHdr, _ = runChatCompletion(simplePrompt, simModelName) - gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) - gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + ginkgo.It("should report metrics", func() { + numTargetPorts := 1 + infPoolObjects = createInferencePool(numTargetPorts, true) + temp := strings.Split(infPoolObjects[0], "/") + infPoolName := temp[1] + + modelServers := createModelServersDecode(1) + + epp := createEndPointPicker(simpleConfig) + + verifyMetrics(infPoolName, numTargetPorts) testutils.DeleteObjects(testConfig, epp) testutils.DeleteObjects(testConfig, modelServers) }) }) + ginkgo.When("Running leader election", func() { + ginkgo.It("Should elect one leader and have other pods as not ready", func() { + numOfPods := 3 + numTargetPorts := 1 + + infPoolObjects = createInferencePool(numTargetPorts, true) + + modelServers := createModelServersDecode(1) + + epp := createEndPointPickerHelper(simpleConfig, numOfPods, true, false) + + ginkgo.By("Verifying that exactly one EPP pod is ready") + waitForReadyLeader(numOfPods) + + testutils.DeleteObjects(testConfig, epp) + testutils.DeleteObjects(testConfig, modelServers) + }) + + ginkgo.It("Should successfully failover and serve traffic after the leader pod is deleted", func() { + numOfPods := 3 + numTargetPorts := 1 + + infPoolObjects = createInferencePool(numTargetPorts, true) + temp := strings.Split(infPoolObjects[0], "/") + infPoolName := temp[1] + + modelServers := createModelServersDecode(1) + + epp := createEndPointPickerHelper(simpleConfig, numOfPods, true, false) + + ginkgo.By("STEP 1: Verifying initial leader is working correctly before failover") + leaderPod := waitForReadyLeader(numOfPods) + generateAndCheckLoad(5) + verifyMetrics(infPoolName, numTargetPorts) + + ginkgo.By("Found initial leader pod: " + leaderPod.Name) + + ginkgo.By(fmt.Sprintf("Deleting leader pod %s to trigger failover", leaderPod.Name)) + gomega.Expect(testConfig.K8sClient.Delete(testConfig.Context, leaderPod)).To(gomega.Succeed()) + + ginkgo.By("STEP 3: Waiting for a new and different leader to be elected") + // The deployment controller will create a new pod. We need to wait for the total number of pods + // to be back to 3, and for one of the other pods to become the new leader. + var newLeaderPod *corev1.Pod + gomega.Eventually(func(g gomega.Gomega) { + newLeaderPod = waitForReadyLeader(numOfPods) + g.Expect(newLeaderPod.Name).NotTo(gomega.Equal(leaderPod.Name), "The new leader should not be the same as the old deleted leader") + }, testConfig.ReadyTimeout, testConfig.Interval).Should(gomega.Succeed()) + ginkgo.By("Found new leader pod: " + newLeaderPod.Name) + + ginkgo.By("STEP 4: Verifying the new leader is working correctly after failover") + generateAndCheckLoad(5) + verifyMetrics(infPoolName, numTargetPorts) + + testutils.DeleteObjects(testConfig, epp) + testutils.DeleteObjects(testConfig, modelServers) + + }) + }) + ginkgo.When("Running a PD configuration with nixlv2 connector(deprecated pd-profile-handler)", ginkgo.Label(metricsTestLabel, deprecatedPDTestLabel), func() { ginkgo.It("should run successfully", func() { infPoolObjects = createInferencePool(1, true) @@ -862,3 +934,27 @@ var _ = ginkgo.Describe("Run end to end tests", ginkgo.Ordered, func() { }) }) }) + +func waitForReadyLeader(numOfPods int) *corev1.Pod { + var leaderPod *corev1.Pod + gomega.Eventually(func(g gomega.Gomega) { + podList := &corev1.PodList{} + err := testConfig.K8sClient.List(testConfig.Context, podList, client.InNamespace(testConfig.NsName), client.MatchingLabels{"app": eppName}) + g.Expect(err).NotTo(gomega.HaveOccurred()) + + // The deployment should have 3 replicas for leader election. + g.Expect(podList.Items).To(gomega.HaveLen(numOfPods)) + + readyPods := 0 + for _, pod := range podList.Items { + for _, cond := range pod.Status.Conditions { + if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue { + readyPods++ + leaderPod = &pod + } + } + } + g.Expect(readyPods).To(gomega.Equal(1), "Expected exactly one pod to be ready") + }, testConfig.ReadyTimeout, testConfig.Interval).Should(gomega.Succeed()) + return leaderPod +} diff --git a/test/e2e/requests_test.go b/test/e2e/requests_test.go index befb02b783..071b87b3e5 100644 --- a/test/e2e/requests_test.go +++ b/test/e2e/requests_test.go @@ -11,6 +11,7 @@ import ( "github.com/onsi/gomega" "github.com/openai/openai-go" "github.com/openai/openai-go/option" + "github.com/openai/openai-go/packages/param" ) func newOpenAIClient() *openai.Client { @@ -24,6 +25,30 @@ func extractInferenceHeaders(httpResp *http.Response) (string, string, string) { httpResp.Header.Get("x-inference-port") } +func generateAndCheckLoad(count int) { + for range count { + prefillPods, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector) + gomega.Expect(prefillPods).Should(gomega.BeEmpty()) + gomega.Expect(decodePods).Should(gomega.HaveLen(1)) + + nsHdr, podHdr, _ := runCompletion(simplePrompt, simModelName) + gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) + gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + + nsHdr, podHdr, _ = runChatCompletion(simplePrompt, simModelName) + gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) + gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + + nsHdr, podHdr, _ = runEmbeddings(singleEmbedding, simModelName) + gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) + gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + + nsHdr, podHdr, _ = runEmbeddings(doubleEmbedding, simModelName) + gomega.Expect(nsHdr).Should(gomega.Equal(nsName)) + gomega.Expect(podHdr).Should(gomega.Equal(decodePods[0])) + } +} + // doPost sends a POST request with a JSON body to the given path, asserts HTTP 200, // and returns the x-inference-namespace, x-inference-pod headers and the response body. func doPost(path, body string, extraHeaders map[string]string) (string, string, []byte) { @@ -46,6 +71,27 @@ func doPost(path, body string, extraHeaders map[string]string) (string, string, return resp.Header.Get("x-inference-namespace"), resp.Header.Get("x-inference-pod"), respBody } +// doPostWithError sends a POST request with a JSON body to the given path +// and returns the status code and the response body. +func doPostWithError(path, body string, extraHeaders map[string]string) (int, []byte) { + req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://localhost:%s%s", port, path), strings.NewReader(body)) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + req.Header.Set("Content-Type", "application/json") + for k, v := range extraHeaders { + req.Header.Set(k, v) + } + resp, err := http.DefaultClient.Do(req) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + defer func() { + gomega.Expect(resp.Body.Close()).ToNot(gomega.HaveOccurred()) + }() + + respBody, err := io.ReadAll(resp.Body) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + + return resp.StatusCode, respBody +} + func runCompletion(prompt string, theModel openai.CompletionNewParamsModel) (string, string, string) { var httpResp *http.Response @@ -115,6 +161,27 @@ func runChatCompletion(prompt, modelName string) (string, string, string) { return extractInferenceHeaders(httpResp) } +func runEmbeddings(embeddings []string, modelName string) (string, string, string) { + var httpResp *http.Response + + input := openai.EmbeddingNewParamsInputUnion{} + if len(embeddings) == 1 { + input.OfString = param.NewOpt(embeddings[0]) + } else { + input.OfArrayOfStrings = embeddings + } + + params := openai.EmbeddingNewParams{ + Input: input, + Model: modelName, + } + resp, err := newOpenAIClient().Embeddings.New(testConfig.Context, params, option.WithResponseInto(&httpResp)) + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + gomega.Expect(resp.Data).Should(gomega.HaveLen(len(embeddings))) + + return extractInferenceHeaders(httpResp) +} + // runRawChatCompletion POSTs the given JSON body to /v1/chat/completions and returns // the x-inference-namespace and x-inference-pod response headers. func runRawChatCompletion(body string) (string, string) { @@ -217,3 +284,86 @@ func cacheThresholdHeaders(force bool) map[string]string { } return nil } + +func verifyMetrics(infPoolName string, numTargetPorts int) { + + generateAndCheckLoad(10) + + // Send a few errors + for range 10 { + doPostWithError("/v1/chat/completions", "an invalid body", nil) + } + + metricsURL := fmt.Sprintf("http://localhost:%s/metrics", metricsPort) + + if k8sContext != "" { + // Use port-forward to access the EPP pod's metrics endpoint. + startEPPMetricsPortForward() + } + + theMetrics := getMetrics(metricsURL) + gomega.Expect(theMetrics).ShouldNot(gomega.BeEmpty()) + metricsAsString := strings.Join(theMetrics, "\n") + + _, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector) + + // Define the metrics we expect to see + preset := []string{ //nolint:prealloc + "inference_objective_request_total", + "inference_objective_request_error_total", + "inference_objective_request_duration_seconds", + "inference_objective_normalized_time_per_output_token_seconds", + "inference_objective_request_sizes", + "inference_objective_response_sizes", + "inference_objective_input_tokens", + "inference_objective_output_tokens", + "inference_pool_average_kv_cache_utilization", + "inference_pool_average_queue_size", + "inference_pool_per_pod_queue_size", + "inference_objective_running_requests", + "inference_pool_ready_pods", + "inference_extension_info", + + // llm_d metrics + "llm_d_epp_request_total", + "llm_d_epp_request_error_total", + "llm_d_epp_request_duration_seconds", + "llm_d_epp_normalized_time_per_output_token_seconds", + "llm_d_epp_request_sizes", + "llm_d_epp_response_sizes", + "llm_d_epp_input_tokens", + "llm_d_epp_output_tokens", + "llm_d_epp_average_kv_cache_utilization", + "llm_d_epp_average_queue_size", + "llm_d_epp_per_endpoint_queue_size", + "llm_d_epp_running_requests", + "llm_d_epp_ready_endpoints", + "llm_d_epp_info", + } + expectedMetrics := make([]string, 0, len(preset)+len(decodePods)*numTargetPorts*2) + expectedMetrics = append(expectedMetrics, preset...) + + for _, modelServerPodName := range decodePods { + for rank := range numTargetPorts { + metricQueueSize := fmt.Sprintf( + "inference_pool_per_pod_queue_size{model_server_pod=\"%s-rank-%d\",name=\"%s\"}", + modelServerPodName, + rank, + infPoolName) + expectedMetrics = append(expectedMetrics, metricQueueSize) + + metricQueueSizeNew := fmt.Sprintf( + "llm_d_epp_per_endpoint_queue_size{model_server_endpoint=\"%s-rank-%d\",name=\"%s\"}", + modelServerPodName, + rank, + infPoolName, + ) + expectedMetrics = append(expectedMetrics, metricQueueSizeNew) + } + } + + // Check if all expected metrics are present in the metrics output. + for _, metric := range expectedMetrics { + gomega.Expect(metricsAsString).Should(gomega.ContainSubstring(metric)) + } +} From a913ec735686fe1e77096cae2f3bb4aeeb1b4ff9 Mon Sep 17 00:00:00 2001 From: Shmuel Kallner Date: Mon, 22 Jun 2026 07:31:57 +0300 Subject: [PATCH 10/10] Fixed lint error Signed-off-by: Shmuel Kallner --- test/e2e/setup_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/e2e/setup_test.go b/test/e2e/setup_test.go index f9956e21e2..6fa0636db6 100644 --- a/test/e2e/setup_test.go +++ b/test/e2e/setup_test.go @@ -188,10 +188,9 @@ func createEndPointPickerHelper(eppConfig string, replicas int, isLeaderElection if waitForReady { return append(objects, testutils.CreateObjsFromYaml(testConfig, eppYamls)...) - } else { - objs := testutils.CreateUnstructuredObjs(testConfig, eppYamls) - return append(objects, testutils.CreateObjsWithVerifier(testConfig, objs, func(kind string, clientObj client.Object) {})...) } + objs := testutils.CreateUnstructuredObjs(testConfig, eppYamls) + return append(objects, testutils.CreateObjsWithVerifier(testConfig, objs, func(kind string, clientObj client.Object) {})...) } func usesTokenProducer(eppConfig string) bool {