A75: Add missing E2E test cases for xDS aggregate cluster

internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go

@@ -38,20 +38,25 @@ import (
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils/pickfirst"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/testutils/xds/fakeserver"
 	"google.golang.org/grpc/internal/xds/bootstrap"
 	"google.golang.org/grpc/internal/xds/xdsclient"
 	"google.golang.org/grpc/internal/xds/xdsclient/xdsresource/version"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/testing/protocmp"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
 	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
 	v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
+	v3lrspb "github.com/envoyproxy/go-control-plane/envoy/service/load_stats/v3"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
 
@@ -1282,3 +1287,181 @@ func addrsToEndpoints(addrs []resolver.Address) []resolver.Endpoint {
 	}
 	return endpoints
 }
+
+// Tests that when an aggregate cluster (Root -> [PrimaryEDS, SecondaryEDS]) has
+// LRS enabled on both leaf clusters:
+//  1. LRS reports are sent only for the primary leaf cluster while healthy.
+//  2. When the primary leaf cluster fails, traffic shifts to the secondary leaf
+//     cluster and LRS stats are reported for the secondary leaf cluster.
+func (s) TestAggregateCluster_LRS(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	const (
+		clusterName1 = clusterName + "-cluster-1"
+		clusterName2 = clusterName + "-cluster-2"
+	)
+
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		AllowResourceSubset:         true,
+		SupportLoadReportingService: true,
+	})
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	servers, cleanup := startTestServiceBackends(t, 2)
+	defer cleanup()
+	_, ports := backendAddressesAndPorts(t, servers)
+
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(serviceName, routeName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig(routeName, serviceName, clusterName)},
+		Clusters: []*v3clusterpb.Cluster{
+			makeAggregateClusterResource(clusterName, []string{clusterName1, clusterName2}),
+			e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+				ClusterName: clusterName1,
+				Type:        e2e.ClusterTypeEDS,
+				EnableLRS:   true,
+			}),
+			e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+				ClusterName: clusterName2,
+				Type:        e2e.ClusterTypeEDS,
+				EnableLRS:   true,
+			}),
+		},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{
+			e2e.DefaultEndpoint(clusterName1, "localhost", []uint32{uint32(ports[0])}),
+			e2e.DefaultEndpoint(clusterName2, "localhost", []uint32{uint32(ports[1])}),
+		},
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	cc, cleanup := setupAndDial(t, bootstrapContents)
+	defer cleanup()
+
+	client := testgrpc.NewTestServiceClient(cc)
+
+	// Ensure LRS stream is opened.
+	if _, err := managementServer.LRSServer.LRSStreamOpenChan.Receive(ctx); err != nil {
+		t.Fatalf("Timeout waiting for LRS stream open: %v", err)
+	}
+	if _, err := managementServer.LRSServer.LRSRequestChan.Receive(ctx); err != nil {
+		t.Fatalf("Timeout waiting for initial LRS request: %v", err)
+	}
+
+	// Make RPC calls and verify every call reaches the primary cluster
+	// (clusterName1).
+	for i := 0; i < 10; i++ {
+		peer := &peer.Peer{}
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(peer)); err != nil {
+			t.Fatalf("EmptyCall() failed: %v", err)
+		}
+		if got, want := peer.Addr.String(), servers[0].Address; got != want {
+			t.Fatalf("EmptyCall() call #%d routed to %q, want %q", i, got, want)
+		}
+	}
+
+	// Send LRS response requesting load reporting every 10ms.
+	managementServer.LRSServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: durationpb.New(defaultLoadReportingInterval),
+		},
+	}
+
+	// Verify LRS reports load for clusterName1 and 0 load for clusterName2.
+	wantClusterStats := &v3endpointpb.ClusterStats{
+		ClusterName: clusterName1,
+		UpstreamLocalityStats: []*v3endpointpb.UpstreamLocalityStats{
+			{
+				Locality:                &v3corepb.Locality{Region: "region-1", Zone: "zone-1", SubZone: "subzone-1"},
+				TotalSuccessfulRequests: 10,
+				TotalIssuedRequests:     10,
+			},
+		},
+	}
+
+waitForPrimaryStats:
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatalf("Timeout waiting for LRS load report for %q", clusterName1)
+		case req := <-managementServer.LRSServer.LRSRequestChan.C:
+			loadStats := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest)
+			if len(loadStats.ClusterStats) == 0 {
+				continue
+			}
+			opts := []cmp.Option{
+				protocmp.Transform(),
+				protocmp.IgnoreFields(&v3endpointpb.ClusterStats{}, "load_report_interval"),
+			}
+			if diff := cmp.Diff([]*v3endpointpb.ClusterStats{wantClusterStats}, loadStats.ClusterStats, opts...); diff != "" {
+				t.Fatalf("Unexpected diff in LRS ClusterStats for %q (-want +got):\n%s", clusterName1, diff)
+			}
+			break waitForPrimaryStats
+		}
+	}
+
+	// Fail primary cluster by clearing endpoints for clusterName1.
+	resources.Endpoints = []*v3endpointpb.ClusterLoadAssignment{
+		e2e.DefaultEndpoint(clusterName1, "localhost", nil),
+		e2e.DefaultEndpoint(clusterName2, "localhost", []uint32{uint32(ports[1])}),
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Make RPCs and verify failover to secondary backend.
+	peer := &peer.Peer{}
+	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(peer)); err == nil && peer.Addr.String() == servers[1].Address {
+			break
+		}
+	}
+	if ctx.Err() != nil {
+		t.Fatalf("Timeout waiting for RPCs to switch to secondary backend %q", servers[1].Address)
+	}
+
+	// Verify LRS report now includes load for secondary cluster (clusterName2).
+	wantClusterStats = &v3endpointpb.ClusterStats{
+		ClusterName: clusterName2,
+		UpstreamLocalityStats: []*v3endpointpb.UpstreamLocalityStats{
+			{
+				Locality:                &v3corepb.Locality{Region: "region-1", Zone: "zone-1", SubZone: "subzone-1"},
+				TotalSuccessfulRequests: 1,
+			},
+		},
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			t.Fatalf("Timeout waiting for LRS load report for secondary cluster %q", clusterName2)
+		case req := <-managementServer.LRSServer.LRSRequestChan.C:
+			loadStats := req.(*fakeserver.Request).Req.(*v3lrspb.LoadStatsRequest)
+			for _, load := range loadStats.ClusterStats {
+				if load.ClusterName != clusterName2 {
+					continue
+				}
+				// The periodic LRS timer may fire while a failover RPC is
+				// in flight, producing a transitional report with 0
+				// successful requests. Skip such reports until the report
+				// with the completed RPC arrives.
+				if len(load.UpstreamLocalityStats) == 0 || load.UpstreamLocalityStats[0].TotalSuccessfulRequests == 0 {
+					continue
+				}
+				opts := []cmp.Option{
+					protocmp.Transform(),
+					protocmp.IgnoreFields(&v3endpointpb.ClusterStats{}, "load_report_interval"),
+					protocmp.IgnoreFields(&v3endpointpb.UpstreamLocalityStats{}, "total_issued_requests", "total_requests_in_progress"),
+				}
+				if diff := cmp.Diff(wantClusterStats, load, opts...); diff != "" {
+					t.Fatalf("Unexpected diff in LRS ClusterStats for %q (-want +got):\n%s", clusterName2, diff)
+				}
+				return
+			}
+		}
+	}
+}

internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go

@@ -72,6 +72,7 @@ const (
 	defaultTestTimeout            = 10 * time.Second
 	defaultTestShortTimeout       = 10 * time.Millisecond
 	defaultTestWatchExpiryTimeout = 500 * time.Millisecond
+	defaultLoadReportingInterval  = 10 * time.Millisecond
 )
 
 type s struct {

test/xds/xds_telemetry_labels_test.go

@@ -21,21 +21,27 @@ import (
 	"context"
 	"fmt"
 	"testing"
+	"time"
 
+	"github.com/google/go-cmp/cmp"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
-	telemetry "google.golang.org/grpc/experimental/stats/telemetry"
+	"google.golang.org/grpc/experimental/stats/telemetry"
 	"google.golang.org/grpc/internal/stubserver"
 	"google.golang.org/grpc/internal/testutils"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
 	"google.golang.org/grpc/internal/testutils/xds/e2e/setup"
-	testgrpc "google.golang.org/grpc/interop/grpc_testing"
-	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/stats"
+	"google.golang.org/protobuf/types/known/structpb"
 
+	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	"github.com/google/go-cmp/cmp"
-	"google.golang.org/protobuf/types/known/structpb"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
 )
 
 const serviceNameKey = "service_name"
@@ -102,6 +108,115 @@ func (s) TestTelemetryLabels(t *testing.T) {
 	}
 }
 
