Skip to content

Commit 815e4cf

Browse files
ulascansenturkulascansenturk
committed
review comments
1 parent 9c11b34 commit 815e4cf

4 files changed

Lines changed: 114 additions & 134 deletions

File tree

‎internal/xds/resolver/cluster_specifier_plugin_test.go‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"context"
2323
"encoding/json"
2424
"fmt"
25+
"sync"
2526
"testing"
2627

2728
"github.com/google/uuid"
@@ -342,6 +343,85 @@ func (s) TestXDSResolverDelayedOnCommittedCSP(t *testing.T) {
342343
verifyUpdateFromResolver(ctx, t, stateCh, wantSC)
343344
}
344345

346+
// TestResolverClusterSpecifierPluginRefCountRace verifies that a cluster
347+
// specifier plugin is handled correctly when its last in-flight RPC is
348+
// committed at the same time as an xDS update that names it again. Whichever
349+
// happens first, the plugin must end up present in the service config: if the
350+
// commit lands first the entry is torn down and a fresh one replaces it, and if
351+
// the update lands first the existing entry is simply reused.
352+
func (s) TestResolverClusterSpecifierPluginRefCountRace(t *testing.T) {
353+
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
354+
defer cancel()
355+
nodeID := uuid.New().String()
356+
mgmtServer, _, _, bc := setupManagementServerForTest(t, nodeID)
357+
358+
routeConfigForPlugin := func(name, value string) []*v3routepb.RouteConfiguration {
359+
return []*v3routepb.RouteConfiguration{e2e.RouteConfigResourceWithOptions(e2e.RouteConfigOptions{
360+
RouteConfigName: defaultTestRouteConfigName,
361+
ListenerName: defaultTestServiceName,
362+
ClusterSpecifierType: e2e.RouteConfigClusterSpecifierTypeClusterSpecifierPlugin,
363+
ClusterSpecifierPluginName: name,
364+
ClusterSpecifierPluginConfig: testutils.MarshalAny(t, &wrapperspb.StringValue{Value: value}),
365+
})}
366+
}
367+
wantConfigForPlugin := func(name, value string) string {
368+
return fmt.Sprintf(`{
369+
"loadBalancingConfig": [{
370+
"xds_cluster_manager_experimental": {
371+
"children": {
372+
"cluster_specifier_plugin:%s": {
373+
"childPolicy": [{"csp_experimental": {"arbitrary_field": "%s"}}]
374+
}
375+
}
376+
}
377+
}]
378+
}`, name, value)
379+
}
380+
381+
listeners := []*v3listenerpb.Listener{e2e.DefaultClientListener(defaultTestServiceName, defaultTestRouteConfigName)}
382+
configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspA", "anythingA"), nil, nil)
383+
384+
stateCh, _, _ := buildResolverForTarget(t, resolver.Target{URL: *testutils.MustParseURL("xds:///" + defaultTestServiceName)}, bc)
385+
cs := verifyUpdateFromResolver(ctx, t, stateCh, wantConfigForPlugin("cspA", "anythingA"))
386+
387+
// Start an RPC on cspA and leave it uncommitted, so cspA stays referenced.
388+
res, err := cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
389+
if err != nil {
390+
t.Fatalf("cs.SelectConfig(): %v", err)
391+
}
392+
if got, want := clustermanager.PickedCluster(res.Context), "cluster_specifier_plugin:cspA"; got != want {
393+
t.Fatalf("Config selector returned cluster %q, want %q", got, want)
394+
}
395+
396+
// Move the route to cspB. cspA is now held only by the in-flight RPC.
397+
configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspB", "anythingB"), nil, nil)
398+
399+
// Commit the RPC, dropping cspA's last reference, while an update naming
400+
// cspA again is pushed concurrently. The two orderings exercise different
401+
// paths through the refcounted entry, and both must converge on cspA being
402+
// in the service config.
403+
var wg sync.WaitGroup
404+
wg.Add(1)
405+
go func() {
406+
defer wg.Done()
407+
res.OnCommitted()
408+
}()
409+
configureResources(ctx, t, mgmtServer, nodeID, listeners, routeConfigForPlugin("cspA", "anythingA"), nil, nil)
410+
wg.Wait()
411+
412+
cs = waitForServiceConfig(ctx, t, stateCh, wantConfigForPlugin("cspA", "anythingA"))
413+
414+
// The surviving entry must still be usable for new RPCs.
415+
res, err = cs.SelectConfig(iresolver.RPCInfo{Context: ctx, Method: "/service/method"})
416+
if err != nil {
417+
t.Fatalf("cs.SelectConfig() after the race: %v", err)
418+
}
419+
if got, want := clustermanager.PickedCluster(res.Context), "cluster_specifier_plugin:cspA"; got != want {
420+
t.Fatalf("Config selector returned cluster %q, want %q", got, want)
421+
}
422+
res.OnCommitted()
423+
}
424+
345425
// TestResolverClusterSpecifierPlugin_WithFilters tests the case where a route
346426
// configuration containing cluster specifier plugins is sent by the management
347427
// server, and HTTP filters are configured. The test verifies that the

