A75: Add missing E2E test cases for xDS aggregate cluster

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

@@ -38,20 +38,23 @@ 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/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
 	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/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 +1285,168 @@ 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"
+		numBackends       = 2
+		wantRPCCount      = 10
+		wantLocalityCount = 1
+	)
+
+	managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+		AllowResourceSubset:         true,
+		SupportLoadReportingService: true,
+	})
+	nodeID := uuid.New().String()
+	bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
+
+	servers, cleanup2 := startTestServiceBackends(t, numBackends)
+	defer cleanup2()
+	_, 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)
+	}
+
+	// Send LRS response requesting load reporting every 10ms.
+	managementServer.LRSServer.LRSResponseChan <- &fakeserver.Response{
+		Resp: &v3lrspb.LoadStatsResponse{
+			SendAllClusters:       true,
+			LoadReportingInterval: durationpb.New(defaultLoadReportingInterval),
+		},
+	}
+
+	// Make RPC calls and verify every call reaches the primary cluster
+	// (clusterName1).
+	for i := 0; i < wantRPCCount; i++ {
+		peer := &peer.Peer{}
+		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.Peer(peer), grpc.WaitForReady(true)); 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)
+		}
+	}
+
+	// Verify LRS reports load for clusterName1 and 0 load for clusterName2.
+	for gotCluster1Report := false; !gotCluster1Report; {
+		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)
+			for _, load := range loadStats.ClusterStats {
+				// For unused leaf clusters (when SendAllClusters: true), LRS
+				// reports are expected to be empty (0 successful requests
+				// across any locality).
+				if load.ClusterName == clusterName2 {
+					for _, loc := range load.UpstreamLocalityStats {
+						if loc.TotalSuccessfulRequests > 0 {
+							t.Fatalf("Received unexpected LRS load report for secondary cluster %q: %+v", clusterName2, load)
+						}
+					}
+				}
+				if load.ClusterName == clusterName1 {
+					// Each leaf cluster has exactly one locality.
+					if len(load.UpstreamLocalityStats) != wantLocalityCount {
+						t.Fatalf("UpstreamLocalityStats length = %d, want %d", len(load.UpstreamLocalityStats), wantLocalityCount)
+					}
+					if load.UpstreamLocalityStats[0].TotalSuccessfulRequests > 0 {
+						gotCluster1Report = true
+					}
+				}
+				if load.ClusterName != clusterName1 && load.ClusterName != clusterName2 {
+					t.Fatalf("Unexpected cluster name %q", load.ClusterName)
+				}
+			}
+		}
+	}
+
+	// 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 {
+		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).
+	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 {
+					if len(load.UpstreamLocalityStats) != wantLocalityCount {
+						t.Fatalf("UpstreamLocalityStats length = %d, want %d", len(load.UpstreamLocalityStats), wantLocalityCount)
+					}
+					if load.UpstreamLocalityStats[0].TotalSuccessfulRequests > 0 {
+						return
+					}
+				}
+				if load.ClusterName != clusterName1 && load.ClusterName != clusterName2 {
+					t.Fatalf("Unexpected cluster name %q", load.ClusterName)
+				}
+			}
+		}
+	}
+}

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

@@ -22,20 +22,25 @@ import (
 	"fmt"
 	"testing"
 
+	"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 +107,117 @@ 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 (
+		numServers     = 2
+		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, numServers)
+	for i := 0; i < numServers; i++ {
+		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 cluster1Labels map[string]string
+	callCtx1 := telemetry.NewContextWithLabelCallback(ctx, func(l map[string]string) {
+		cluster1Labels = l
+	})
+	if _, err := client.EmptyCall(callCtx1, &testpb.Empty{}, grpc.Peer(peer), grpc.WaitForReady(true)); 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)
+	}
+
+	wantCluster1Labels := map[string]string{
+		localityKey:       localityValue,
+		backendServiceKey: cluster1Name,
+	}
+	if diff := cmp.Diff(cluster1Labels, wantCluster1Labels); 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.
+	var cluster2Labels map[string]string
+	for ctx.Err() == nil {
+		callCtx2 := telemetry.NewContextWithLabelCallback(ctx, func(l map[string]string) {
+			cluster2Labels = l
+		})
+		if _, err := client.EmptyCall(callCtx2, &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)
+	}
+
+	wantCluster2Labels := map[string]string{
+		localityKey:       localityValue,
+		backendServiceKey: cluster2Name,
+	}
+	if diff := cmp.Diff(cluster2Labels, wantCluster2Labels); diff != "" {
+		t.Fatalf("Telemetry labels after failover (-got +want): %v", diff)
+	}
+}
+
 type fakeStatsHandler struct {
 	labels map[string]string
 
@@ -146,3 +262,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
+}