test: Parallelize the End to End tests - Part one infrastructure (llm-d#1976)

* Removed unused service

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Added ginkgo CLI to the builder container

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Removed the namespace from the TestConfig struct

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Namespace name is now a parameter to utility functions

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Create envoy and metrics ports based on the number of parallel processes

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Use envoy and metrics port based on process number

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Update Cleanup calls with new parameters

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Fixed lint issues

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Updated co-ordinator tests with new utility interfaces

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Check that E2E_NUM_PROCS equals the total number of processes

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* If only one process is being used, dump using the basse namespace

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Comment corrected

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Update from review comment

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

* Added comments to descibe important helper functions

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

---------

Signed-off-by: Shmuel Kallner <kallner@il.ibm.com>

Author: shmuelk

Dockerfile.builder

@@ -63,6 +63,11 @@ ENV ENVTEST_K8S_VERSION=${ENVTEST_K8S_VERSION}
 # /usr/local/bin.
 RUN GOBIN=/usr/local/bin go install golang.org/x/vuln/cmd/govulncheck@${GOVULNCHECK_VERSION}
 
+# Install the ginkgo CLI. Build-time install ensures runtime invocations under
+# --userns=keep-id / -u <uid> can use the binary without writing to root-owned
+# /usr/local/bin.
+RUN GOBIN=/usr/local/bin go install github.com/onsi/ginkgo/v2/ginkgo@v2.28.3
+
 # Go caches are mounted as volumes at runtime for persistence across image rebuilds.
 # Directories are created with open permissions so non-root users (docker -u) can write.
 ENV GOMODCACHE=/go/pkg/mod

deploy/environments/dev/e2e-infra/services.yaml

@@ -20,22 +20,6 @@ spec:
 ---
 apiVersion: v1
 kind: Service
-metadata:
-  name: e2e-epp-health
-spec:
-  selector:
-    app: e2e-epp
-  ports:
-  - name: health
-    protocol: TCP
-    port: 9003
-    targetPort: 9003
-    nodePort: 30081
-    appProtocol: http2
-  type: NodePort
----
-apiVersion: v1
-kind: Service
 metadata:
   name: e2e-epp-metrics
 spec:

test/coordinator/e2e/coordinator/coordinator_test.go

@@ -110,14 +110,14 @@ func runCoordinatorPipeline(body []byte, expectedSteps []string, expectedImages
 		if keepClusterOnFailure && ginkgo.CurrentSpecReport().Failed() {
 			return
 		}
-		testutils.DeleteObjects(testConfig, coordinator)
-		testutils.DeleteObjects(testConfig, modelServers)
-		testutils.DeleteObjects(testConfig, decodeEPP)
-		testutils.DeleteObjects(testConfig, prefillEPP)
-		testutils.DeleteObjects(testConfig, encodeEPP)
-		testutils.DeleteObjects(testConfig, decodePool)
-		testutils.DeleteObjects(testConfig, prefillPool)
-		testutils.DeleteObjects(testConfig, encodePool)
+		testutils.DeleteObjects(testConfig, coordinator, nsName)
+		testutils.DeleteObjects(testConfig, modelServers, nsName)
+		testutils.DeleteObjects(testConfig, decodeEPP, nsName)
+		testutils.DeleteObjects(testConfig, prefillEPP, nsName)
+		testutils.DeleteObjects(testConfig, encodeEPP, nsName)
+		testutils.DeleteObjects(testConfig, decodePool, nsName)
+		testutils.DeleteObjects(testConfig, prefillPool, nsName)
+		testutils.DeleteObjects(testConfig, encodePool, nsName)
 	})
 
 	// Dump coordinator logs on failure, or always when E2E_PRINT_COORDINATOR_LOGS is

test/coordinator/e2e/coordinator/e2e_suite_test.go