‎internal/xds/resolver/helpers_test.go‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,36 @@ func verifyUpdateFromResolver(ctx context.Context, t *testing.T, stateCh chan re
173173
return cs
174174
}
175175

176+
// waitForServiceConfig drains updates from the resolver until one carries a
177+
// service config matching wantSC, and fails if none does before ctx expires.
178+
// Use this instead of verifyUpdateFromResolver when the resolver is expected to
179+
// publish intermediate configs on the way to the wanted one.
180+
//
181+
// Returns the config selector from the matching update.
182+
func waitForServiceConfig(ctx context.Context, t *testing.T, stateCh chan resolver.State, wantSC string) iresolver.ConfigSelector {
183+
t.Helper()
184+
185+
want := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(wantSC)
186+
for {
187+
select {
188+
case <-ctx.Done():
189+
t.Fatalf("Timeout waiting for the resolver to publish service config:\n%s", wantSC)
190+
case state := <-stateCh:
191+
if err := state.ServiceConfig.Err; err != nil {
192+
t.Fatalf("Received error in service config: %v", err)
193+
}
194+
if !internal.EqualServiceConfigForTesting(state.ServiceConfig.Config, want.Config) {
195+
continue
196+
}
197+
cs := iresolver.GetConfigSelector(state)
198+
if cs == nil {
199+
t.Fatal("Received nil config selector in update from resolver")
200+
}
201+
return cs
202+
}
203+
}
204+
}
205+
176206
// verifyNoUpdateFromResolver verifies that no update is pushed on stateCh.
177207
// Calls t.Fatal() if an update is received before defaultTestShortTimeout
178208
// expires.

‎internal/xds/resolver/serviceconfig_test.go‎

Lines changed: 0 additions & 112 deletions
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,14 @@ package resolver
2020

