diff --git a/internal/extensionserver/extensionserver_test.go b/internal/extensionserver/extensionserver_test.go index 23157118f2..f86f2a23a9 100644 --- a/internal/extensionserver/extensionserver_test.go +++ b/internal/extensionserver/extensionserver_test.go @@ -642,7 +642,7 @@ func TestMaybeModifyClusterExtended(t *testing.T) { FilterMetadata: map[string]*structpb.Struct{ internalapi.InternalEndpointMetadataNamespace: { Fields: map[string]*structpb.Value{ - "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false"), + "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false/FailClose"), }, }, }, @@ -858,7 +858,7 @@ func TestMaybeModifyListenerAndRoutes(t *testing.T) { FilterMetadata: map[string]*structpb.Struct{ internalapi.InternalEndpointMetadataNamespace: { Fields: map[string]*structpb.Value{ - "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false"), + "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false/FailClose"), }, }, }, @@ -1246,7 +1246,7 @@ func TestPatchVirtualHostWithInferencePool(t *testing.T) { internalapi.InternalEndpointMetadataNamespace: { Fields: map[string]*structpb.Value{ "per_route_rule_inference_pool": structpb.NewStringValue( - fmt.Sprintf("%s/%s/test-epp/9002/duplex/false", pool.Namespace, pool.Name), + fmt.Sprintf("%s/%s/test-epp/9002/duplex/false/FailClose", pool.Namespace, pool.Name), ), }, }, @@ -2200,7 +2200,7 @@ func TestBuildClustersForInferencePoolEndpointPickers(t *testing.T) { FilterMetadata: map[string]*structpb.Struct{ internalapi.InternalEndpointMetadataNamespace: { Fields: map[string]*structpb.Value{ - "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false"), + "per_route_rule_inference_pool": structpb.NewStringValue("test-ns/test-pool/test-epp/9002/duplex/false/FailClose"), }, }, }, diff --git a/internal/extensionserver/inferencepool.go b/internal/extensionserver/inferencepool.go index 34eaf5aae0..72aaad210f 100644 --- a/internal/extensionserver/inferencepool.go +++ b/internal/extensionserver/inferencepool.go @@ -99,7 +99,7 @@ func getInferencePoolByMetadata(meta *corev3.Metadata) *gwaiev1.InferencePool { } result := strings.Split(metadata, "/") - if len(result) != 6 { + if len(result) != 7 { return nil } ns := result[0] @@ -111,6 +111,7 @@ func getInferencePoolByMetadata(meta *corev3.Metadata) *gwaiev1.InferencePool { } processingBodyMode := result[4] allowModeOverride := result[5] + failureMode := result[6] return &gwaiev1.InferencePool{ ObjectMeta: metav1.ObjectMeta{ Name: name, @@ -122,15 +123,16 @@ func getInferencePoolByMetadata(meta *corev3.Metadata) *gwaiev1.InferencePool { }, Spec: gwaiev1.InferencePoolSpec{ EndpointPickerRef: gwaiev1.EndpointPickerRef{ - Name: gwaiev1.ObjectName(serviceName), - Port: ptr.To(gwaiev1.Port{Number: gwaiev1.PortNumber(port)}), + Name: gwaiev1.ObjectName(serviceName), + Port: ptr.To(gwaiev1.Port{Number: gwaiev1.PortNumber(port)}), + FailureMode: gwaiev1.EndpointPickerFailureMode(failureMode), }, }, } } // buildMetadataForInferencePool adds InferencePool metadata to the cluster for reference by other components. -// encoded as a string in the format: "namespace/name/serviceName/port/bodyMode/allowModeOverride". +// encoded as a string in the format: "namespace/name/serviceName/port/bodyMode/allowModeOverride/failureMode". func buildEPPMetadataForCluster(cluster *clusterv3.Cluster, inferencePool *gwaiev1.InferencePool) { // Initialize cluster metadata structure if not present. if cluster.Metadata == nil { @@ -179,6 +181,7 @@ func buildEPPMetadata(metadata *corev3.Metadata, inferencePool *gwaiev1.Inferenc portForInferencePool(inferencePool), processingBodyMode, allowModeOverride, + string(inferencePool.Spec.EndpointPickerRef.FailureMode), ), ) } @@ -311,7 +314,11 @@ func buildHTTPFilterForInferencePool(pool *gwaiev1.InferencePool) *extprocv3.Ext }, AllowModeOverride: allowModeOverride, MessageTimeout: durationpb.New(300 * time.Second), - FailureModeAllow: false, + // FailureModeAllow controls what Envoy does when the EPP is unreachable: true lets the + // request continue (fail open), false drops it (fail close). This must mirror the + // InferencePool's own failureMode since nothing downstream of this injected filter can + // influence it. + FailureModeAllow: pool.Spec.EndpointPickerRef.FailureMode == gwaiev1.EndpointPickerFailOpen, } } diff --git a/internal/extensionserver/inferencepool_test.go b/internal/extensionserver/inferencepool_test.go index d820ea0061..5da477ce98 100644 --- a/internal/extensionserver/inferencepool_test.go +++ b/internal/extensionserver/inferencepool_test.go @@ -60,8 +60,9 @@ func TestBuildAndParseMetadata_RoundTrip(t *testing.T) { }, Spec: gwaiev1.InferencePoolSpec{ EndpointPickerRef: gwaiev1.EndpointPickerRef{ - Name: "test-picker", - Port: ptr.To(gwaiev1.Port{Number: 9090}), + Name: "test-picker", + Port: ptr.To(gwaiev1.Port{Number: 9090}), + FailureMode: gwaiev1.EndpointPickerFailOpen, }, }, } @@ -79,8 +80,8 @@ func TestBuildAndParseMetadata_RoundTrip(t *testing.T) { assert.True(t, ok) encodedStr := val.GetStringValue() - // Check encoded format: ns/name/svc/port/mode/override - assert.Equal(t, "test-ns/test-pool/test-picker/9090/buffered/true", encodedStr) + // Check encoded format: ns/name/svc/port/mode/override/failureMode + assert.Equal(t, "test-ns/test-pool/test-picker/9090/buffered/true/FailOpen", encodedStr) // 2. Parse Metadata back to Pool parsedPool := getInferencePoolByMetadata(metadata) @@ -91,6 +92,7 @@ func TestBuildAndParseMetadata_RoundTrip(t *testing.T) { assert.Equal(t, "test-ns", parsedPool.Namespace) assert.Equal(t, "test-picker", string(parsedPool.Spec.EndpointPickerRef.Name)) assert.Equal(t, gwaiev1.PortNumber(9090), parsedPool.Spec.EndpointPickerRef.Port.Number) + assert.Equal(t, gwaiev1.EndpointPickerFailOpen, parsedPool.Spec.EndpointPickerRef.FailureMode) // Verify restored annotations extractedMode := parsedPool.Annotations[processingBodyModeAnnotation] @@ -125,7 +127,7 @@ func TestGetInferencePoolByMetadata_Malformed(t *testing.T) { FilterMetadata: map[string]*structpb.Struct{ internalapi.InternalEndpointMetadataNamespace: { Fields: map[string]*structpb.Value{ - internalMetadataInferencePoolKey: structpb.NewStringValue("ns/name/svc/not-a-port/duplex/false"), + internalMetadataInferencePoolKey: structpb.NewStringValue("ns/name/svc/not-a-port/duplex/false/FailClose"), }, }, }, @@ -151,4 +153,37 @@ func TestBuildHTTPFilterForInferencePool_Defaults(t *testing.T) { assert.Equal(t, extprocv3.ProcessingMode_FULL_DUPLEX_STREAMED, filter.ProcessingMode.RequestBodyMode) assert.Equal(t, extprocv3.ProcessingMode_FULL_DUPLEX_STREAMED, filter.ProcessingMode.ResponseBodyMode) assert.False(t, filter.AllowModeOverride) + // Unset failureMode defaults to FailClose, so the injected filter must not fail open. + assert.False(t, filter.FailureModeAllow) +} + +// TestBuildHTTPFilterForInferencePool_FailureMode verifies that the injected ext_proc filter's +// FailureModeAllow mirrors the InferencePool's endpointPickerRef.failureMode, so that a pool +// configured with FailOpen actually degrades to normal load balancing when the EPP is unreachable +// instead of dropping traffic like FailClose. +func TestBuildHTTPFilterForInferencePool_FailureMode(t *testing.T) { + tests := []struct { + name string + failureMode gwaiev1.EndpointPickerFailureMode + want bool + }{ + {name: "FailOpen", failureMode: gwaiev1.EndpointPickerFailOpen, want: true}, + {name: "FailClose", failureMode: gwaiev1.EndpointPickerFailClose, want: false}, + {name: "unset", failureMode: "", want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pool := &gwaiev1.InferencePool{ + ObjectMeta: metav1.ObjectMeta{Name: "pool", Namespace: "ns"}, + Spec: gwaiev1.InferencePoolSpec{ + EndpointPickerRef: gwaiev1.EndpointPickerRef{ + Name: "picker", + FailureMode: tt.failureMode, + }, + }, + } + filter := buildHTTPFilterForInferencePool(pool) + assert.Equal(t, tt.want, filter.FailureModeAllow) + }) + } } diff --git a/internal/extensionserver/post_cluster_modify.go b/internal/extensionserver/post_cluster_modify.go index ca0123f43f..511e9aeab6 100644 --- a/internal/extensionserver/post_cluster_modify.go +++ b/internal/extensionserver/post_cluster_modify.go @@ -19,8 +19,8 @@ import ( ) // clusterRefInferencePool generates a unique reference for an InferencePool cluster. -func clusterRefInferencePool(namespace, name, serviceName string, servicePort uint32, bodyMode string, allowModeOverride string) string { - return fmt.Sprintf("%s/%s/%s/%d/%s/%s", namespace, name, serviceName, servicePort, bodyMode, allowModeOverride) +func clusterRefInferencePool(namespace, name, serviceName string, servicePort uint32, bodyMode, allowModeOverride, failureMode string) string { + return fmt.Sprintf("%s/%s/%s/%d/%s/%s/%s", namespace, name, serviceName, servicePort, bodyMode, allowModeOverride, failureMode) } // PostClusterModify is called by Envoy Gateway to allow extensions to modify clusters after they are generated.