@@ -114,7 +114,7 @@ var _ = ginkgo.BeforeSuite(func() {
 	if k8sContext == "" {
 		setupK8sCluster()
 	}
-	testConfig = testutils.NewTestConfig(nsName, k8sContext)
+	testConfig = testutils.NewTestConfig(k8sContext)
 	setupK8sClient()
 	setupNameSpace()
 
@@ -138,7 +138,7 @@ var _ = ginkgo.ReportAfterSuite("cleanup", func(report ginkgo.Report) {
 		return
 	}
 	if len(rendererObjects) > 0 {
-		testutils.DeleteObjects(testConfig, rendererObjects)
+		testutils.DeleteObjects(testConfig, rendererObjects, nsName)
 	}
 	for _, session := range portForwardSessions {
 		session.Terminate()

test/coordinator/e2e/coordinator/pools_check_test.go

@@ -39,12 +39,12 @@ func expectedPools() []string {
 func expectAllPoolsExist() {
 	for _, name := range expectedPools() {
 		pool := &inferenceapi.InferencePool{}
-		key := types.NamespacedName{Namespace: testConfig.NsName, Name: name}
+		key := types.NamespacedName{Namespace: nsName, Name: name}
 		gomega.Eventually(func() error {
 			return testConfig.K8sClient.Get(testConfig.Context, key, pool)
 		}, readyTimeout, defaultInterval).Should(
 			gomega.Succeed(),
-			"InferencePool %q not found in namespace %q", name, testConfig.NsName,
+			"InferencePool %q not found in namespace %q", name, nsName,
 		)
 	}
 }

test/coordinator/e2e/coordinator/setup_test.go

@@ -65,7 +65,7 @@ func setupInfra() {
 	createCRDs()
 
 	ginkgo.By("Applying shared Role/epp-reader from " + baseRbacManifest)
-	_ = testutils.CreateObjsFromYaml(testConfig, testutils.ReadYaml(baseRbacManifest))
+	_ = testutils.CreateObjsFromYaml(testConfig, testutils.ReadYaml(baseRbacManifest), nsName)
 
 	ginkgo.By("Applying Envoy from " + envoyManifest)
 	applyManifest(envoyManifest, map[string]string{
@@ -79,11 +79,11 @@ func createCRDs() {
 	ginkgo.By("Installing Gateway API CRDs from " + crdGatewayAPIPath)
 	crds := e2eutil.RunKustomize(crdGatewayAPIPath)
 	crds = e2eutil.FilterKinds(crds, "ValidatingAdmissionPolicy", "ValidatingAdmissionPolicyBinding")
-	_ = testutils.CreateObjsFromYaml(testConfig, crds)
+	_ = testutils.CreateObjsFromYaml(testConfig, crds, "")
 
 	ginkgo.By("Installing GIE CRDs from " + crdGIEPath)
 	gieCRDs := e2eutil.RunKustomize(crdGIEPath)
-	_ = testutils.CreateObjsFromYaml(testConfig, gieCRDs)
+	_ = testutils.CreateObjsFromYaml(testConfig, gieCRDs, "")
 }
 
 // createEndPointPicker creates the scheduling ConfigMap and EPP Deployment (plus
@@ -122,7 +122,7 @@ func createInferencePool(phase string, toDelete bool) []string {
 
 	docs := testutils.ReadYaml(manifest)
 	docs = e2eutil.SubstituteMany(docs, eppSubstitutions())
-	return testutils.CreateObjsFromYaml(testConfig, docs)
+	return testutils.CreateObjsFromYaml(testConfig, docs, nsName)
 }
 
 // deletePoolIfExists removes the named InferencePool when present so a rerun
@@ -131,12 +131,12 @@ func createInferencePool(phase string, toDelete bool) []string {
 func deletePoolIfExists(name string) {
 	pool := &inferenceapi.InferencePool{}
 	err := testConfig.K8sClient.Get(testConfig.Context,
-		types.NamespacedName{Namespace: testConfig.NsName, Name: name}, pool)
+		types.NamespacedName{Namespace: nsName, Name: name}, pool)
 	if apierrors.IsNotFound(err) {
 		return
 	}
 	gomega.Expect(err).NotTo(gomega.HaveOccurred(), "checking InferencePool %s", name)
-	testutils.DeleteObjects(testConfig, []string{"InferencePool/" + name})
+	testutils.DeleteObjects(testConfig, []string{"InferencePool/" + name}, nsName)
 }
 
 // createModelServers deploys the vLLM encode/prefill/decode workers from the
@@ -152,7 +152,7 @@ func createModelServers(encodeReplicas, prefillReplicas, decodeReplicas int) []s
 	docs = e2eutil.SubstituteMany(docs, subs)
 	docs = e2eutil.RemoveEmptyArgs(docs)
 	docs = e2eutil.RemoveEmptyLabels(docs)
-	objects := testutils.CreateObjsFromYaml(testConfig, docs)
+	objects := testutils.CreateObjsFromYaml(testConfig, docs, nsName)
 	podsInDeploymentsReady(objects)
 	return objects
 }
@@ -180,7 +180,7 @@ func createCoordinator(config string) []string {
 	docs = e2eutil.FilterKinds(docs, "ConfigMap")
 	docs = e2eutil.SubstituteMany(docs, coordinatorSubstitutions())
 	docs = e2eutil.RemoveEmptyArgs(docs)
-	objects = append(objects, testutils.CreateObjsFromYaml(testConfig, docs)...)
+	objects = append(objects, testutils.CreateObjsFromYaml(testConfig, docs, nsName)...)
 
 	podsInDeploymentsReady(objects)
 	if k8sContext != "" {
@@ -239,7 +239,7 @@ func applyManifest(path string, subs map[string]string) []string {
 	docs := testutils.ReadYaml(path)
 	docs = e2eutil.SubstituteMany(docs, subs)
 	docs = e2eutil.RemoveEmptyArgs(docs)
-	return testutils.CreateObjsFromYaml(testConfig, docs)
+	return testutils.CreateObjsFromYaml(testConfig, docs, nsName)
 }
 
 func eppSubstitutions() map[string]string {
@@ -301,7 +301,7 @@ func createRenderer() []string {
 	docs := e2eutil.RunKustomize(rendererComponentDir)
 	docs = e2eutil.SubstituteMany(docs, rendererSubstitutions())
 	docs = e2eutil.RemoveEmptyArgs(docs)
-	objects := testutils.CreateObjsFromYaml(testConfig, docs)
+	objects := testutils.CreateObjsFromYaml(testConfig, docs, nsName)
 	podsInDeploymentsReady(objects)
 	return objects
 }

test/e2e/disruption_test.go

@@ -34,7 +34,7 @@ var disruptionClient = &http.Client{Timeout: 10 * time.Second}
 // sendRawCompletion sends a completion request and returns the HTTP status code.
 func sendRawCompletion() (int, error) {
 	body := fmt.Sprintf(`{"model":"%s","prompt":"%s","max_tokens":10}`, simModelName, simplePrompt)
-	req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://localhost:%s/v1/completions", port), strings.NewReader(body))
+	req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://localhost:%d/v1/completions", getPort()), strings.NewReader(body))
 	if err != nil {
 		return 0, err
 	}
@@ -85,7 +85,7 @@ func eppPodReady(oldPodName string) func() bool {
 
 // completionRoutedToNamespace sends one completion and reports an error on any
 // failure or namespace-header mismatch, for use inside Eventually blocks.
-func completionRoutedToNamespace() error {
+func completionRoutedToNamespace(nsName string) error {
 	nsHdr, _, err := tryCompletion(simplePrompt, simModelName)
 	if err != nil {
 		return err
@@ -101,11 +101,13 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 		ginkgo.It("should recover and route to surviving pods", func() {
 			infPoolObjects = createInferencePool(1, true)
 
+			nsName := getNamespace()
+
 			modelServers := createModelServersDecode(2)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers, nsName)
 
 			epp := createEndPointPicker(simpleConfig)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp, nsName)
 
 			prefillPods, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector)
 			gomega.Expect(prefillPods).Should(gomega.BeEmpty())
@@ -145,7 +147,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			}, readyTimeout, 2*time.Second).Should(gomega.Equal(2))
 
 			ginkgo.By("Verifying requests succeed consistently after recovery")
-			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).
+			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).WithArguments(nsName).
 				MustPassRepeatedly(3).Should(gomega.Succeed())
 		})
 	})
@@ -154,11 +156,13 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 		ginkgo.It("should not hang and should recover routing", func() {
 			infPoolObjects = createInferencePool(1, true)
 
+			nsName := getNamespace()
+
 			modelServers := createModelServersDecode(2)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers, nsName)
 
 			epp := createEndPointPicker(simpleConfig)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp, nsName)
 
 			prefillPods, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector)
 			gomega.Expect(prefillPods).Should(gomega.BeEmpty())
@@ -195,7 +199,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			}, readyTimeout, 2*time.Second).Should(gomega.Equal(2))
 
 			ginkgo.By("Verifying requests succeed consistently after recovery")
-			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).
+			gomega.Eventually(completionRoutedToNamespace, eppRecoveryTimeout, 1*time.Second).WithArguments(nsName).
 				MustPassRepeatedly(3).Should(gomega.Succeed())
 		})
 	})