2121
import (
2222
"context"
23-
"errors"
2423
"regexp"
2524
"testing"
2625
"time"
2726

2827
xxhash "github.com/cespare/xxhash/v2"
29-
"google.golang.org/grpc/internal/grpcsync"
3028
"google.golang.org/grpc/internal/grpctest"
3129
"google.golang.org/grpc/internal/grpcutil"
3230
iresolver "google.golang.org/grpc/internal/resolver"
33-
"google.golang.org/grpc/internal/testutils"
3431
_ "google.golang.org/grpc/internal/xds/balancer/cdsbalancer" // To parse LB config
3532
"google.golang.org/grpc/internal/xds/xdsclient/xdsresource"
3633
"google.golang.org/grpc/metadata"
@@ -46,115 +43,6 @@ func Test(t *testing.T) {
4643
grpctest.RunSubTests(t, s{})
4744
}
4845

49-
// newResolverForActiveEntryTests returns a resolver with just enough state to
50-
// exercise acquireActiveClusterInfo and the cleanup it registers. The current
51-
// config selector is an erroring one so that pushing a new service config does
52-
// not require an xDS config to be present.
53-
func newResolverForActiveEntryTests(t *testing.T) *xdsResolver {
54-
t.Helper()
55-
56-
ctx, cancel := context.WithCancel(context.Background())
57-
t.Cleanup(cancel)
58-
r := &xdsResolver{
59-
cc: &testutils.ResolverClientConn{Logger: t},
60-
activeClusters: make(map[string]*grpcsync.RefCounted[*clusterInfo]),
61-
activePlugins: make(map[string]*grpcsync.RefCounted[*clusterInfo]),
62-
serializer: grpcsync.NewCallbackSerializer(ctx),
63-
serializerCancel: cancel,
64-
curConfigSelector: newErroringConfigSelector(errors.New("test"), ""),
65-
}
66-
r.logger = prefixLogger(r)
67-
return r
68-
}
69-
70-
// runOnSerializer runs f in the context of a serializer callback, which is the
71-
// only place the resolver's active cluster and plugin maps may be touched, and
72-
// blocks until it has run. Any callback queued by an earlier call to this
73-
// helper, including a removal queued when a reference count reached zero, is
74-
// guaranteed to have run by the time f is invoked.
75-
func runOnSerializer(ctx context.Context, t *testing.T, r *xdsResolver, f func()) {
76-
t.Helper()
77-
78-
done := make(chan struct{})
79-
r.serializer.TrySchedule(func(context.Context) {
80-
defer close(done)
81-
f()
82-
})
83-
select {
84-
case <-done:
85-
case <-ctx.Done():
86-
t.Fatal("Timeout waiting for serializer callback to run")
87-
}
88-
}
89-
90-
// TestActivePluginRefCounting verifies that repeated acquisitions of a cluster
91-
// specifier plugin share a single entry, and that the entry is removed from
92-
// activePlugins only once the last reference is released.
93-
func (s) TestActivePluginRefCounting(t *testing.T) {
94-
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
95-
defer cancel()
96-
97-
r := newResolverForActiveEntryTests(t)
98-
const key = "cluster_specifier_plugin:test-plugin"
99-
100-
var first, second *grpcsync.RefCounted[*clusterInfo]
101-
runOnSerializer(ctx, t, r, func() {
102-
first = r.acquireActiveClusterInfo(key, "")
103-
second = r.acquireActiveClusterInfo(key, "")
104-
})
105-
if first != second {
106-
t.Fatalf("acquireActiveClusterInfo(%q) returned a new entry; want the existing one to be reused", key)
107-
}
108-
109-
// Two references are outstanding, so releasing one must keep the entry.
110-
runOnSerializer(ctx, t, r, func() { first.Decrement() })
111-
runOnSerializer(ctx, t, r, func() {
112-
if _, ok := r.activePlugins[key]; !ok {
113-
t.Errorf("activePlugins[%q] was removed while a reference is still held", key)
114-
}
115-
})
116-
117-
// Releasing the last reference must remove the entry.
118-
runOnSerializer(ctx, t, r, func() { second.Decrement() })
119-
runOnSerializer(ctx, t, r, func() {
120-
if _, ok := r.activePlugins[key]; ok {
121-
t.Errorf("activePlugins[%q] still present after the last reference was released", key)
122-
}
123-
})
124-
}
125-
126-
// TestActivePluginNotRevivedAfterRelease verifies that an entry whose reference
127-
// count has already dropped to zero is replaced by a fresh entry rather than
128-
// resurrected, and that the pending removal of the dead entry does not delete
129-
// its replacement.
130-
func (s) TestActivePluginNotRevivedAfterRelease(t *testing.T) {
131-
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
132-
defer cancel()
133-
134-
r := newResolverForActiveEntryTests(t)
135-
const key = "cluster_specifier_plugin:test-plugin"
136-
137-
var dead, live *grpcsync.RefCounted[*clusterInfo]
138-
runOnSerializer(ctx, t, r, func() {
139-
// Release the only reference. The entry is now dead, but its removal is
140-
// queued behind this callback and so has not run yet. Acquiring again
141-
// must therefore hand back a fresh entry rather than revive this one.
142-
dead = r.acquireActiveClusterInfo(key, "")
143-
dead.Decrement()
144-
live = r.acquireActiveClusterInfo(key, "")
145-
})
146-
if live == dead {
147-
t.Fatal("acquireActiveClusterInfo() returned an entry whose refcount had already reached zero; want a new entry")
148-
}
149-
150-
// The dead entry's queued removal must not evict the replacement.
151-
runOnSerializer(ctx, t, r, func() {
152-
if got := r.activePlugins[key]; got != live {
153-
t.Errorf("activePlugins[%q] = %p, want the newly created entry %p", key, got, live)
154-
}
155-
})
156-
}
157-
15846
func (s) TestGenerateRequestHash(t *testing.T) {
15947
const channelID = 12378921
16048
cs := &configSelector{channelID: channelID}

‎internal/xds/resolver/xds_resolver_test.go‎

Lines changed: 4 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1382,18 +1382,10 @@ func (s) TestResolverKeepWatchOpen_ActiveRPCs(t *testing.T) {
13821382
res.OnCommitted()
13831383
}
13841384

1385-
// TestResolver_XDSConfigInRPCContext verifies that the xDS resolver's config
1386-
// selector places the complete XDSConfig into the RPC context during config
1387-
// selection, making it available to HTTP filters.
13881385
// TestResolverClusterSharedByMultipleRoutes verifies the reference accounting
13891386
// for a cluster that more than one route points at. A config selector takes a
13901387
// single reference per distinct cluster, however many routes name it, and
13911388
// releases exactly that one reference when it is stopped.
1392-
//
1393-
// If a reference were taken per route instead, the counts would not balance:
1394-
// configSelector.stop() decrements once per entry in its cluster map, so the
1395-
// surplus references would keep the cluster in activeClusters and in the
1396-
// service config forever, with its CDS watch open.
13971389
func (s) TestResolverClusterSharedByMultipleRoutes(t *testing.T) {
13981390
clusterA := "cluster-A"
13991391
clusterB := "cluster-B"
@@ -1458,22 +1450,12 @@ func (s) TestResolverClusterSharedByMultipleRoutes(t *testing.T) {
14581450
// cluster-A's reference count reaches zero it is dropped, so wait for the
14591451
// service config that names cluster-B alone. A surplus reference on
14601452
// cluster-A would keep it in every subsequent update and time out here.
1461-
wantFinal := internal.ParseServiceConfig.(func(string) *serviceconfig.ParseResult)(wantServiceConfig(clusterB))
1462-
for {
1463-
select {
1464-
case <-ctx.Done():
1465-
t.Fatal("Timeout waiting for cluster-A to be dropped from the service config")
1466-
case state := <-stateCh:
1467-
if err := state.ServiceConfig.Err; err != nil {
1468-
t.Fatalf("Received error in service config: %v", err)
1469-
}
1470-
if internal.EqualServiceConfigForTesting(state.ServiceConfig.Config, wantFinal.Config) {
1471-
return
1472-
}
1473-
}
1474-
}
1453+
waitForServiceConfig(ctx, t, stateCh, wantServiceConfig(clusterB))
14751454
}
14761455

1456+
// TestResolver_XDSConfigInRPCContext verifies that the xDS resolver's config
1457+
// selector places the complete XDSConfig into the RPC context during config
1458+
// selection, making it available to HTTP filters.
14771459
func (s) TestResolver_XDSConfigInRPCContext(t *testing.T) {
14781460
// Spin up an xDS management server for the test.
14791461
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)

0 commit comments

Comments
 (0)