diff --git a/internal/xds/resolver/cluster_specifier_plugin_test.go b/internal/xds/resolver/cluster_specifier_plugin_test.go index df3ff2e746c0..503c71fd80f7 100644 --- a/internal/xds/resolver/cluster_specifier_plugin_test.go +++ b/internal/xds/resolver/cluster_specifier_plugin_test.go @@ -22,6 +22,7 @@ import ( "context" "encoding/json" "fmt" + "sync" "testing" "github.com/google/uuid" @@ -342,6 +343,103 @@ func (s) TestXDSResolverDelayedOnCommittedCSP(t *testing.T) { verifyUpdateFromResolver(ctx, t, stateCh, wantSC) } +// TestResolverClusterSpecifierPluginRefCountRace verifies that a cluster +// specifier plugin is handled correctly when its last in-flight RPC is +// committed at the same time as an xDS update that names it again. Whichever +// happens first, the plugin must end up present in the service config: if the +// commit lands first the entry is torn down and a fresh one replaces it, and if +// the update lands first the existing entry is simply reused. +func (s) TestResolverClusterSpecifierPluginRefCountRace(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + nodeID := uuid.New().String() + mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID) + + routeConfigForPlugin := func(name, value string) []*v3routepb.RouteConfiguration { + return []*v3routepb.RouteConfiguration{e2e.RouteConfigResourceWithOptions(e2e.RouteConfigOptions{ + RouteConfigName: defaultTestRouteConfigName, + ListenerName: defaultTestServiceName, + ClusterSpecifierType: e2e.RouteConfigClusterSpecifierTypeClusterSpecifierPlugin, + ClusterSpecifierPluginName: name, + ClusterSpecifierPluginConfig: testutils.MarshalAny(t, &wrapperspb.StringValue{Value: value}), + })} + } + wantConfigForPlugin := func(name, value string) string { + return fmt.Sprintf(`{ + "loadBalancingConfig": [{ + "xds_cluster_manager_experimental": { + "children": { + "cluster_specifier_plugin:%s": { + "childPolicy": [{"csp_experimental": {"arbitrary_field": "%s"}}] + } + } + } + }] + }`, name, value) + } + + listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)} + configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspA", "anythingA"), nil, nil) + + stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc) + cs := verifyUpdateFromResolver(ctx, t, stateCh, wantConfigForPlugin("cspA", "anythingA")) + + // Start an RPC on cspA and leave it uncommitted, so cspA stays referenced. + res, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"}) + if err != nil { + t.Fatalf("cs.SelectConfig(): %v", err) + } + if got, want := clustermanager.PickedCluster(res.Context), "cluster_specifier_plugin:cspA"; got != want { + t.Fatalf("Config selector returned cluster %q, want %q", got, want) + } + + // Move the route to cspB. cspA is now held only by the in-flight RPC, so + // both plugins stay in the service config. Waiting for that config matters: + // it confirms the client processed this update before the race below, and + // without it the next update could coalesce with this one and leave the + // route config unchanged from the client's point of view. + configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspB", "anythingB"), nil, nil) + verifyUpdateFromResolver(ctx, t, stateCh, `{ + "loadBalancingConfig": [{ + "xds_cluster_manager_experimental": { + "children": { + "cluster_specifier_plugin:cspA": { + "childPolicy": [{"csp_experimental": {"arbitrary_field": "anythingA"}}] + }, + "cluster_specifier_plugin:cspB": { + "childPolicy": [{"csp_experimental": {"arbitrary_field": "anythingB"}}] + } + } + } + }] + }`) + + // Commit the RPC, dropping cspA's last reference, while an update naming + // cspA again is pushed concurrently. The two orderings exercise different + // paths through the refcounted entry, and both must converge on cspA being + // in the service config. + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + res.OnCommitted() + }() + configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspA", "anythingA"), nil, nil) + wg.Wait() + + cs = waitForServiceConfig(ctx, t, stateCh, wantConfigForPlugin("cspA", "anythingA")) + + // The surviving entry must still be usable for new RPCs. + res, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"}) + if err != nil { + t.Fatalf("cs.SelectConfig() after the race: %v", err) + } + if got, want := clustermanager.PickedCluster(res.Context), "cluster_specifier_plugin:cspA"; got != want { + t.Fatalf("Config selector returned cluster %q, want %q", got, want) + } + res.OnCommitted() +} + // TestResolverClusterSpecifierPlugin_WithFilters tests the case where a route // configuration containing cluster specifier plugins is sent by the management // server, and HTTP filters are configured. The test verifies that the diff --git a/internal/xds/resolver/helpers_test.go b/internal/xds/resolver/helpers_test.go index f168b663ca5c..d8c0ac47ddfa 100644 --- a/internal/xds/resolver/helpers_test.go +++ b/internal/xds/resolver/helpers_test.go @@ -173,6 +173,36 @@ func verifyUpdateFromResolver(ctx context.Context, t *testing.T, stateCh chan re return cs } +// waitForServiceConfig drains updates from the resolver until one carries a +// service config matching wantSC, and fails if none does before ctx expires. +// Use this instead of verifyUpdateFromResolver when the resolver is expected to +// publish intermediate configs on the way to the wanted one. +// +// Returns the config selector from the matching update. +func waitForServiceConfig(ctx context.Context, t *testing.T, stateCh chan resolver.State, wantSC string) iresolver.ConfigSelector { + t.Helper() + + want := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(wantSC) + for { + select { + case <-ctx.Done(): + t.Fatalf("Timeout waiting for the resolver to publish service config:\n%s", wantSC) + case state := <-stateCh: + if err := state.ServiceConfig.Err; err != nil { + t.Fatalf("Received error in service config: %v", err) + } + if !internal.EqualServiceConfigForTesting(state.ServiceConfig.Config, want.Config) { + continue + } + cs := iresolver.GetConfigSelector(state) + if cs == nil { + t.Fatal("Received nil config selector in update from resolver") + } + return cs + } + } +} + // verifyNoUpdateFromResolver verifies that no update is pushed on stateCh. // Calls t.Fatal() if an update is received before defaultTestShortTimeout // expires. diff --git a/internal/xds/resolver/serviceconfig.go b/internal/xds/resolver/serviceconfig.go index bdbaea699956..321696faf43a 100644 --- a/internal/xds/resolver/serviceconfig.go +++ b/internal/xds/resolver/serviceconfig.go @@ -75,14 +75,14 @@ type xdsClusterManagerConfig struct { // serviceConfigJSON produces a service config in JSON format that contains LB // policy config for the "xds_cluster_manager" LB policy, with entries in the // children map for all active clusters. -func serviceConfigJSON(activeClusters map[string]*clusterInfo, activePlugins map[string]*clusterInfo) []byte { +func serviceConfigJSON(activeClusters, activePlugins map[string]*grpcsync.RefCounted[*clusterInfo]) []byte { // Generate children (all entries in activeClusters). children := make(map[string]xdsChildConfig) for cluster, ci := range activeClusters { - children[cluster] = ci.cfg + children[cluster] = ci.Value().cfg } for plugin, ci := range activePlugins { - children[plugin] = ci.cfg + children[plugin] = ci.Value().cfg } sc := serviceConfig{ @@ -109,6 +109,10 @@ type virtualHost struct { type routeCluster struct { name string // Name of the cluster. interceptor httpfilter.ClientInterceptor // HTTP filters to run for RPCs matching this route. + // info is the resolver-wide entry for this cluster, shared by every route + // that references it. An RPC routed here holds a reference on it until the + // RPC is committed. + info *grpcsync.RefCounted[*clusterInfo] } type route struct { @@ -159,8 +163,8 @@ type configSelector struct { // Configuration received from the xDS management server. virtualHost virtualHost routes []route - clusters map[string]*clusterInfo - plugins map[string]*clusterInfo + clusters map[string]*grpcsync.RefCounted[*clusterInfo] + plugins map[string]*grpcsync.RefCounted[*clusterInfo] httpFilterConfig []xdsresource.HTTPFilter xdsConfig *xdsresource.XDSConfig } @@ -223,43 +227,16 @@ func (cs *configSelector) SelectConfig(rpcInfo iresolver.RPCInfo) (*iresolver.RP // Add a ref to the selected cluster to keep the interceptors alive until RPC // is committed. rc.Increment() - if info, ok := cs.clusters[cluster.name]; ok { - // Add a ref to the selected cluster, as this RPC needs this - // cluster until it is committed. - info.refCount.Add(1) - config.OnCommitted = sync.OnceFunc(func() { - if v := info.refCount.Add(-1); v == 0 { - // We call unsubscribe rather than sendNewServiceConfig to - // prevent redundant updates. If the reference count in the - // dependency manager drops to zero, it will automatically - // trigger a service config update with this cluster - // removed. Calling unsubscribe allows the dependency - // manager to handle the update flow once and for all. - info.unsubscribe() - } - // Decrement the refcount of the route cluster and close the interceptor - // if refcount goes to zero. - rc.Decrement() - }) - } else if info, ok := cs.plugins[cluster.name]; ok { - // Add a ref to the selected plugin, as this RPC needs this - // plugin until it is committed. - info.refCount.Add(1) - config.OnCommitted = sync.OnceFunc(func() { - if v := info.refCount.Add(-1); v == 0 { - // This entry will be removed from activePlugins when - // producing a new service config update. - cs.sendNewServiceConfig() - } - // Decrement the refcount of the route cluster and close the interceptor - // if refcount goes to zero. - rc.Decrement() - }) - } else { - // This should be unreachable because all route clusters are normalized - // into cs.clusters or cs.plugins during config selector creation. - panic(fmt.Sprintf("matched cluster %q not found in ConfigSelector", cluster.name)) - } + // Add a ref to the selected cluster or plugin, as this RPC needs it until it + // is committed. Releasing the last reference unsubscribes from the cluster + // or pushes a new service config for a plugin. + cluster.info.Increment() + config.OnCommitted = sync.OnceFunc(func() { + cluster.info.Decrement() + // Decrement the refcount of the route cluster and close the interceptor + // if refcount goes to zero. + rc.Decrement() + }) if rt.maxStreamDuration != 0 { config.MethodConfig.Timeout = &rt.maxStreamDuration @@ -364,18 +341,14 @@ func (cs *configSelector) stop() { } } - // If any reference counts drop to zero, a service config update is required - // to remove the clusters. Since the old config selector is stopped - // after a new one is active, we must trigger a subsequent update to delete - // the now-unused clusters. + // Release this config selector's reference on each cluster and plugin. If + // any reference count drops to zero, the cleanup registered when the entry + // was created removes it from the resolver's active maps and triggers the + // service config update needed to drop it from the channel's config. for _, ci := range cs.clusters { - if v := ci.refCount.Add(-1); v == 0 { - ci.unsubscribe() - } + ci.Decrement() } for _, ci := range cs.plugins { - if v := ci.refCount.Add(-1); v == 0 { - cs.sendNewServiceConfig() - } + ci.Decrement() } } diff --git a/internal/xds/resolver/serviceconfig_test.go b/internal/xds/resolver/serviceconfig_test.go index 4fa25d630723..8032184dfced 100644 --- a/internal/xds/resolver/serviceconfig_test.go +++ b/internal/xds/resolver/serviceconfig_test.go @@ -25,7 +25,6 @@ import ( "time" xxhash "github.com/cespare/xxhash/v2" - "github.com/google/go-cmp/cmp" "google.golang.org/grpc/internal/grpctest" "google.golang.org/grpc/internal/grpcutil" iresolver "google.golang.org/grpc/internal/resolver" @@ -44,52 +43,6 @@ func Test(t *testing.T) { grpctest.RunSubTests(t, s{}) } -func (s) TestPruneActiveClusters(t *testing.T) { - newClusterInfo := func(ref int32, unsubscribe func()) *clusterInfo { - ci := &clusterInfo{unsubscribe: unsubscribe} - ci.refCount.Store(ref) - return ci - } - r := &xdsResolver{ - activeClusters: map[string]*clusterInfo{ - "zero": newClusterInfo(0, func() {}), - "one": newClusterInfo(1, func() {}), - "two": newClusterInfo(2, func() {}), - "anotherzero": newClusterInfo(0, func() {}), - }, - activePlugins: map[string]*clusterInfo{ - "zero": newClusterInfo(0, nil), - "one": newClusterInfo(1, nil), - "two": newClusterInfo(2, nil), - "anotherzero": newClusterInfo(0, nil), - }, - } - wantActiveClusters := map[string]int32{ - "one": 1, - "two": 2, - } - wantActivePlugins := map[string]int32{ - "one": 1, - "two": 2, - } - r.pruneActiveClustersAndPlugins() - - getRefCounts := func(m map[string]*clusterInfo) map[string]int32 { - res := make(map[string]int32) - for k, v := range m { - res[k] = v.refCount.Load() - } - return res - } - - if d := cmp.Diff(getRefCounts(r.activeClusters), wantActiveClusters); d != "" { - t.Fatalf("r.activeClusters refCounts mismatch (-got +want):\n%s", d) - } - if d := cmp.Diff(getRefCounts(r.activePlugins), wantActivePlugins); d != "" { - t.Fatalf("r.activePlugins refCounts mismatch (-got +want):\n%s", d) - } -} - func (s) TestGenerateRequestHash(t *testing.T) { const channelID = 12378921 cs := &configSelector{channelID: channelID} diff --git a/internal/xds/resolver/xds_resolver.go b/internal/xds/resolver/xds_resolver.go index 98f06fce6f98..3e2e6843c8c3 100644 --- a/internal/xds/resolver/xds_resolver.go +++ b/internal/xds/resolver/xds_resolver.go @@ -25,7 +25,6 @@ import ( rand "math/rand/v2" "slices" "strings" - "sync/atomic" "google.golang.org/grpc" estats "google.golang.org/grpc/experimental/stats" @@ -137,8 +136,8 @@ func (b *xdsResolverBuilder) Build(target resolver.Target, cc resolver.ClientCon cc: cc, xdsClient: client, xdsClientClose: xdsClientClose, - activeClusters: make(map[string]*clusterInfo), - activePlugins: make(map[string]*clusterInfo), + activeClusters: make(map[string]*grpcsync.RefCounted[*clusterInfo]), + activePlugins: make(map[string]*grpcsync.RefCounted[*clusterInfo]), httpFilters: make(map[clientFilterKey]httpfilter.ClientFilter), channelID: rand.Uint64(), ldsResourceName: ldsResourceName, @@ -249,21 +248,21 @@ type xdsResolver struct { // The following fields are accessed only from within the serializer // callbacks. xdsConfig *xdsresource.XDSConfig - // activeClusters is a map from cluster name to information about the - // weighted cluster that includes a reference count and load balancing - // configuration. These counts are used only by the resolver. The current - // configSelector holds one reference, and each ongoing RPC holds an - // additional reference. When the count hits zero, the resolver removes the - // cluster from this map and calls unsubscribe. This signals the dependency - // manager to stop the xDS watch once its own reference count reaches zero. - activeClusters map[string]*clusterInfo - // activePlugins is a map from cluster specifier plugin name to information - // about the cluster specifier plugin that includes a ref count and load + // activeClusters is a map from cluster name to refcounted information about + // the weighted cluster that includes its load balancing configuration. + // These counts are used only by the resolver. The current configSelector + // holds one reference, and each ongoing RPC holds an additional reference. + // When the count hits zero, the resolver calls unsubscribe and removes the + // cluster from this map. The unsubscribe signals the dependency manager to + // stop the xDS watch once its own reference count reaches zero. + activeClusters map[string]*grpcsync.RefCounted[*clusterInfo] + // activePlugins is a map from cluster specifier plugin name to refcounted + // information about the cluster specifier plugin that includes its load // balancing configuration. These counts are used only by the resolver. The // current configSelector holds one reference, and each ongoing RPC holds an // additional reference. When the count hits zero, the resolver removes the - // plugin name from this map. - activePlugins map[string]*clusterInfo + // plugin name from this map and pushes a new service config. + activePlugins map[string]*grpcsync.RefCounted[*clusterInfo] curConfigSelector stoppableConfigSelector // httpFilters is a map from client filter key to client filter instance. It // lives here so that the resolver can reuse filter instances across config @@ -344,11 +343,6 @@ func (r *xdsResolver) Error(err error) { // // Only executed in the context of a serializer callback. func (r *xdsResolver) sendNewServiceConfig(cs stoppableConfigSelector) { - // Delete entries from r.activeClusters with zero references; - // otherwise serviceConfigJSON will generate a config including - // them. - r.pruneActiveClustersAndPlugins() - if errCS, ok := cs.(*erroringConfigSelector); ok { // Send an empty config, which picks pick-first, with no address, and // puts the ClientConn into transient failure. @@ -396,8 +390,8 @@ func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) { retryConfig: r.xdsConfig.VirtualHost.RetryConfig, }, routes: make([]route, len(r.xdsConfig.VirtualHost.Routes)), - clusters: make(map[string]*clusterInfo), - plugins: make(map[string]*clusterInfo), + clusters: make(map[string]*grpcsync.RefCounted[*clusterInfo]), + plugins: make(map[string]*grpcsync.RefCounted[*clusterInfo]), httpFilterConfig: r.xdsConfig.Listener.APIListener.HTTPFilters, xdsConfig: r.xdsConfig, } @@ -429,16 +423,25 @@ func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) { } return nil, err } + // Take one reference per distinct cluster, not per route that names + // it: cs.stop() releases references by ranging over cs.plugins, so + // it decrements once per entry no matter how many routes point + // here. The reference is also released by the deferred cs.stop() + // above if a later route fails to build. + ci, ok := cs.plugins[clusterName] + if !ok { + ci = r.acquireActiveClusterInfo(clusterName, "") + cs.plugins[clusterName] = ci + } + ci.Value().cfg = xdsChildConfig{ChildPolicy: balancerConfig(r.xdsConfig.RouteConfig.ClusterSpecifierPlugins[rt.ClusterSpecifierPlugin])} routeCluster := &routeCluster{ name: clusterName, interceptor: interceptor, + info: ci, } rc := grpcsync.NewRefCounted(routeCluster, func() { interceptor.Close() }) cs.routes[i].routeClusters = append(cs.routes[i].routeClusters, rc) clusters.Add(rc, 1) - ci := r.addOrGetActiveClusterInfo(clusterName, "") - ci.cfg = xdsChildConfig{ChildPolicy: balancerConfig(r.xdsConfig.RouteConfig.ClusterSpecifierPlugins[rt.ClusterSpecifierPlugin])} - cs.plugins[clusterName] = ci } else { for _, wc := range rt.WeightedClusters { clusterName := clusterPrefix + wc.Name @@ -453,16 +456,25 @@ func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) { } return nil, err } + // Take one reference per distinct cluster, not per route that + // names it: cs.stop() releases references by ranging over + // cs.clusters, so it decrements once per entry no matter how + // many routes point here. The reference is also released by the + // deferred cs.stop() above if a later route fails to build. + ci, ok := cs.clusters[clusterName] + if !ok { + ci = r.acquireActiveClusterInfo(clusterName, wc.Name) + cs.clusters[clusterName] = ci + } + ci.Value().cfg = xdsChildConfig{ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: wc.Name})} routeCluster := &routeCluster{ name: clusterName, interceptor: interceptor, + info: ci, } rc := grpcsync.NewRefCounted(routeCluster, func() { interceptor.Close() }) cs.routes[i].routeClusters = append(cs.routes[i].routeClusters, rc) clusters.Add(rc, int64(wc.Weight)) - ci := r.addOrGetActiveClusterInfo(clusterName, wc.Name) - ci.cfg = xdsChildConfig{ChildPolicy: newBalancerConfig(cdsName, cdsBalancerConfig{Cluster: wc.Name})} - cs.clusters[clusterName] = ci } } cs.routes[i].clusters = clusters @@ -479,16 +491,6 @@ func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) { cs.routes[i].autoHostRewrite = rt.AutoHostRewrite } - // Account for this config selector's clusters. Do this after no further - // errors may occur. Note: cs.clusters are pointers to entries in - // activeClusters. - for _, ci := range cs.clusters { - ci.refCount.Add(1) - } - for _, ci := range cs.plugins { - ci.refCount.Add(1) - } - // Cleanup filter instances that are no longer specified in the current // listener resource. filtersInNewConfig := make(map[clientFilterKey]bool) @@ -506,66 +508,97 @@ func (r *xdsResolver) newConfigSelector() (_ *configSelector, err error) { return cs, nil } -// pruneActiveClustersAndPlugins removes entries from activeClusters and -// activePlugins that have a reference count of zero. For clusters, it also -// invokes the unsubscribe function to signal the dependency manager to stop the -// xDS watch. Because cluster specifier plugins do not have their own watches, -// they are simply removed from the map without an unsubscribe call. -// -// Only executed in the context of a serializer callback. -func (r *xdsResolver) pruneActiveClustersAndPlugins() { - for cluster, ci := range r.activeClusters { - if ci.refCount.Load() == 0 { - ci.unsubscribe() - delete(r.activeClusters, cluster) - } - } - for cluster, ci := range r.activePlugins { - if ci.refCount.Load() == 0 { - delete(r.activePlugins, cluster) - } - } -} - -// addOrGetActiveClusterInfo returns the clusterInfo for the provided key, -// creating it if it does not exist. It accepts the following parameters: +// acquireActiveClusterInfo returns the refcounted clusterInfo for the provided +// key with one reference held on behalf of the caller, creating the entry if it +// does not exist. It accepts the following parameters: // - key: Formatted as "cluster:" or "cluster_specifier_plugin:", // this is the lookup key for the activeClusters or activePlugins maps. // - name: The actual xDS resource name used to initiate a CDS watch. // If empty (e.g., for plugins), no resource watch is triggered. // -// This function manages entry creation and xDS subscriptions but does not -// increment the reference count of the returned clusterInfo. -func (r *xdsResolver) addOrGetActiveClusterInfo(key string, name string) *clusterInfo { +// An entry whose reference count has already dropped to zero cannot be revived, +// so a fresh entry (with a fresh subscription) replaces it in that case. +// +// Only executed in the context of a serializer callback. +func (r *xdsResolver) acquireActiveClusterInfo(key string, name string) *grpcsync.RefCounted[*clusterInfo] { if name == "" { - ci, ok := r.activePlugins[key] - if !ok { - ci = &clusterInfo{} - r.activePlugins[key] = ci + if ci, ok := r.activePlugins[key]; ok && ci.TryIncrement() { + return ci } + var ci *grpcsync.RefCounted[*clusterInfo] + ci = grpcsync.NewRefCounted(&clusterInfo{}, func() { + r.scheduleActiveEntryRemoval(r.activePlugins, key, ci, func() { + // Plugins have no xDS watch of their own, so nothing else tells + // the channel to drop this child. Push a service config that no + // longer mentions it. + r.sendNewServiceConfig(r.curConfigSelector) + }) + }) + r.activePlugins[key] = ci return ci } - ci, ok := r.activeClusters[key] - if !ok { - ci = &clusterInfo{unsubscribe: r.dm.SubscribeToCluster(name)} - r.activeClusters[key] = ci + + if ci, ok := r.activeClusters[key]; ok && ci.TryIncrement() { + return ci } + unsubscribe := r.dm.SubscribeToCluster(name) + var ci *grpcsync.RefCounted[*clusterInfo] + ci = grpcsync.NewRefCounted(&clusterInfo{}, func() { + // Queue the removal before unsubscribing. Unsubscribing can make the + // dependency manager push an update, which schedules a callback on this + // same serializer; queueing first guarantees this entry is gone from + // activeClusters before that update regenerates the service config, so + // the dropped cluster does not reappear in it. + // + // Nothing needs to run after the removal: the update the dependency + // manager pushes is what drops this cluster from the channel's config, + // and sending our own service config here would only duplicate it. + r.scheduleActiveEntryRemoval(r.activeClusters, key, ci, nil) + // Unsubscribing decrements the reference count in the dependency + // manager; once that count reaches zero, the underlying CDS watch is + // terminated. + // + // This runs synchronously rather than inside the scheduled removal: that + // removal returns early if a newer entry has taken this key, and the + // newer entry holds its own subscription, so deferring the unsubscribe + // would leak this one. + unsubscribe() + }) + r.activeClusters[key] = ci return ci } +// scheduleActiveEntryRemoval schedules the removal of key from m, which must be +// one of the resolver's active entry maps, once ci has released its last +// reference. onRemoved, if non-nil, runs after the entry has been removed. +// +// The removal is scheduled rather than performed inline because the last +// reference is usually released by an RPC completing on an arbitrary goroutine, +// while the active maps may only be touched from within a serializer callback. +// By the time the scheduled callback runs, a newer entry may already have +// replaced this one under the same key, so the entry is removed only if it is +// still the current one. +// +// This must remain a TrySchedule and not a ScheduleAndWait. The last reference +// is also dropped by configSelector.stop(), which itself runs inside a +// serializer callback, so blocking here until the callback ran would deadlock +// the serializer against itself. +func (r *xdsResolver) scheduleActiveEntryRemoval(m map[string]*grpcsync.RefCounted[*clusterInfo], key string, ci *grpcsync.RefCounted[*clusterInfo], onRemoved func()) { + r.serializer.TrySchedule(func(context.Context) { + if m[key] != ci { + return + } + delete(m, key) + if onRemoved != nil { + onRemoved() + } + }) +} + type clusterInfo struct { - // refCount is the number of references to this cluster. - refCount atomic.Int32 // cfg is the child configuration for this cluster, containing either the // csp config or the cds cluster config. cfg xdsChildConfig - // unsubscribe is the function to call to unsubscribe from this cluster's - // CDS resource. It is populated only for clusters in activeClusters and not - // for cluster specifier plugins. When invoked, it decrements the reference - // count in the dependency manager; once that count reaches zero, the - // underlying CDS watch is terminated. Plugins do not have associated - // watches and therefore do not require an unsubscribe function. - unsubscribe func() } // Contains common functionality to be executed when resources of either type diff --git a/internal/xds/resolver/xds_resolver_test.go b/internal/xds/resolver/xds_resolver_test.go index b8ed304b885b..574b602dfeb4 100644 --- a/internal/xds/resolver/xds_resolver_test.go +++ b/internal/xds/resolver/xds_resolver_test.go @@ -1382,6 +1382,77 @@ func (s) TestResolverKeepWatchOpen_ActiveRPCs(t *testing.T) { res.OnCommitted() } +// TestResolverClusterSharedByMultipleRoutes verifies the reference accounting +// for a cluster that more than one route points at. A config selector takes a +// single reference per distinct cluster, however many routes name it, and +// releases exactly that one reference when it is stopped. +func (s) TestResolverClusterSharedByMultipleRoutes(t *testing.T) { + clusterA := "cluster-A" + clusterB := "cluster-B" + + ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) + defer cancel() + mgmtServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true}) + + nodeID := uuid.New().String() + bc := e2e.DefaultBootstrapContents(t, nodeID, mgmtServer.Address) + + // A route configuration whose two distinct routes both send traffic to the + // same cluster, so the config selector encounters that cluster twice. + routeToCluster := func(prefix, cluster string) *v3routepb.Route { + return &v3routepb.Route{ + Match: &v3routepb.RouteMatch{PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: prefix}}, + Action: &v3routepb.Route_Route{Route: &v3routepb.RouteAction{ + ClusterSpecifier: &v3routepb.RouteAction_WeightedClusters{WeightedClusters: &v3routepb.WeightedCluster{ + Clusters: []*v3routepb.WeightedCluster_ClusterWeight{{ + Name: cluster, + Weight: &wrapperspb.UInt32Value{Value: 100}, + }}, + }}, + }}, + } + } + routeConfigForCluster := func(cluster string) *v3routepb.RouteConfiguration { + return &v3routepb.RouteConfiguration{ + Name: defaultTestRouteConfigName, + VirtualHosts: []*v3routepb.VirtualHost{{ + Domains: []string{defaultTestServiceName}, + Routes: []*v3routepb.Route{ + routeToCluster("/service/first", cluster), + routeToCluster("/", cluster), + }, + }}, + } + } + + listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)} + clusters := []*v3clusterpb.Cluster{ + e2e.DefaultCluster(clusterA, "endpoint-A", e2e.SecurityLevelNone), + e2e.DefaultCluster(clusterB, "endpoint-B", e2e.SecurityLevelNone), + } + endpoints := []*v3endpointpb.ClusterLoadAssignment{ + e2e.DefaultEndpoint("endpoint-A", "localhost", []uint32{8080}), + e2e.DefaultEndpoint("endpoint-B", "localhost", []uint32{8081}), + } + configureResources(ctx, t, mgmtServer, nodeID, listeners, []*v3routepb.RouteConfiguration{routeConfigForCluster(clusterA)}, clusters, endpoints) + + stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc) + + // Both routes name cluster-A, so it appears once in the service config. + verifyUpdateFromResolver(ctx, t, stateCh, wantServiceConfig(clusterA)) + + // Move both routes to cluster-B. With no RPCs in flight, stopping the + // outgoing config selector releases cluster-A's only reference. + configureResources(ctx, t, mgmtServer, nodeID, listeners, []*v3routepb.RouteConfiguration{routeConfigForCluster(clusterB)}, clusters, endpoints) + + // The resolver first republishes with both clusters present, because that + // update is generated before the outgoing config selector is stopped. Once + // cluster-A's reference count reaches zero it is dropped, so wait for the + // service config that names cluster-B alone. A surplus reference on + // cluster-A would keep it in every subsequent update and time out here. + waitForServiceConfig(ctx, t, stateCh, wantServiceConfig(clusterB)) +} + // TestResolver_XDSConfigInRPCContext verifies that the xDS resolver's config // selector places the complete XDSConfig into the RPC context during config // selection, making it available to HTTP filters.