test: Another step in merging the IGW and router end to end tests

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

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

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}

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

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.

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
+}