feat(conformance) add EPP unavailable fail-open test (#999)

* Add test for epp becoming unavailable and the extensionRef.failureMode is set to failOpen.

* resolve minor comments.

* format.

* import format.

Author: zetxqx

conformance/resources/manifests/manifests.yaml

@@ -110,6 +110,7 @@ metadata:
   labels:
     app: secondary-inference-model-server
 spec:
+  replicas: 3
   selector:
     matchLabels:
       app: secondary-inference-model-server
@@ -245,6 +246,7 @@ spec:
   targetPortNumber: 3000
   extensionRef:
     name: secondary-endpoint-picker-svc
+    failureMode: FailOpen
 ---
 # --- Secondary Conformance EPP service Definition ---
 apiVersion: v1

conformance/tests/basic/epp_unavailable_fail_open.go

@@ -0,0 +1,117 @@
+/*
+Copyright 2025 The Kubernetes 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 basic
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/gateway-api/conformance/utils/suite"
+	"sigs.k8s.io/gateway-api/pkg/features"
+
+	"sigs.k8s.io/gateway-api-inference-extension/conformance/tests"
+	k8sutils "sigs.k8s.io/gateway-api-inference-extension/conformance/utils/kubernetes"
+	trafficutils "sigs.k8s.io/gateway-api-inference-extension/conformance/utils/traffic"
+)
+
+func init() {
+	tests.ConformanceTests = append(tests.ConformanceTests, EppUnAvailableFailOpen)
+}
+
+var EppUnAvailableFailOpen = suite.ConformanceTest{
+	ShortName:   "EppUnAvailableFailOpen",
+	Description: "Inference gateway should send traffic to backends even when the EPP is unavailable (fail-open)",
+	Manifests:   []string{"tests/basic/epp_unavailable_fail_open.yaml"},
+	Features: []features.FeatureName{
+		features.FeatureName("SupportInferencePool"),
+		features.SupportGateway,
+	},
+	Test: func(t *testing.T, s *suite.ConformanceTestSuite) {
+		const (
+			appBackendNamespace    = "gateway-conformance-app-backend"
+			infraNamespace         = "gateway-conformance-infra"
+			hostname               = "secondary.example.com"
+			path                   = "/failopen-pool-test"
+			expectedPodReplicas    = 3
+			eppSelectionHeaderName = "test-epp-endpoint-selection"
+			appPodBackendPrefix    = "secondary-inference-model-server"
+			requestBody            = `{
+                "model": "conformance-fake-model",
+                "prompt": "Write as if you were a critic: San Francisco"
+            }`
+		)
+
+		httpRouteNN := types.NamespacedName{Name: "httproute-for-failopen-pool-gw", Namespace: appBackendNamespace}
+		gatewayNN := types.NamespacedName{Name: "conformance-secondary-gateway", Namespace: infraNamespace}
+		poolNN := types.NamespacedName{Name: "secondary-inference-pool", Namespace: appBackendNamespace}
+		eppDeploymentNN := types.NamespacedName{Name: "secondary-app-endpoint-picker", Namespace: appBackendNamespace}
+		backendPodLabels := map[string]string{"app": "secondary-inference-model-server"}
+
+		k8sutils.HTTPRouteMustBeAcceptedAndResolved(t, s.Client, s.TimeoutConfig, httpRouteNN, gatewayNN)
+		k8sutils.InferencePoolMustBeAcceptedByParent(t, s.Client, poolNN)
+		gwAddr := k8sutils.GetGatewayEndpoint(t, s.Client, s.TimeoutConfig, gatewayNN)
+
+		pods, err := k8sutils.GetPodsWithLabel(t, s.Client, appBackendNamespace, backendPodLabels)
+		require.NoError(t, err, "Failed to get backend pods")
+		require.Len(t, pods, expectedPodReplicas, "Expected to find %d backend pod, but found %d.", expectedPodReplicas, len(pods))
+
+		targetPodIP := pods[0].Status.PodIP
+		t.Run("Phase 1: Verify baseline connectivity with EPP available", func(t *testing.T) {
+			t.Log("Sending request to ensure the Gateway and EPP are working correctly...")
+			trafficutils.MakeRequestWithRequestParamAndExpectSuccess(
+				t,
+				s.RoundTripper,
+				s.TimeoutConfig,
+				gwAddr,
+				trafficutils.Request{
+					Host:      hostname,
+					Path:      path,
+					Headers:   map[string]string{eppSelectionHeaderName: targetPodIP},
+					Method:    http.MethodPost,
+					Body:      requestBody,
+					Backend:   pods[0].Name, // Make sure the request is from the targetPod when the EPP is alive.
+					Namespace: appBackendNamespace,
+				},
+			)
+		})
+
+		t.Run("Phase 2: Verify fail-open behavior after EPP becomes unavailable", func(t *testing.T) {
+			t.Log("Simulating an EPP failure by deleting its deployment...")
+			deleteErr := k8sutils.DeleteDeployment(t, s.Client, s.TimeoutConfig, eppDeploymentNN)
+			require.NoError(t, deleteErr, "Failed to delete the EPP deployment")
+
+			t.Log("Sending request again, expecting success to verify fail-open...")
+			trafficutils.MakeRequestWithRequestParamAndExpectSuccess(
+				t,
+				s.RoundTripper,
+				s.TimeoutConfig,
+				gwAddr,
+				trafficutils.Request{
+					Host:      hostname,
+					Path:      path,
+					Headers:   map[string]string{eppSelectionHeaderName: targetPodIP},
+					Method:    http.MethodPost,
+					Body:      requestBody,
+					Backend:   appPodBackendPrefix, // Only checks the prefix since the EPP is not alive and the response can return from any Pod.
+					Namespace: appBackendNamespace,
+				},
+			)
+		})
+	},
+}

conformance/tests/basic/epp_unavailable_fail_open.yaml

@@ -0,0 +1,36 @@
+# --- InferenceModel Definition ---
+# TODO: remove inferenceModel dependency https://github.com/kubernetes-sigs/gateway-api-inference-extension/issues/1002
+apiVersion: inference.networking.x-k8s.io/v1alpha2
+kind: InferenceModel
+metadata:
+  name: conformance-fake-model-server
+  namespace: gateway-conformance-app-backend
+spec:
+  modelName: conformance-fake-model
+  criticality: Critical # Mark it as critical to bypass the saturation check since the model server is fake and don't have such metrics. 
+  poolRef:
+    name: secondary-inference-pool
+---
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: httproute-for-failopen-pool-gw
+  namespace: gateway-conformance-app-backend
+spec:
+  parentRefs:
+  - group: gateway.networking.k8s.io
+    kind: Gateway
+    name: conformance-secondary-gateway
+    namespace: gateway-conformance-infra
+    sectionName: http
+  hostnames:
+  - "secondary.example.com"
+  rules:
+  - backendRefs:
+    - group: inference.networking.x-k8s.io
+      kind: InferencePool
+      name: secondary-inference-pool # Use secondary-inferencePool because it has failureMode set to failOpen
+    matches:
+    - path:
+        type: PathPrefix
+        value: /failopen-pool-test

conformance/utils/kubernetes/helpers.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -361,11 +362,50 @@ func GetPod(t *testing.T, c client.Client, namespace string, selector labels.Sel
 		}
 		return false, nil
 	})
-
 	require.NoErrorf(t, waitErr, "timed out waiting for Pod with selector %s in namespace %s to be ready", selector.String(), namespace)
 	require.NotEmpty(t, pods.Items, "expected at least one pod for selector %s in namespace %s, but found none", selector.String(), namespace)
 
 	pod := &pods.Items[0]
 	t.Logf("Successfully found ready Pod %s with IP %s for selector %s", pod.Name, pod.Status.PodIP, selector.String())
 	return pod
 }
+
+// DeleteDeployment deletes the specified Deployment and waits until it is no longer
+// present in the cluster.
+func DeleteDeployment(t *testing.T, c client.Client, timeoutConfig gatewayapiconfig.TimeoutConfig, deploymentRef types.NamespacedName) error {
+	t.Helper()
+
+	deploymentToDelete := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      deploymentRef.Name,
+			Namespace: deploymentRef.Namespace,
+		},
+	}
+
+	t.Logf("Deleting Deployment %s/%s...", deploymentRef.Namespace, deploymentRef.Name)
+	if err := c.Delete(context.Background(), deploymentToDelete); err != nil {
+		// If the resource is already gone, we don't consider it an error.
+		if !apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to delete Deployment %s/%s: %w", deploymentRef.Namespace, deploymentRef.Name, err)
+		}
+	}
+
+	// Wait for the Deployment to be fully removed.
+	waitErr := wait.PollUntilContextTimeout(context.Background(), 1*time.Second, timeoutConfig.DeleteTimeout, true, func(ctx context.Context) (bool, error) {
+		var dep appsv1.Deployment
+		err := c.Get(ctx, deploymentRef, &dep)
+		if apierrors.IsNotFound(err) {
+			return true, nil
+		}
+		if err != nil {
+			return false, fmt.Errorf("error waiting for Deployment %s/%s to be deleted: %w", deploymentRef.Namespace, deploymentRef.Name, err)
+		}
+		return false, nil
+	})
+
+	if waitErr != nil {
+		return fmt.Errorf("timed out waiting for Deployment %s/%s to be deleted: %w", deploymentRef.Namespace, deploymentRef.Name, waitErr)
+	}
+	t.Logf("Successfully deleted Deployment %s/%s", deploymentRef.Namespace, deploymentRef.Name)
+	return nil
+}