Repository: grpc/grpc-go
File: xds/internal/balancer/cdsbalancer/cdsbalancer.go (Go)

Code before:
		// ExitIdle (but still checks for the interface's existence to
		// avoid a panic if not).  If the child does not, no subconns
		// will be connected.
		b.childLB.ExitIdle()
	})
}

// Node ID needs to be manually added to errors generated in the following
// scenarios:
//   - resource-does-not-exist: since the xDS watch API uses a separate callback
//     instead of returning an error value. TODO(gRFC A88): Once A88 is
//     implemented, the xDS client will be able to add the node ID to
//     resource-does-not-exist errors as well, and we can get rid of this
//     special handling.
//   - received a good update from the xDS client, but the update either contains
//     an invalid security configuration or contains invalid aggragate cluster
//     config.
func (b *cdsBalancer) annotateErrorWithNodeID(err error) error {
	nodeID := b.xdsClient.BootstrapConfig().Node().GetId()
	return fmt.Errorf("[xDS node id: %v]: %w", nodeID, err)
}

var onwatcherUpdated = func() {
	// This function is a no-op, but can be overridden in tests to signal that
	// the watchers map has been updated.
}

// Handles a good Cluster update from the xDS client. Kicks off the discovery
// mechanism generation process from the top-level cluster and if the cluster
// graph is resolved, generates child policy config and pushes it down.
//
// Only executed in the context of a serializer callback.
func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpdate) {
	state := b.watchers[name]
	if state == nil {
		// We are currently not watching this cluster anymore. Return early.
		return
	}

	b.logger.Infof("Received Cluster resource: %s", pretty.ToJSON(update))

	// Update the watchers map with the update for the cluster.
	state.lastUpdate = &update

	// For an aggregate cluster, always use the security configuration on the
	// root cluster.
	if name == b.lbCfg.ClusterName {
		// Process the security config from the received update before building the
		// child policy or forwarding the update to it. We do this because the child
		// policy may try to create a new subConn inline. Processing the security
		// configuration here and setting up the handshakeInfo will make sure that

Review comment (suggestion):
I think you can avoid adding a function specifically for tests and instead verify that the unsubscription happens by intercepting the requests in the xDS management server. When creating the xDS management server, you can provide an `OnStreamRequest` in `e2e.ManagementServerOptions`. Once the management server removes the cluster, you can wait for an update CDS discovery request from the xds client which should not have the removed cluster.

Code after:
		// ExitIdle (but still checks for the interface's existence to
		// avoid a panic if not).  If the child does not, no subconns
		// will be connected.
		b.childLB.ExitIdle()
	})
}

// Node ID needs to be manually added to errors generated in the following
// scenarios:
//   - resource-does-not-exist: since the xDS watch API uses a separate callback
//     instead of returning an error value. TODO(gRFC A88): Once A88 is
//     implemented, the xDS client will be able to add the node ID to
//     resource-does-not-exist errors as well, and we can get rid of this
//     special handling.
//   - received a good update from the xDS client, but the update either contains
//     an invalid security configuration or contains invalid aggragate cluster
//     config.
func (b *cdsBalancer) annotateErrorWithNodeID(err error) error {
	nodeID := b.xdsClient.BootstrapConfig().Node().GetId()
	return fmt.Errorf("[xDS node id: %v]: %w", nodeID, err)
}

// Handles a good Cluster update from the xDS client. Kicks off the discovery
// mechanism generation process from the top-level cluster and if the cluster
// graph is resolved, generates child policy config and pushes it down.
//
// Only executed in the context of a serializer callback.
func (b *cdsBalancer) onClusterUpdate(name string, update xdsresource.ClusterUpdate) {
	state := b.watchers[name]
	if state == nil {
		// We are currently not watching this cluster anymore. Return early.
		return
	}

	b.logger.Infof("Received Cluster resource: %s", pretty.ToJSON(update))

	// Update the watchers map with the update for the cluster.
	state.lastUpdate = &update

	// For an aggregate cluster, always use the security configuration on the
	// root cluster.
	if name == b.lbCfg.ClusterName {
		// Process the security config from the received update before building the
		// child policy or forwarding the update to it. We do this because the child
		// policy may try to create a new subConn inline. Processing the security
		// configuration here and setting up the handshakeInfo will make sure that
		// such attempts are handled properly.
		if err := b.handleSecurityConfig(update.SecurityCfg); err != nil {
			// If the security config is invalid, for example, if the provider
			// instance is not found in the bootstrap config, we need to put the
			// channel in transient failure.