+// Tests that telemetry labels for an aggregate cluster hierarchy reflect the
+// active leaf cluster receiving traffic, and correctly switch labels when
+// failing over to a secondary leaf cluster.
+func (s) TestTelemetryLabels_AggregateCluster(t *testing.T) {
+	managementServer, nodeID, _, xdsResolver := setup.ManagementServerAndResolver(t)
+
+	const (
+		xdsServiceName = "my-service-client-side-xds"
+		cluster1Name   = "cluster-1"
+		cluster2Name   = "cluster-2"
+
+		csmName1 = "service-1"
+		csmNs1   = "namespace-1"
+		csmName2 = "service-2"
+		csmNs2   = "namespace-2"
+	)
+
+	servers := make([]*stubserver.StubServer, 2)
+	for i := range 2 {
+		servers[i] = stubserver.StartTestService(t, nil)
+		defer servers[i].Stop()
+	}
+
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{e2e.DefaultClientListener(xdsServiceName, "route-"+xdsServiceName)},
+		Routes:    []*v3routepb.RouteConfiguration{e2e.DefaultRouteConfig("route-"+xdsServiceName, xdsServiceName, xdsServiceName)},
+		Clusters: []*v3clusterpb.Cluster{
+			e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+				ClusterName: xdsServiceName,
+				Type:        e2e.ClusterTypeAggregate,
+				ChildNames:  []string{cluster1Name, cluster2Name},
+			}),
+			makeClusterResourceWithMetadata(cluster1Name, csmName1, csmNs1),
+			makeClusterResourceWithMetadata(cluster2Name, csmName2, csmNs2),
+		},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{
+			e2e.DefaultEndpoint(cluster1Name, "localhost", []uint32{uint32(testutils.ParsePort(t, servers[0].Address))}),
+			e2e.DefaultEndpoint(cluster2Name, "localhost", []uint32{uint32(testutils.ParsePort(t, servers[1].Address))}),
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", xdsServiceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
+	if err != nil {
+		t.Fatalf("failed to create a new client to local test server: %v", err)
+	}
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+
+	// Make RPC to primary cluster and verify primary telemetry labels.
+	peer := &peer.Peer{}
+	var gotLabels map[string]string
+	callCtx := telemetry.NewContextWithLabelCallback(ctx, func(l map[string]string) {
+		gotLabels = l
+	})
+	if _, err := client.EmptyCall(callCtx, &testpb.Empty{}, grpc.Peer(peer)); err != nil {
+		t.Fatalf("EmptyCall() failed: %v", err)
+	}
+	if got, want := peer.Addr.String(), servers[0].Address; got != want {
+		t.Fatalf("EmptyCall() routed to %q, want %q", got, want)
+	}
+
+	wantLabels := map[string]string{
+		localityKey:       localityValue,
+		backendServiceKey: cluster1Name,
+	}
+	if diff := cmp.Diff(gotLabels, wantLabels); diff != "" {
+		t.Fatalf("Telemetry labels for primary cluster (-got +want): %v", diff)
+	}
+
+	// Trigger failover to secondary cluster by clearing primary endpoints.
+	resources.Endpoints = []*v3endpointpb.ClusterLoadAssignment{
+		e2e.DefaultEndpoint(cluster1Name, "localhost", nil),
+		e2e.DefaultEndpoint(cluster2Name, "localhost", []uint32{uint32(testutils.ParsePort(t, servers[1].Address))}),
+	}
+	if err := managementServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	// Make RPCs until traffic switches to secondary cluster and capture
+	// secondary labels.
+	for ; ctx.Err() == nil; <-time.After(defaultTestShortTimeout) {
+		callCtx := telemetry.NewContextWithLabelCallback(ctx, func(l map[string]string) {
+			gotLabels = l
+		})
+		if _, err := client.EmptyCall(callCtx, &testpb.Empty{}, grpc.Peer(peer)); err == nil && peer.Addr.String() == servers[1].Address {
+			break
+		}
+	}
+	if ctx.Err() != nil {
+		t.Fatalf("Timeout waiting for RPCs to switch to secondary cluster %q", servers[1].Address)
+	}
+
+	wantLabels = map[string]string{
+		localityKey:       localityValue,
+		backendServiceKey: cluster2Name,
+	}
+	if diff := cmp.Diff(gotLabels, wantLabels); diff != "" {
+		t.Fatalf("Telemetry labels after failover (-got +want): %v", diff)
+	}
+}
+
 type fakeStatsHandler struct {
 	labels map[string]string
 
@@ -146,3 +261,21 @@ func (fsh *fakeStatsHandler) HandleRPC(_ context.Context, rs stats.RPCStats) {
 
 	}
 }
+
+func makeClusterResourceWithMetadata(clusterName, serviceName, serviceNamespace string) *v3clusterpb.Cluster {
+	cluster := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
+		ClusterName: clusterName,
+		Type:        e2e.ClusterTypeEDS,
+	})
+	cluster.Metadata = &v3corepb.Metadata{
+		FilterMetadata: map[string]*structpb.Struct{
+			"com.google.csm.telemetry_labels": {
+				Fields: map[string]*structpb.Value{
+					serviceNameKey:      structpb.NewStringValue(serviceName),
+					serviceNamespaceKey: structpb.NewStringValue(serviceNamespace),
+				},
+			},
+		},
+	}
+	return cluster
+}