@@ -204,11 +208,13 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 		ginkgo.It("should return 503 to the client", func() {
 			infPoolObjects = createInferencePool(1, true)
 
+			nsName := getNamespace()
+
 			modelServers := createModelServersDecode(1)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers, nsName)
 
 			epp := createEndPointPicker(simpleConfig)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp, nsName)
 
 			_, decodePods := getModelServerPods(podSelector, prefillSelector, decodeSelector)
 			gomega.Expect(decodePods).Should(gomega.HaveLen(1))
@@ -218,7 +224,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			gomega.Expect(nsHdr).Should(gomega.Equal(nsName))
 
 			ginkgo.By("Scaling deployment to zero")
-			scaleDeployment(modelServers, -1)
+			scaleDeployment(nsName, modelServers, -1)
 
 			ginkgo.By("Waiting for all pods to be removed")
 			gomega.Eventually(func() int {
@@ -236,7 +242,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			}, trafficProbeTimeout, 500*time.Millisecond).Should(gomega.Equal(http.StatusServiceUnavailable))
 
 			ginkgo.By("Scaling deployment back up")
-			scaleDeployment(modelServers, 1)
+			scaleDeployment(nsName, modelServers, 1)
 
 			ginkgo.By("Verifying requests succeed after recovery")
 			gomega.Eventually(func() string {
@@ -250,15 +256,17 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 		ginkgo.It("should recover and resume routing after restart", func() {
 			infPoolObjects = createInferencePool(1, true)
 
+			nsName := getNamespace()
+
 			modelServers := createModelServersDecode(1)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers, nsName)
 
 			epp := createEndPointPicker(simpleConfig)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp, nsName)
 
 			ginkgo.By("Verifying requests succeed before EPP disruption")
 			nsHdr, _, _ := runCompletion(simplePrompt, simModelName)
-			gomega.Expect(nsHdr).Should(gomega.Equal(nsName))
+			gomega.Expect(nsHdr).Should(gomega.Equal(getNamespace()))
 
 			ginkgo.By("Finding EPP pod")
 			eppPods := getPods(map[string]string{"app": "e2e-epp"})
@@ -298,16 +306,17 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 	ginkgo.When("Traffic is flowing during scale-to-zero and back", func() {
 		ginkgo.It("should return 503s when empty and recover when scaled back", func() {
 			infPoolObjects = createInferencePool(1, true)
+			nsName := getNamespace()
 
 			modelServers := createModelServersDecode(1)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, modelServers, nsName)
 
 			epp := createEndPointPicker(simpleConfig)
-			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp)
+			ginkgo.DeferCleanup(testutils.DeleteObjects, testConfig, epp, nsName)
 
 			ginkgo.By("Verifying requests succeed before disruption")
 			nsHdr, _, _ := runCompletion(simplePrompt, simModelName)
-			gomega.Expect(nsHdr).Should(gomega.Equal(nsName))
+			gomega.Expect(nsHdr).Should(gomega.Equal(getNamespace()))
 
 			ginkgo.By("Starting background traffic")
 			ctx, cancel := context.WithCancel(context.Background())
@@ -319,7 +328,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			}()
 
 			ginkgo.By("Scaling to zero")
-			scaleDeployment(modelServers, -1)
+			scaleDeployment(nsName, modelServers, -1)
 
 			ginkgo.By("Waiting for all pods to be removed")
 			gomega.Eventually(func() int {
@@ -331,7 +340,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 			gomega.Eventually(tc.failures, trafficProbeTimeout, 500*time.Millisecond).Should(gomega.BeNumerically(">", 0))
 
 			ginkgo.By("Scaling back to 1")
-			scaleDeployment(modelServers, 1)
+			scaleDeployment(nsName, modelServers, 1)
 
 			ginkgo.By("Waiting for traffic to observe recovery")
 			successBaseline := tc.successes()
@@ -351,7 +360,7 @@ var _ = ginkgo.Describe("Disruption tests", ginkgo.Ordered, ginkgo.Label(disrupt
 func sendStreamingCompletion(connected chan<- string) error {
 	longPrompt := strings.Repeat("This is a longer prompt to keep the stream open. ", 20)
 	body := fmt.Sprintf(`{"model":"%s","prompt":"%s","max_tokens":100,"stream":true}`, simModelName, longPrompt)
-	req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://localhost:%s/v1/completions", port), strings.NewReader(body))
+	req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("http://localhost:%d/v1/completions", getPort()), strings.NewReader(body))
 	if err != nil {
 		connected <- ""
 		return err
@@ -418,7 +427,7 @@ func deletePodByName(podName string, gracePeriodSeconds int64) {
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      podName,
-			Namespace: nsName,
+			Namespace: getNamespace(),
 		},
 	}
 	opts := &client.DeleteOptions{