Skip to content

xds/cdsbalancer: changed the setupManagementServer helper to take listener and OnStreamReq as parameter #8467

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eshitachandwani merged 12 commits into from
Aug 18, 2025
Merged

xds/cdsbalancer: changed the setupManagementServer helper to take listener and OnStreamReq as parameter #8467

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
f0909e5
Update cdsbalancer_test.go
eshitachandwani Jul 22, 2025
2410c79
changes
eshitachandwani Aug 5, 2025
916870d
changes
eshitachandwani Aug 5, 2025
2892f4b
Merge branch 'grpc:master' into eshitachandwani-patch-2
eshitachandwani Aug 5, 2025
6ab5c5e
change test
eshitachandwani Aug 13, 2025
ba6af81
remove ctx
eshitachandwani Aug 13, 2025
2974342
add comment
eshitachandwani Aug 13, 2025
e74f6d1
change from to 2 function
eshitachandwani Aug 13, 2025
2c3d63b
make only one test
eshitachandwani Aug 13, 2025
2b47178
add s
eshitachandwani Aug 13, 2025
715d076
remove defaults
eshitachandwani Aug 14, 2025
97b05ce
remove unwanted commment
eshitachandwani Aug 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 25 additions & 12 deletions xds/internal/balancer/cdsbalancer/aggregate_cluster_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,11 @@ import (
"google.golang.org/grpc/status"
"google.golang.org/grpc/xds/internal"
"google.golang.org/grpc/xds/internal/balancer/clusterresolver"
"google.golang.org/grpc/xds/internal/xdsclient/xdsresource/version"

v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
v3discoverypb "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
testgrpc "google.golang.org/grpc/interop/grpc_testing"
testpb "google.golang.org/grpc/interop/grpc_testing"
)
Expand Down Expand Up @@ -131,7 +133,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Push the first cluster resource through the management server and
// verify the configuration pushed to the child policy.
Expand Down Expand Up @@ -174,7 +176,7 @@ func (s) TestAggregateClusterSuccess_LeafNode(t *testing.T) {
// contains the expected discovery mechanisms.
func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Configure the management server with the aggregate cluster resource
// pointing to two child clusters, one EDS and one LogicalDNS. Include the
Expand Down Expand Up @@ -281,7 +283,7 @@ func (s) TestAggregateClusterSuccess_ThenUpdateChildClusters(t *testing.T) {
// policy contains a single discovery mechanism.
func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Configure the management server with the aggregate cluster resource
// pointing to two child clusters.
Expand Down Expand Up @@ -356,7 +358,7 @@ func (s) TestAggregateClusterSuccess_ThenChangeRootToEDS(t *testing.T) {
// discovery mechanisms.
func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Start off with the requested cluster being a leaf EDS cluster.
resources := e2e.UpdateOptions{
Expand Down Expand Up @@ -450,7 +452,7 @@ func (s) TestAggregatedClusterSuccess_SwitchBetweenLeafAndAggregate(t *testing.T
// longer exceed maximum depth, but be at the maximum allowed depth, and
// verifies that an RPC can be made successfully.
func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

resources := e2e.UpdateOptions{
NodeID: nodeID,
Expand Down Expand Up @@ -538,7 +540,7 @@ func (s) TestAggregatedClusterFailure_ExceedsMaxStackDepth(t *testing.T) {
// pushed only after all child clusters are resolved.
func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Configure the management server with an aggregate cluster resource having
// a diamond dependency pattern, (A->[B,C]; B->D; C->D). Includes resources
Expand Down Expand Up @@ -605,7 +607,7 @@ func (s) TestAggregatedClusterSuccess_DiamondDependency(t *testing.T) {
// pushed only after all child clusters are resolved.
func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

// Configure the management server with an aggregate cluster resource that
// has duplicates in the graph, (A->[B, C]; B->[C, D]). Include resources
Expand Down Expand Up @@ -683,7 +685,7 @@ func (s) TestAggregatedClusterSuccess_IgnoreDups(t *testing.T) {
// child policy and that an RPC can be successfully made.
func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

const (
clusterNameA = clusterName // cluster name in cds LB policy config
Expand Down Expand Up @@ -768,7 +770,7 @@ func (s) TestAggregatedCluster_NodeChildOfItself(t *testing.T) {
// that the aggregate cluster graph has no leaf clusters.
func (s) TestAggregatedCluster_CycleWithNoLeafNode(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

const (
clusterNameA = clusterName // cluster name in cds LB policy config
Expand Down Expand Up @@ -816,7 +818,7 @@ func (s) TestAggregatedCluster_CycleWithNoLeafNode(t *testing.T) {
// child policy and RPCs should get routed to that leaf cluster.
func (s) TestAggregatedCluster_CycleWithLeafNode(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

// Start a test service backend.
server := stubserver.StartTestService(t, nil)
Expand Down Expand Up @@ -872,10 +874,21 @@ func (s) TestAggregatedCluster_CycleWithLeafNode(t *testing.T) {
// removed from the tree no longer has a watcher and the new cluster added has a
// new watcher.
func (s) TestWatchers(t *testing.T) {
mgmtServer, nodeID, _, _, _, cdsResourceRequestedCh, _ := setupWithManagementServer(t)

ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
cdsResourceRequestedCh := make(chan []string, 1)
onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
if len(req.GetResourceNames()) > 0 {
select {
case cdsResourceRequestedCh <- req.GetResourceNames():
case <-ctx.Done():
}
}
}
return nil
}
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, onStreamReq)

const (
clusterA = clusterName
Expand Down
124 changes: 70 additions & 54 deletions xds/internal/balancer/cdsbalancer/cdsbalancer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,46 +188,19 @@ func registerWrappedCDSPolicy(t *testing.T) chan balancer.Balancer {
// - Creates a manual resolver that configures the cds LB policy as the
// top-level policy, and pushes an initial configuration to it
// - Creates a gRPC channel with the above manual resolver
// - Executes OnStreamRequest callback provided by the caller

@easwars easwars Aug 13, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't make sense. This should be part of the first bullet point. Something like

  • Spins up an xDS management server and passes it the onStreamRequest callback that will get invoked every time a request is received on the ADS stream.

@eshitachandwani eshitachandwani Aug 14, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

@easwars easwars Aug 14, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don't want the bullet point "// - Executes OnStreamRequest callback provided by the caller", because we don't directly call it. We only pass it to function that creates the management server.

//
// Returns the following:
// - the xDS management server
// - the nodeID expected by the management server
// - the grpc channel to the test backend service
// - the manual resolver configured on the channel
// - the xDS client used the grpc channel
// - a channel on which requested cluster resource names are sent
// - a channel used to signal that previously requested cluster resources are
// no longer requested
func setupWithManagementServer(t *testing.T) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient, chan []string, chan struct{}) {
return setupWithManagementServerAndListener(t, nil)
}

// Same as setupWithManagementServer, but also allows the caller to specify
// a listener to be used by the management server.
func setupWithManagementServerAndListener(t *testing.T, lis net.Listener) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient, chan []string, chan struct{}) {
func setupWithManagementServer(t *testing.T, lis net.Listener, onStreamRequest func(int64, *v3discoverypb.DiscoveryRequest) error) (*e2e.ManagementServer, string, *grpc.ClientConn, *manual.Resolver, xdsclient.XDSClient) {
t.Helper()

cdsResourceRequestedCh := make(chan []string, 1)
cdsResourceCanceledCh := make(chan struct{}, 1)
mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
Listener: lis,
OnStreamRequest: func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
switch len(req.GetResourceNames()) {
case 0:
select {
case cdsResourceCanceledCh <- struct{}{}:
default:
}
default:
select {
case cdsResourceRequestedCh <- req.GetResourceNames():
default:
}
}
}
return nil
},
Listener: lis,
OnStreamRequest: onStreamRequest,
// Required for aggregate clusters as all resources cannot be requested
// at once.
AllowResourceSubset: true,
Expand Down Expand Up @@ -268,7 +241,7 @@ func setupWithManagementServerAndListener(t *testing.T, lis net.Listener) (*e2e.
cc.Connect()
t.Cleanup(func() { cc.Close() })

return mgmtServer, nodeID, cc, r, xdsC, cdsResourceRequestedCh, cdsResourceCanceledCh
return mgmtServer, nodeID, cc, r, xdsC
}

// Helper function to compare the load balancing configuration received on the
Expand Down Expand Up @@ -321,11 +294,23 @@ func verifyRPCError(gotErr error, wantCode codes.Code, wantErr, wantNodeID strin
// configuration changes, it stops requesting the old cluster resource and
// starts requesting the new one.
func (s) TestConfigurationUpdate_Success(t *testing.T) {
_, _, _, r, xdsClient, cdsResourceRequestedCh, _ := setupWithManagementServer(t)

// Verify that the specified cluster resource is requested.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
cdsResourceRequestedCh := make(chan []string, 1)
onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
if len(req.GetResourceNames()) > 0 {
select {
case cdsResourceRequestedCh <- req.GetResourceNames():
case <-ctx.Done():
}
}
}
return nil
}
_, _, _, r, xdsClient := setupWithManagementServer(t, nil, onStreamReq)

// Verify that the specified cluster resource is requested.
wantNames := []string{clusterName}
if err := waitForResourceNames(ctx, cdsResourceRequestedCh, wantNames); err != nil {
t.Fatal(err)
Expand Down Expand Up @@ -616,7 +601,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
Expand All @@ -640,7 +625,7 @@ func (s) TestClusterUpdate_Success(t *testing.T) {
// balancing configuration pushed to the child is as expected.
func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
lbCfgCh, _, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, _, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, _, _, _ := setupWithManagementServer(t, nil, nil)

clusterResource := e2e.ClusterResourceWithOptions(e2e.ClusterOptions{
ClusterName: clusterName,
Expand Down Expand Up @@ -689,15 +674,21 @@ func (s) TestClusterUpdate_SuccessWithLRS(t *testing.T) {
// continue using the previous good update.
func (s) TestClusterUpdate_Failure(t *testing.T) {
_, resolverErrCh, _, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServer(t)

// Verify that the specified cluster resource is requested.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
wantNames := []string{clusterName}
if err := waitForResourceNames(ctx, cdsResourceRequestedCh, wantNames); err != nil {
t.Fatal(err)
cdsResourceCanceledCh := make(chan struct{}, 1)
onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
if len(req.GetResourceNames()) == 0 {
select {
case cdsResourceCanceledCh <- struct{}{}:
default:
}
}
}
return nil
}
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, onStreamReq)

// Configure the management server to return a cluster resource that
// contains a config_source_specifier for the `lrs_server` field which is not
Expand Down Expand Up @@ -806,12 +797,31 @@ func (s) TestClusterUpdate_Failure(t *testing.T) {
func (s) TestResolverError(t *testing.T) {
_, resolverErrCh, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
lis := testutils.NewListenerWrapper(t, nil)
mgmtServer, nodeID, cc, r, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServerAndListener(t, lis)
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
cdsResourceCanceledCh := make(chan struct{}, 1)
cdsResourceRequestedCh := make(chan []string, 1)
onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
switch len(req.GetResourceNames()) {
case 0:
select {
case cdsResourceCanceledCh <- struct{}{}:
default:

@easwars easwars Aug 13, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we still need this default case here?

@eshitachandwani eshitachandwani Aug 14, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

changed.

}
default:
select {
case cdsResourceRequestedCh <- req.GetResourceNames():
case <-ctx.Done():
}
}
}
return nil
}
mgmtServer, nodeID, cc, r, _ := setupWithManagementServer(t, lis, onStreamReq)

// Grab the wrapped connection from the listener wrapper. This will be used
// to verify the connection is closed.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
val, err := lis.NewConnCh.Receive(ctx)
if err != nil {
t.Fatalf("Failed to receive new connection from wrapped listener: %v", err)
Expand Down Expand Up @@ -949,15 +959,21 @@ func (s) TestResolverError(t *testing.T) {
// - when the cluster resource is re-sent by the management server, RPCs
// should start succeeding.
func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
mgmtServer, nodeID, cc, _, _, cdsResourceRequestedCh, cdsResourceCanceledCh := setupWithManagementServer(t)

// Verify that the specified cluster resource is requested.
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
wantNames := []string{clusterName}
if err := waitForResourceNames(ctx, cdsResourceRequestedCh, wantNames); err != nil {
t.Fatal(err)
cdsResourceCanceledCh := make(chan struct{}, 1)
onStreamReq := func(_ int64, req *v3discoverypb.DiscoveryRequest) error {
if req.GetTypeUrl() == version.V3ClusterURL {
if len(req.GetResourceNames()) == 0 {
select {
case cdsResourceCanceledCh <- struct{}{}:
default:

@easwars easwars Aug 13, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And here?

@eshitachandwani eshitachandwani Aug 14, 2025

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

changed.

}
}
}
return nil
}
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, onStreamReq)

// Start a test service backend.
server := stubserver.StartTestService(t, nil)
Expand Down Expand Up @@ -1028,7 +1044,7 @@ func (s) TestClusterUpdate_ResourceNotFound(t *testing.T) {
func (s) TestClose(t *testing.T) {
cdsBalancerCh := registerWrappedCDSPolicy(t)
_, _, _, childPolicyCloseCh := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

// Start a test service backend.
server := stubserver.StartTestService(t, nil)
Expand Down Expand Up @@ -1075,7 +1091,7 @@ func (s) TestClose(t *testing.T) {
func (s) TestExitIdle(t *testing.T) {
cdsBalancerCh := registerWrappedCDSPolicy(t)
_, _, exitIdleCh, _ := registerWrappedClusterResolverPolicy(t)
mgmtServer, nodeID, cc, _, _, _, _ := setupWithManagementServer(t)
mgmtServer, nodeID, cc, _, _ := setupWithManagementServer(t, nil, nil)

// Start a test service backend.
server := stubserver.StartTestService(t, nil)
Expand Down