Conformance test for GRPCRouteRequestMirror #3514

conformance/tests/grpcroute-request-mirror.go

@@ -0,0 +1,111 @@
+/*
+Copyright 2023 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 tests
+
+import (
+	"testing"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/metadata"
+	"k8s.io/apimachinery/pkg/types"
+	v1 "sigs.k8s.io/gateway-api/apis/v1"
+	pb "sigs.k8s.io/gateway-api/conformance/echo-basic/grpcechoserver"
+	"sigs.k8s.io/gateway-api/conformance/utils/grpc"
+	"sigs.k8s.io/gateway-api/conformance/utils/kubernetes"
+	"sigs.k8s.io/gateway-api/conformance/utils/suite"
+	"sigs.k8s.io/gateway-api/pkg/features"
+)
+
+func init() {
+	ConformanceTests = append(ConformanceTests, GRPCRouteRequestMirror)
+}
+
+var GRPCRouteRequestMirror = suite.ConformanceTest{
+	ShortName:   "GRPCRouteRequestMirror",
+	Description: "A GRPCRoute with request mirror filter",
+	Manifests:   []string{"tests/grpcroute-request-mirror.yaml"},
+	Features: []features.FeatureName{
+		features.SupportGRPCRoute,
+		features.SupportGateway,
+		features.SupportGRPCRouteRequestMirror,
+	},
+	Test: func(t *testing.T, suite *suite.ConformanceTestSuite) {
+		ns := "gateway-conformance-infra"
+		routeNN := types.NamespacedName{Name: "grpc-request-mirror", Namespace: ns}
+		gwNN := types.NamespacedName{Name: "same-namespace", Namespace: ns}
+		gwAddr := kubernetes.GatewayAndRoutesMustBeAccepted(t, suite.Client, suite.TimeoutConfig, suite.ControllerName, kubernetes.NewGatewayRef(gwNN), &v1.GRPCRoute{}, true, routeNN)
+
+		testCases := []grpc.ExpectedResponse{
+			{
+				EchoRequest: &pb.EchoRequest{},
+				Backend:     "grpc-infra-backend-v1",
+				Namespace:   ns,
+				MirroredTo: []grpc.MirroredBackend{
+					{
+						BackendRef: grpc.BackendRef{
+							Name:      "grpc-infra-backend-v2",
+							Namespace: ns,
+						},
+					},
+				},
+				Response: grpc.Response{
+					Code: codes.OK,
+				},
+			},
+			{
+				EchoTwoRequest: &pb.EchoRequest{},
+				Backend:        "grpc-infra-backend-v1",
+				RequestMetadata: &grpc.RequestMetadata{
+					Metadata: map[string]string{
+						"X-Header-Remove":     "remove-val",
+						"X-Header-Add-Append": "append-val-1",
+					},
+				},
+				Namespace: ns,
+				MirroredTo: []grpc.MirroredBackend{
+					{
+						BackendRef: grpc.BackendRef{
+							Name:      "grpc-infra-backend-v2",
+							Namespace: ns,
+						},
+					},
+				},
+				Response: grpc.Response{
+					Code: codes.OK,
+					Headers: func() *metadata.MD {
+						md := metadata.Pairs(
+							"x-header-set", "set-overwrites-values",
+							"x-header-add", "header-val-1",
+							"x-header-add-append", "append-val-1",
+							"x-header-add-append", "header-val-2",
+						)
+						return &md
+					}(),
+					AbsentHeaders: []string{"X-Header-Remove"},
+				},
+			},
+		}
+		for i := range testCases {
+			tc := testCases[i]
+			t.Run(tc.GetTestCaseName(i), func(t *testing.T) {
+				t.Parallel()
+				grpc.MakeRequestAndExpectEventuallyConsistentResponse(t, suite.GRPCClient, suite.TimeoutConfig, gwAddr, tc)
+				grpc.ExpectMirroredRequest(t, suite.Client, suite.Clientset, tc.MirroredTo)
+			})
+		}
+	},
+}

conformance/tests/grpcroute-request-mirror.yaml

@@ -0,0 +1,51 @@
+apiVersion: gateway.networking.k8s.io/v1
+kind: GRPCRoute
+metadata:
+  name: grpc-request-mirror
+  namespace: gateway-conformance-infra
+spec:
+  parentRefs:
+  - name: same-namespace
+  rules:
+  - matches:
+    - method:
+        service: gateway_api_conformance.echo_basic.grpcecho.GrpcEcho
+        method: Echo
+    filters:
+    - type: RequestMirror
+      requestMirror:
+        backendRef:
+          name: grpc-infra-backend-v2
+          namespace: gateway-conformance-infra
+          port: 8080
+    backendRefs:
+    - name: grpc-infra-backend-v1
+      port: 8080
+      namespace: gateway-conformance-infra
+  - matches:
+    - method:
+        service: gateway_api_conformance.echo_basic.grpcecho.GrpcEcho
+        method: EchoTwo
+    filters:
+    - type: RequestHeaderModifier
+      requestHeaderModifier:
+        set:
+        - name: X-Header-Set
+          value: set-overwrites-values
+        add:
+        - name: X-Header-Add
+          value: header-val-1
+        - name: X-Header-Add-Append
+          value: header-val-2
+        remove:
+        - X-Header-Remove
+    - type: RequestMirror
+      requestMirror:
+        backendRef:
+          name: grpc-infra-backend-v2
+          namespace: gateway-conformance-infra
+          port: 8080
+    backendRefs:
+    - name: grpc-infra-backend-v1
+      port: 8080
+      namespace: gateway-conformance-infra

conformance/utils/grpc/grpc.go

@@ -19,6 +19,7 @@ package grpc
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sort"
 	"strings"
 	"testing"
@@ -56,10 +57,11 @@ type DefaultClient struct {
 }
 
 type Response struct {
-	Code     codes.Code
-	Headers  *metadata.MD
-	Trailers *metadata.MD
-	Response *pb.EchoResponse
+	Code          codes.Code
+	Headers       *metadata.MD
+	Trailers      *metadata.MD
+	Response      *pb.EchoResponse
+	AbsentHeaders []string
 }
 
 type RequestMetadata struct {
@@ -70,6 +72,16 @@ type RequestMetadata struct {
 	Metadata map[string]string
 }
 
+type BackendRef struct {
+	Name      string
+	Namespace string
+}
+
+type MirroredBackend struct {
+	BackendRef
+	Percent *int32
+}
+
 // ExpectedResponse defines the response expected for a given request.
 type ExpectedResponse struct {
 	// Defines the request to make. Only one of EchoRequest and EchoTwoRequest
@@ -88,6 +100,9 @@ type ExpectedResponse struct {
 	Backend   string
 	Namespace string
 
+	// MirroredTo is the destination BackendRefs of the mirrored request
+	MirroredTo []MirroredBackend
+
 	// User Given TestCase name
 	TestCaseName string
 }
@@ -249,6 +264,51 @@ func compareResponse(expected *ExpectedResponse, response *Response) error {
 		if !strings.HasPrefix(response.Response.GetAssertions().GetContext().GetPod(), expected.Backend) {
 			return fmt.Errorf("expected pod name to start with %s, got %s", expected.Backend, response.Response.GetAssertions().GetContext().GetPod())
 		}
+
+		// Check if the correct headers were received by the backend
+		receivedHeadersMap := make(map[string][]string)
+		receivedHeaders := response.Response.GetAssertions().GetHeaders()
+		for _, receivedHeader := range receivedHeaders {
+			receivedKey := strings.ToLower(receivedHeader.GetKey())
+			receivedValue := receivedHeader.GetValue()
+			receivedHeadersMap[receivedKey] = append(receivedHeadersMap[receivedKey], receivedValue)
+		}
+
+		expectedHeaders := expected.Response.Headers
+		if expectedHeaders != nil {
+
+			if receivedHeaders == nil {
+				return fmt.Errorf("No headers captured: expected %v headers", len(*expectedHeaders))
+			}
+
+			for expectedHeader, expectedValues := range *expectedHeaders {
+				expectedHeader = strings.ToLower(expectedHeader)
+				receivedValues, ok := receivedHeadersMap[expectedHeader]
+				if !ok {
+					return fmt.Errorf("expected header %s not found", expectedHeader)
+				}
+				sortedExpectedValues := slices.Clone(expectedValues)
+				sortedReceivedValues := slices.Clone(receivedValues)
+
+				slices.Sort(sortedExpectedValues)
+				slices.Sort(sortedReceivedValues)
+
+				if !slices.Equal(sortedExpectedValues, sortedReceivedValues) {
+					return fmt.Errorf("Header: %s, Expected values %v not equal to received values %v", expectedHeader, sortedExpectedValues, sortedReceivedValues)
+				}
+			}
+		}
+
+		// Check if the headers that were supposed to be removed by the Gateway are removed
+		if len(expected.Response.AbsentHeaders) > 0 {
+			for _, absentHeader := range expected.Response.AbsentHeaders {
+				val, ok := receivedHeadersMap[strings.ToLower(absentHeader)]
+				if ok {
+					return fmt.Errorf("Header: %s, should not be present, got %s", absentHeader, val)
+				}
+			}
+		}
+
 	}
 	return nil
 }

conformance/utils/grpc/mirror.go

@@ -0,0 +1,73 @@
+/*
+Copyright 2023 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 grpc
+
+import (
+	"regexp"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"sigs.k8s.io/gateway-api/conformance/utils/kubernetes"
+	"sigs.k8s.io/gateway-api/conformance/utils/tlog"
+
+	clientset "k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func ExpectMirroredRequest(t *testing.T, client client.Client, clientset clientset.Interface, mirrorPods []MirroredBackend) {
+	for i, mirrorPod := range mirrorPods {
+		if mirrorPod.Name == "" {
+			tlog.Fatalf(t, "Mirrored BackendRef[%d].Name wasn't provided in the testcase, this test should only check grpc request mirror.", i)
+		}
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(mirrorPods))
+
+	assertionStart := time.Now()
+
+	for _, mirrorPod := range mirrorPods {
+		go func(mirrorPod MirroredBackend) {
+			defer wg.Done()
+
+			require.Eventually(t, func() bool {
+				mirrorLogRegexp := regexp.MustCompile("Received over plaintext:")
+
+				tlog.Log(t, "Searching for the mirrored request log")
+				tlog.Logf(t, `Reading "%s/%s" logs`, mirrorPod.Namespace, mirrorPod.Name)
+				logs, err := kubernetes.DumpEchoLogs(mirrorPod.Namespace, mirrorPod.Name, client, clientset, assertionStart)
+				if err != nil {
+					tlog.Logf(t, `Couldn't read "%s/%s" logs: %v`, mirrorPod.Namespace, mirrorPod.Name, err)
+					return false
+				}
+
+				for _, log := range logs {
+					if mirrorLogRegexp.MatchString(log) {
+						return true
+					}
+				}
+				return false
+			}, 60*time.Second, time.Millisecond*100, `Couldn't find mirrored request in "%s/%s" logs`, mirrorPod.Namespace, mirrorPod.Name)
+		}(mirrorPod)
+	}
+
+	wg.Wait()
+
+	tlog.Log(t, "Found mirrored request log in all desired backends")
+}

pkg/features/grpcroute.go

@@ -46,17 +46,28 @@ var GRPCRouteCoreFeatures = sets.New(
 const (
 	// This option indicates support for the name field in the GRPCRouteRule (extended conformance)
 	SupportGRPCRouteNamedRouteRule FeatureName = "GRPCRouteNamedRouteRule"
+
+	// This option indicates support for GRPCRoute request mirror (extended conformance)
+	SupportGRPCRouteRequestMirror FeatureName = "GRPCRouteRequestMirror"
 )
 
-// GRPCRouteNamedRouteRule contains metadata for the SupportGRPCRouteNamedRouteRule feature.
-var GRPCRouteNamedRouteRule = Feature{
-	Name:    SupportGRPCRouteNamedRouteRule,
-	Channel: FeatureChannelStandard,
-}
+var (
+	// GRPCRouteNamedRouteRule contains metadata for the SupportGRPCRouteNamedRouteRule feature.
+	GRPCRouteNamedRouteRule = Feature{
+		Name:    SupportGRPCRouteNamedRouteRule,
+		Channel: FeatureChannelStandard,
+	}
+	// GRPCRouteRequestMirrorFeature contains metadata for the GRPCRouteRequestMirror feature.
+	GRPCRouteRequestMirrorFeature = Feature{
+		Name:    SupportGRPCRouteRequestMirror,
+		Channel: FeatureChannelStandard,
+	}
+)
 
 // GRPCRouteExtendedFeatures includes all extended features for GRPCRoute
 // conformance and can be used to opt-in to run all GRPCRoute extended features tests.
 // This does not include any Core Features.
 var GRPCRouteExtendedFeatures = sets.New(
 	GRPCRouteNamedRouteRule,
+	GRPCRouteRequestMirrorFeature,
 )