priority: implement changes to child policy caching specified in A115 (#8997)
Summary of changes:
- Add a new API to the `balancergroup` that allows removal of child
policies with immediate effect, even when a cache is configured.
- Make the `priority` LB use this new API when removing child policies
that have been removed from its configuration.
- This change is protected with an environment variable that turns on
the new behavior by default.
- Update the name generation algorithm to give higher preference to
reuse names of existing priority (if there is a locality match) over
lower priorities.
- Fix a bunch of tests in the `priority` LB policy to ensure that they
work with the *old* and *new* behavior.
- Fix a couple of aggregate cluster tests such that they don't
continuously handle the same NACKed resource from the management server.
- Other minor cleanups in tests that I had to look at as part of making
this change.
RELEASE NOTES:
* xds/priority: Stop caching child LB policies removed from the
configuration. This will help reduce memory and cpu usage when
localities are constantly switching between priorities.
diff --git a/balancer/weightedtarget/weightedtarget_test.go b/balancer/weightedtarget/weightedtarget_test.go
index d78db53..8468892 100644
--- a/balancer/weightedtarget/weightedtarget_test.go
+++ b/balancer/weightedtarget/weightedtarget_test.go
@@ -1348,7 +1348,7 @@
return ret
}
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
var errTestInitIdle = fmt.Errorf("init Idle balancer error 0")
@@ -1388,7 +1388,7 @@
"targets": {
"cluster_1": {
"weight":1,
- "childPolicy": [{"test-init-Idle-balancer": ""}]
+ "childPolicy": [{"test-init-idle-balancer": ""}]
}
}
}`))
diff --git a/internal/balancergroup/balancergroup.go b/internal/balancergroup/balancergroup.go
index cd1a9ed..8acb942 100644
--- a/internal/balancergroup/balancergroup.go
+++ b/internal/balancergroup/balancergroup.go
@@ -338,6 +338,16 @@
// closed after timeout. Cleanup work (closing sub-balancer and removing
// subconns) will be done after timeout.
func (bg *BalancerGroup) Remove(id string) {
+ bg.removeInternal(id, true)
+}
+
+// RemoveImmediately removes and closes the balancer with id from the group
+// immediately.
+func (bg *BalancerGroup) RemoveImmediately(id string) {
+ bg.removeInternal(id, false)
+}
+
+func (bg *BalancerGroup) removeInternal(id string, withCaching bool) {
bg.logger.Infof("Removing child policy for child %q", id)
bg.outgoingMu.Lock()
@@ -356,32 +366,40 @@
// Unconditionally remove the sub-balancer config from the map.
delete(bg.idToBalancerConfig, id)
- if bg.deletedBalancerCache != nil {
- if bg.logger.V(2) {
- bg.logger.Infof("Adding child policy for child %q to the balancer cache", id)
- bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
- }
-
- bg.deletedBalancerCache.Add(id, sbToRemove, func() {
+ if withCaching {
+ if bg.deletedBalancerCache != nil {
if bg.logger.V(2) {
- bg.logger.Infof("Removing child policy for child %q from the balancer cache after timeout", id)
+ bg.logger.Infof("Adding child policy for child %q to the balancer cache", id)
+ }
+ bg.deletedBalancerCache.Add(id, sbToRemove, func() {
+ if bg.logger.V(2) {
+ bg.logger.Infof("Removing child policy for child %q from the balancer cache after timeout", id)
+ bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
+ }
+
+ // A sub-balancer evicted from the timeout cache needs to closed
+ // and its subConns need to removed, unconditionally. There is a
+ // possibility that a sub-balancer might be removed (thereby
+ // moving it to the cache) around the same time that the
+ // balancergroup is closed, and by the time we get here the
+ // balancergroup might be closed. Check for `outgoingStarted ==
+ // true` at that point can lead to a leaked sub-balancer.
+ bg.outgoingMu.Lock()
+ sbToRemove.stopBalancer()
+ bg.outgoingMu.Unlock()
+ bg.cleanupSubConns(sbToRemove)
+ })
+ if bg.logger.V(2) {
bg.logger.Infof("Number of items remaining in the balancer cache: %d", bg.deletedBalancerCache.Len())
}
-
- // A sub-balancer evicted from the timeout cache needs to closed
- // and its subConns need to removed, unconditionally. There is a
- // possibility that a sub-balancer might be removed (thereby
- // moving it to the cache) around the same time that the
- // balancergroup is closed, and by the time we get here the
- // balancergroup might be closed. Check for `outgoingStarted ==
- // true` at that point can lead to a leaked sub-balancer.
- bg.outgoingMu.Lock()
- sbToRemove.stopBalancer()
bg.outgoingMu.Unlock()
- bg.cleanupSubConns(sbToRemove)
- })
- bg.outgoingMu.Unlock()
- return
+ return
+ }
+
+ // Fall through to remove the sub-balancer with immediate effect if we are not caching.
+ if bg.logger.V(2) {
+ bg.logger.Infof("Child policy for child %q was requested to be cached before eventual removal. No such cache exists. Removing right away.", id)
+ }
}
// Remove the sub-balancer with immediate effect if we are not caching.
@@ -481,7 +499,7 @@
// from map. Delete sc from the map only when state changes to Shutdown. Since
// it's just forwarding the action, there's no need for a removeSubConn()
// wrapper function.
-func (bg *BalancerGroup) newSubConn(config *subBalancerWrapper, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
+func (bg *BalancerGroup) newSubConn(sbw *subBalancerWrapper, addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
// NOTE: if balancer with id was already removed, this should also return
// error. But since we call balancer.stopBalancer when removing the balancer, this
// shouldn't happen.
@@ -493,12 +511,12 @@
var sc balancer.SubConn
oldListener := opts.StateListener
opts.StateListener = func(state balancer.SubConnState) { bg.updateSubConnState(sc, state, oldListener) }
- sc, err := bg.cc.NewSubConn(addrs, opts)
+ sc, err := sbw.ClientConn.NewSubConn(addrs, opts)
if err != nil {
bg.incomingMu.Unlock()
return nil, err
}
- bg.scToSubBalancer[sc] = config
+ bg.scToSubBalancer[sc] = sbw
bg.incomingMu.Unlock()
return sc, nil
}
diff --git a/internal/balancergroup/balancergroup_ext_test.go b/internal/balancergroup/balancergroup_ext_test.go
new file mode 100644
index 0000000..bf53d99
--- /dev/null
+++ b/internal/balancergroup/balancergroup_ext_test.go
@@ -0,0 +1,124 @@
+/*
+ * Copyright 2026 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package balancergroup_test
+
+import (
+ "context"
+ "strings"
+ "testing"
+ "time"
+
+ "google.golang.org/grpc/balancer"
+ "google.golang.org/grpc/internal/balancer/stub"
+ "google.golang.org/grpc/internal/balancergroup"
+ "google.golang.org/grpc/internal/grpctest"
+ "google.golang.org/grpc/internal/testutils"
+)
+
+type s struct {
+ grpctest.Tester
+}
+
+func Test(t *testing.T) {
+ grpctest.RunSubTests(t, s{})
+}
+
+const (
+ defaultTestTimeout = 5 * time.Second
+ defaultTestShortTimeout = 100 * time.Millisecond
+)
+
+// Tests verifies that the RemoveImmediately method of balancergroup closes the
+// child balancer immediately, while the Remove method does not close the child
+// balancer immediately, when the cache is enabled.
+func (s) TestBalancerGroup_RemoveImmediately(t *testing.T) {
+ // Channels to track child balancer creation and closure.
+ childLBCreated := make(chan string, 1)
+ childLBClosed := make(chan string, 1)
+
+ childLBName1 := strings.ToLower(t.Name()) + "-child-1"
+ t.Logf("Registering a child balancer with name %q", childLBName1)
+ stub.Register(childLBName1, stub.BalancerFuncs{
+ Init: func(*stub.BalancerData) {
+ childLBCreated <- childLBName1
+ },
+ Close: func(*stub.BalancerData) {
+ childLBClosed <- childLBName1
+ },
+ })
+
+ childLBName2 := strings.ToLower(t.Name()) + "-child-2"
+ t.Logf("Registering a child balancer with name %q", childLBName2)
+ stub.Register(childLBName2, stub.BalancerFuncs{
+ Init: func(*stub.BalancerData) {
+ childLBCreated <- childLBName2
+ },
+ Close: func(*stub.BalancerData) {
+ childLBClosed <- childLBName2
+ },
+ })
+
+ t.Logf("Creating a balancergroup with cache enabled")
+ tcc := testutils.NewBalancerClientConn(t)
+ bg := balancergroup.New(balancergroup.Options{
+ CC: tcc,
+ BuildOpts: balancer.BuildOptions{},
+ SubBalancerCloseTimeout: defaultTestTimeout,
+ })
+
+ t.Logf("Adding a child balancer with name %q to the group", "child-1")
+ bg.AddWithClientConn("child-1", childLBName1, tcc)
+ select {
+ case <-childLBCreated:
+ case <-time.After(defaultTestTimeout):
+ t.Fatalf("Timeout when waiting for child LB to be created")
+ }
+
+ t.Logf("Adding a child balancer with name %q to the group", "child-2")
+ bg.AddWithClientConn("child-2", childLBName2, tcc)
+ select {
+ case <-childLBCreated:
+ case <-time.After(defaultTestTimeout):
+ t.Fatalf("Timeout when waiting for child LB to be created")
+ }
+
+ t.Logf("Removing the child balancer with name %q from the group with immediate effect", "child-1")
+ bg.RemoveImmediately("child-1")
+ select {
+ case <-childLBClosed:
+ case <-time.After(defaultTestTimeout):
+ t.Fatalf("Timeout when waiting for child LB to be closed")
+ }
+
+ t.Logf("Removing the child balancer with name %q from the group with caching", "child-2")
+ bg.Remove("child-2")
+ sCtx, sCancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
+ defer sCancel()
+ select {
+ case <-childLBClosed:
+ t.Fatalf("Child LB closed when expected to be cached")
+ case <-sCtx.Done():
+ }
+
+ t.Logf("Closing the balancergroup, which should close the second child balancer")
+ bg.Close()
+ select {
+ case <-childLBClosed:
+ case <-time.After(defaultTestTimeout):
+ t.Fatalf("Timeout when waiting for child LB to be closed")
+ }
+}
diff --git a/internal/envconfig/envconfig.go b/internal/envconfig/envconfig.go
index e3adb91..3ae45fa 100644
--- a/internal/envconfig/envconfig.go
+++ b/internal/envconfig/envconfig.go
@@ -119,6 +119,13 @@
// A future release will remove this environment variable, enabling strict
// path checking behavior unconditionally.
DisableStrictPathChecking = boolFromEnv("GRPC_GO_EXPERIMENTAL_DISABLE_STRICT_PATH_CHECKING", false)
+
+ // EnablePriorityLBChildPolicyCache controls whether the priority balancer
+ // should cache child balancers that are removed from the LB policy config,
+ // for a period of 15 minutes. This is disabled by default, but can be
+ // enabled by setting the env variable
+ // GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE to true.
+ EnablePriorityLBChildPolicyCache = boolFromEnv("GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE", false)
)
func boolFromEnv(envVar string, def bool) bool {
diff --git a/internal/xds/balancer/cdsbalancer/configbuilder_childname.go b/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
index 8f118d4..2b0fedd 100644
--- a/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
+++ b/internal/xds/balancer/cdsbalancer/configbuilder_childname.go
@@ -31,9 +31,10 @@
// struct keeps state between generate() calls, and a later generate() might
// return names returned by the previous call.
type nameGenerator struct {
- existingNames map[clients.Locality]string
- prefix uint64
- nextID uint64
+ prevLocalitiesToChildNames map[clients.Locality]string // locality to child name mapping generated for the previous update
+ prevChildNames []string // prioritized list of child names generated for the previous update
+ prefix uint64
+ nextID uint64
}
func newNameGenerator(prefix uint64) *nameGenerator {
@@ -53,36 +54,58 @@
// - update 3: [[L1, L2], [L3]] --> ["0", "2"] (Two priorities were merged)
// - update 4: [[L1], [L4]] --> ["0", "3",] (A priority was split, and a new priority was added)
func (ng *nameGenerator) generate(priorities [][]xdsresource.Locality) []string {
- var ret []string
+ ret := make([]string, len(priorities))
usedNames := make(map[string]bool)
newNames := make(map[clients.Locality]string)
- for _, priority := range priorities {
- var nameFound string
+
+ // Pass 1: Same priority index match.
+ for i, priority := range priorities {
+ if i >= len(ng.prevChildNames) {
+ continue
+ }
+ targetName := ng.prevChildNames[i]
for _, locality := range priority {
- if name, ok := ng.existingNames[locality.ID]; ok {
+ if name, ok := ng.prevLocalitiesToChildNames[locality.ID]; ok && name == targetName {
+ ret[i] = targetName
+ usedNames[targetName] = true
+ break
+ }
+ }
+ }
+
+ // Pass 2: Greedy reuse.
+ for i, priority := range priorities {
+ if ret[i] != "" {
+ continue
+ }
+ for _, locality := range priority {
+ if name, ok := ng.prevLocalitiesToChildNames[locality.ID]; ok {
if !usedNames[name] {
- nameFound = name
- // Found a name to use. No need to process the remaining
- // localities.
+ ret[i] = name
+ usedNames[name] = true
break
}
}
}
-
- if nameFound == "" {
- // No appropriate used name is found. Make a new name.
- nameFound = fmt.Sprintf("priority-%d-%d", ng.prefix, ng.nextID)
- ng.nextID++
- }
-
- ret = append(ret, nameFound)
- // All localities in this priority share the same name. Add them all to
- // the new map.
- for _, l := range priority {
- newNames[l.ID] = nameFound
- }
- usedNames[nameFound] = true
}
- ng.existingNames = newNames
+
+ // Pass 3: New name.
+ for i, name := range ret {
+ if name == "" {
+ newID := fmt.Sprintf("priority-%d-%d", ng.prefix, ng.nextID)
+ ng.nextID++
+ ret[i] = newID
+ usedNames[newID] = true
+ }
+ }
+
+ // Update state.
+ for i, priority := range priorities {
+ for _, l := range priority {
+ newNames[l.ID] = ret[i]
+ }
+ }
+ ng.prevLocalitiesToChildNames = newNames
+ ng.prevChildNames = ret
return ret
}
diff --git a/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go b/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
index e534d54..7275f55 100644
--- a/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
+++ b/internal/xds/balancer/cdsbalancer/configbuilder_childname_test.go
@@ -29,82 +29,282 @@
tests := []struct {
name string
prefix uint64
- input1 [][]xdsresource.Locality
- input2 [][]xdsresource.Locality
- want []string
+ steps []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }
}{
{
name: "init, two new priorities",
prefix: 3,
- input1: nil,
- input2: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L1"}}},
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-3-0", "priority-3-1"},
+ },
},
- want: []string{"priority-3-0", "priority-3-1"},
},
{
name: "one new priority",
prefix: 1,
- input1: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ },
+ want: []string{"priority-1-0"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-1-0", "priority-1-1"},
+ },
},
- input2: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L1"}}},
- },
- want: []string{"priority-1-0", "priority-1-1"},
},
{
name: "merge two priorities",
prefix: 4,
- input1: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L1"}}},
- {{ID: clients.Locality{Zone: "L2"}}},
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-4-0", "priority-4-1", "priority-4-2"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-4-0", "priority-4-2"},
+ },
},
- input2: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
- {{ID: clients.Locality{Zone: "L2"}}},
- },
- want: []string{"priority-4-0", "priority-4-2"},
},
{
name: "swap two priorities",
- input1: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L1"}}},
- {{ID: clients.Locality{Zone: "L2"}}},
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1", "priority-0-2"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-0-1", "priority-0-0", "priority-0-2"},
+ },
},
- input2: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L1"}}},
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L2"}}},
- },
- want: []string{"priority-0-1", "priority-0-0", "priority-0-2"},
},
{
name: "split priority",
- input1: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
- {{ID: clients.Locality{Zone: "L2"}}},
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-2", "priority-0-1"},
+ },
},
- input2: [][]xdsresource.Locality{
- {{ID: clients.Locality{Zone: "L0"}}},
- {{ID: clients.Locality{Zone: "L1"}}}, // This gets a newly generated name, since "0-0" was already picked.
- {{ID: clients.Locality{Zone: "L2"}}},
+ },
+ {
+ name: "priority index preference",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L0"}}},
+ },
+ want: []string{"priority-0-0"},
+ },
},
- want: []string{"priority-0-0", "priority-0-2", "priority-0-1"},
+ },
+ {
+ name: "merge partial",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ },
+ },
+ {
+ name: "swap shift",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-0-1", "priority-0-0"},
+ },
+ },
+ },
+ {
+ name: "replace priority",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L2"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}},
+ {{ID: clients.Locality{Zone: "L5"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-2"},
+ },
+ },
+ },
+ {
+ name: "reordered merge",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L1"}}, {ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L0"}}, {ID: clients.Locality{Zone: "L3"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ },
+ },
+ {
+ name: "three-step shift stability",
+ steps: []struct {
+ input [][]xdsresource.Locality
+ want []string
+ }{
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L2"}}},
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-0-2", "priority-0-0", "priority-0-1"},
+ },
+ {
+ input: [][]xdsresource.Locality{
+ {{ID: clients.Locality{Zone: "L0"}}},
+ {{ID: clients.Locality{Zone: "L1"}}},
+ },
+ want: []string{"priority-0-0", "priority-0-1"},
+ },
+ },
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
ng := newNameGenerator(tt.prefix)
- got1 := ng.generate(tt.input1)
- t.Logf("%v", got1)
- got := ng.generate(tt.input2)
- if diff := cmp.Diff(got, tt.want); diff != "" {
- t.Errorf("generate() = got: %v, want: %v, diff (-got +want): %s", got, tt.want, diff)
+ for i, step := range tt.steps {
+ got := ng.generate(step.input)
+ if diff := cmp.Diff(got, step.want); diff != "" {
+ t.Errorf("step %d: generate() = got: %v, want: %v, diff (-got +want): %s", i, got, step.want, diff)
+ }
}
})
}
diff --git a/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go b/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
index 76de30a..8b75dbe 100644
--- a/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
+++ b/internal/xds/balancer/cdsbalancer/e2e_test/aggregate_cluster_test.go
@@ -23,6 +23,7 @@
"slices"
"strconv"
"strings"
+ "sync/atomic"
"testing"
"time"
@@ -697,8 +698,8 @@
bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
// Start two test backends.
- servers, cleanup3 := startTestServiceBackends(t, 2)
- defer cleanup3()
+ servers, cleanup := startTestServiceBackends(t, 2)
+ defer cleanup()
addrs, _ := backendAddressesAndPorts(t, servers)
// Configure an aggregate cluster pointing to an EDS and LOGICAL_DNS
@@ -795,8 +796,28 @@
// the DNS resolver pushes an update, the test verifies that we switch to the
// DNS cluster and can make a successful RPC.
func (s) TestAggregateCluster_BadEDSFromError_GoodToBadDNS(t *testing.T) {
- // Start an xDS management server.
- managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+ // Start an xDS management server and block it from continuously re-sending
+ // the same EDS response that triggers a NACK. This is to ensure that the
+ // test can proceed to the point of verifying that we fall back to the
+ // LOGICAL_DNS cluster. More more details see:
+ // https://github.com/grpc/grpc-go/issues/8994.
+ ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+ defer cancel()
+ edsResponseSeen := atomic.Bool{}
+ managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+ OnStreamResponse: func(_ context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, resp *v3discoverypb.DiscoveryResponse) {
+ if resp.GetTypeUrl() != version.V3EndpointsURL {
+ return
+ }
+ if edsResponseSeen.Load() {
+ t.Logf("Received EDS response again, blocking to prevent test from busylooping on the same EDS response: %v", resp)
+ <-ctx.Done()
+ return
+ }
+ edsResponseSeen.Store(true)
+ },
+ AllowResourceSubset: true,
+ })
// Create bootstrap configuration pointing to the above management server.
nodeID := uuid.New().String()
@@ -834,8 +855,6 @@
Endpoints: []*v3endpointpb.ClusterLoadAssignment{nackEndpointResource},
SkipValidation: true,
}
- ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
- defer cancel()
if err := managementServer.Update(ctx, resources); err != nil {
t.Fatal(err)
}
@@ -1093,8 +1112,29 @@
// the LOGICAL_DNS cluster, because it is supposed to treat the bad EDS response
// as though it received an update with no endpoints.
func (s) TestAggregateCluster_Fallback_EDSNackedWithoutPreviousGoodUpdate(t *testing.T) {
- // Start an xDS management server.
- managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{AllowResourceSubset: true})
+ ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+ defer cancel()
+
+ // Start an xDS management server and block it from continuously re-sending
+ // the same EDS response that triggers a NACK. This is to ensure that the
+ // test can proceed to the point of verifying that we fall back to the
+ // LOGICAL_DNS cluster. More more details see:
+ // https://github.com/grpc/grpc-go/issues/8994.
+ edsResponseSeen := atomic.Bool{}
+ managementServer := e2e.StartManagementServer(t, e2e.ManagementServerOptions{
+ OnStreamResponse: func(_ context.Context, _ int64, _ *v3discoverypb.DiscoveryRequest, resp *v3discoverypb.DiscoveryResponse) {
+ if resp.GetTypeUrl() != version.V3EndpointsURL {
+ return
+ }
+ if edsResponseSeen.Load() {
+ t.Logf("Received EDS response again, blocking to prevent test from busylooping on the same EDS response: %v", resp)
+ <-ctx.Done()
+ return
+ }
+ edsResponseSeen.Store(true)
+ },
+ AllowResourceSubset: true,
+ })
// Create bootstrap configuration pointing to the above management server.
nodeID := uuid.New().String()
@@ -1103,8 +1143,8 @@
// Start two test backends and extract their host and port. The first
// backend is used for the EDS cluster and the second backend is used for
// the LOGICAL_DNS cluster.
- servers, cleanup3 := startTestServiceBackends(t, 2)
- defer cleanup3()
+ servers, cleanup := startTestServiceBackends(t, 2)
+ defer cleanup()
addrs, ports := backendAddressesAndPorts(t, servers)
dnsHostName, dnsPort := hostAndPortFromAddress(t, servers[1].Address)
@@ -1128,11 +1168,7 @@
// Set a load balancing weight of 0 for the backend in the EDS resource.
// This is expected to be NACKed by the xDS client. Since the
- // cluster_resolver LB policy has no previously received good EDS resource,
- // it will treat this as though it received an update with no endpoints.
resources.Endpoints[0].Endpoints[0].LbEndpoints[0].LoadBalancingWeight = &wrapperspb.UInt32Value{Value: 0}
- ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
- defer cancel()
if err := managementServer.Update(ctx, resources); err != nil {
t.Fatal(err)
}
diff --git a/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go b/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
index a92a2ac..ee45ca3 100644
--- a/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
+++ b/internal/xds/balancer/cdsbalancer/e2e_test/eds_impl_test.go
@@ -236,8 +236,8 @@
bootstrapContents := e2e.DefaultBootstrapContents(t, nodeID, managementServer.Address)
// Start backend servers which provide an implementation of the TestService.
- servers, cleanup2 := startTestServiceBackends(t, 4)
- defer cleanup2()
+ servers, cleanup := startTestServiceBackends(t, 4)
+ defer cleanup()
addrs, ports := backendAddressesAndPorts(t, servers)
// Create xDS resources for consumption by the test. We start off with two
diff --git a/internal/xds/balancer/clustermanager/clustermanager_test.go b/internal/xds/balancer/clustermanager/clustermanager_test.go
index 17a3026..576f440 100644
--- a/internal/xds/balancer/clustermanager/clustermanager_test.go
+++ b/internal/xds/balancer/clustermanager/clustermanager_test.go
@@ -513,7 +513,7 @@
}
}
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
var errTestInitIdle = fmt.Errorf("init Idle balancer error 0")
@@ -551,7 +551,7 @@
configJSON1 := `{
"children": {
- "cds:cluster_1":{ "childPolicy": [{"test-init-Idle-balancer":""}] }
+ "cds:cluster_1":{ "childPolicy": [{"test-init-idle-balancer":""}] }
}
}`
config1, err := parser.ParseConfig([]byte(configJSON1))
diff --git a/internal/xds/balancer/priority/balancer.go b/internal/xds/balancer/priority/balancer.go
index 950cd13..e842920 100644
--- a/internal/xds/balancer/priority/balancer.go
+++ b/internal/xds/balancer/priority/balancer.go
@@ -34,6 +34,7 @@
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/internal/balancergroup"
"google.golang.org/grpc/internal/buffer"
+ "google.golang.org/grpc/internal/envconfig"
"google.golang.org/grpc/internal/grpclog"
"google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/hierarchy"
@@ -153,7 +154,7 @@
// The balancing policy name is changed, close the old child. But don't
// rebuild, rebuild will happen when syncing priorities.
if currentChild.balancerName != bb.Name() {
- currentChild.stop()
+ currentChild.stop(true)
currentChild.updateBalancerName(bb.Name())
}
@@ -169,7 +170,7 @@
// Cleanup resources used by children removed from the config.
for name, oldChild := range b.children {
if _, ok := newConfig.Children[name]; !ok {
- oldChild.stop()
+ oldChild.stop(!envconfig.EnablePriorityLBChildPolicyCache)
delete(b.children, name)
}
}
@@ -230,7 +231,7 @@
// Stop the child policies, this is necessary to stop the init timers in the
// children.
for _, child := range b.children {
- child.stop()
+ child.stop(true)
}
}
diff --git a/internal/xds/balancer/priority/balancer_child.go b/internal/xds/balancer/priority/balancer_child.go
index 5e76391..ad5b5a8 100644
--- a/internal/xds/balancer/priority/balancer_child.go
+++ b/internal/xds/balancer/priority/balancer_child.go
@@ -133,12 +133,16 @@
// It doesn't do it directly. It asks the balancer group to remove it.
//
// Note that the underlying balancer group could keep the child in a cache.
-func (cb *childBalancer) stop() {
+func (cb *childBalancer) stop(immediate bool) {
if !cb.started {
return
}
cb.stopInitTimer()
- cb.parent.bg.Remove(cb.name)
+ if immediate {
+ cb.parent.bg.RemoveImmediately(cb.name)
+ } else {
+ cb.parent.bg.Remove(cb.name)
+ }
cb.started = false
cb.state = balancer.State{
ConnectivityState: connectivity.Connecting,
diff --git a/internal/xds/balancer/priority/balancer_priority.go b/internal/xds/balancer/priority/balancer_priority.go
index a57df7d..3181ce2 100644
--- a/internal/xds/balancer/priority/balancer_priority.go
+++ b/internal/xds/balancer/priority/balancer_priority.go
@@ -128,7 +128,7 @@
b.logger.Warningf("Priority name %q is not found in list of child policies", name)
continue
}
- child.stop()
+ child.stop(false)
}
}
diff --git a/internal/xds/balancer/priority/balancer_test.go b/internal/xds/balancer/priority/balancer_test.go
index 070e95c..c1c1572 100644
--- a/internal/xds/balancer/priority/balancer_test.go
+++ b/internal/xds/balancer/priority/balancer_test.go
@@ -29,6 +29,7 @@
"google.golang.org/grpc/balancer/roundrobin"
"google.golang.org/grpc/connectivity"
"google.golang.org/grpc/internal/balancer/stub"
+ "google.golang.org/grpc/internal/envconfig"
"google.golang.org/grpc/internal/grpctest"
"google.golang.org/grpc/internal/hierarchy"
internalserviceconfig "google.golang.org/grpc/internal/serviceconfig"
@@ -68,9 +69,6 @@
for i := 0; i < testBackendAddrsCount; i++ {
testBackendAddrStrs = append(testBackendAddrStrs, fmt.Sprintf("%d.%d.%d.%d:%d", i, i, i, i, i))
}
- // Disable sub-balancer caching for all but the tests which exercise the
- // caching behavior.
- DefaultSubBalancerCloseTimeout = time.Duration(0)
balancer.Register(&anotherRR{Builder: balancer.Get(roundrobin.Name)})
}
@@ -85,6 +83,17 @@
//
// Init 0 and 1; 0 is up, use 0; add 2, use 0; remove 2, use 0.
func (s) TestPriority_HighPriorityReady(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityHighPriorityReady(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityHighPriorityReady(t)
+ })
+}
+
+func testPriorityHighPriorityReady(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -199,6 +208,17 @@
// Init 0 and 1; 0 is up, use 0; 0 is down, 1 is up, use 1; add 2, use 1; 1 is
// down, use 2; remove 2, use 1.
func (s) TestPriority_SwitchPriority(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPrioritySwitchPriority(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPrioritySwitchPriority(t)
+ })
+}
+
+func testPrioritySwitchPriority(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -340,16 +360,19 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // p2 SubConns are shut down.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc2 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc2, scToShutdown)
+ // p2 will be closed immediately only if the cache is not enabled.
+ if !envconfig.EnablePriorityLBChildPolicyCache {
+ // p2 SubConns are shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when
+ // they are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc2 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc2, scToShutdown)
+ }
}
// Should get an update with 1's old transient failure picker, to override
@@ -374,6 +397,10 @@
//
// Init 0 and 1; 0 is up, use 0; 0 is connecting, 1 is up, use 1; 0 is ready,
// use 0.
+//
+// Env var GRPC_EXPERIMENTAL_ENABLE_PRIORITY_LB_CHILD_POLICY_CACHE does not
+// affect whether a lower priority child policy is cached when it is removed
+// from the balancer group.
func (s) TestPriority_HighPriorityToConnectingFromReady(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -383,7 +410,7 @@
pb := bb.Build(cc, balancer.BuildOptions{})
defer pb.Close()
- // Two localities, with priorities [0, 1], each with one backend.
+ t.Log("Two localities, with priorities [0, 1], each with one backend.")
if err := pb.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{
Endpoints: []resolver.Endpoint{
@@ -408,7 +435,7 @@
}
sc0 := <-cc.NewSubConnCh
- // p0 is ready.
+ t.Log("Make p0 ready.")
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
@@ -417,16 +444,15 @@
t.Fatal(err.Error())
}
- // Turn 0 to TransientFailure, will start and use 1.
+ t.Log("Turn down 0, will start and use 1.")
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.TransientFailure})
-
// Before 1 gets READY, picker should return NoSubConnAvailable, so RPCs
// will retry.
if err := cc.WaitForPickerWithErr(ctx, balancer.ErrNoSubConnAvailable); err != nil {
t.Fatal(err.Error())
}
- // Handle SubConn creation from 1.
+ t.Log("Handle SubConn creation from 1.")
addrs1 := <-cc.NewSubConnAddrsCh
if got, want := addrs1[0].Addr, testBackendAddrStrs[1]; got != want {
t.Fatalf("sc is created with addr %v, want %v", got, want)
@@ -440,22 +466,15 @@
t.Fatal(err.Error())
}
- // Turn 0 back to Ready.
+ t.Logf("Turn 0 back to Ready.")
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
- // p1 subconn should be shut down.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc1 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
- }
-
+ // At this point p1 is removed from the balancer group, but it gets cached.
+ // Therefore its subchannels will not be removed immediately. So, we do not
+ // check for the shutdown of p1's subchannels here. Instead, we check that
+ // p0 is used again after it becomes ready, which indicates that the
+ // priority switch is successful.
if err := cc.WaitForRoundRobinPicker(ctx, sc0); err != nil {
t.Fatal(err.Error())
}
@@ -646,25 +665,6 @@
// When 0 becomes ready, 0 should be used, 1 and 2 should all be closed.
sc0.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
- // sc1 and sc2 should be shut down.
- //
- // With localities caching, the lower priorities are closed after a timeout,
- // in goroutines. The order is no longer guaranteed.
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- scToShutdown := [2]balancer.SubConn{}
- scToShutdown[0] = <-cc.ShutdownSubConnCh
- <-cc.ShutdownSubConnCh
- scToShutdown[1] = <-cc.ShutdownSubConnCh
- <-cc.ShutdownSubConnCh
-
- if !(scToShutdown[0] == sc1 && scToShutdown[1] == sc2) && !(scToShutdown[0] == sc2 && scToShutdown[1] == sc1) {
- t.Errorf("ShutdownSubConn, want [%v, %v], got %v", sc1, sc2, scToShutdown)
- }
-
// Test pick with 0.
if err := cc.WaitForRoundRobinPicker(ctx, sc0); err != nil {
t.Fatal(err.Error())
@@ -741,6 +741,17 @@
// EDS removes all priorities, and re-adds them.
func (s) TestPriority_RemovesAllPriorities(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityRemovesAllPriorities(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityRemovesAllPriorities(t)
+ })
+}
+
+func testPriorityRemovesAllPriorities(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -797,16 +808,19 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // p0 subconn should be shut down.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc0 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+ // p0 will be closed immediately only if the cache is not enabled.
+ if !envconfig.EnablePriorityLBChildPolicyCache {
+ // p0 SubConns are shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when
+ // they are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc0 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+ }
}
// Test pick return TransientFailure.
@@ -839,6 +853,14 @@
}
sc01 := <-cc.NewSubConnCh
+ if envconfig.EnablePriorityLBChildPolicyCache {
+ // The old p0 SubConn is shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ if scToShutdown != sc0 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+ }
+ }
+
// Don't send any update to p0, so to not override the old state of p0.
// Later, connect to p1 and then remove p1. This will fallback to p0, and
// will send p0's old picker if they are not correctly removed.
@@ -874,16 +896,19 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // p1 subconn should be shut down.
- scToShutdown1 := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown1 != sc11 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc11, scToShutdown1)
+ // p1 will be closed immediately only if the cache is not enabled.
+ if !envconfig.EnablePriorityLBChildPolicyCache {
+ // p1 SubConns are shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when
+ // they are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc11 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc11, scToShutdown)
+ }
}
// Test pick return NoSubConn.
@@ -1004,8 +1029,19 @@
// Test the case where the first and only priority is removed.
func (s) TestPriority_FirstPriorityUnavailable(t *testing.T) {
- const testPriorityInitTimeout = 200 * time.Millisecond
- overrideInitTimeout(t, testPriorityInitTimeout)
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityFirstPriorityUnavailable(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityFirstPriorityUnavailable(t)
+ })
+}
+
+func testPriorityFirstPriorityUnavailable(t *testing.T) {
+ ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+ defer cancel()
cc := testutils.NewBalancerClientConn(t)
bb := balancer.Get(Name)
@@ -1029,6 +1065,12 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
+ addrs0 := <-cc.NewSubConnAddrsCh
+ if got, want := addrs0[0].Addr, testBackendAddrStrs[0]; got != want {
+ t.Fatalf("sc is created with addr %v, want %v", got, want)
+ }
+ sc0 := <-cc.NewSubConnCh
+
// Remove the only localities.
if err := pb.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{
@@ -1042,14 +1084,42 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // Wait after double the init timer timeout, to ensure it doesn't panic.
- time.Sleep(testPriorityInitTimeout * 2)
+ // p0 will be closed immediately only if the cache is not enabled.
+ if !envconfig.EnablePriorityLBChildPolicyCache {
+ // p0 SubConns are shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when
+ // they are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc0 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
+ }
+ }
+
+ // Test pick return TransientFailure.
+ if err := cc.WaitForPickerWithErr(ctx, ErrAllPrioritiesRemoved); err != nil {
+ t.Fatal(err.Error())
+ }
}
// When a child is moved from low priority to high.
//
// Init a(p0) and b(p1); a(p0) is up, use a; move b to p0, a to p1, use b.
func (s) TestPriority_MoveChildToHigherPriority(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityMoveChildToHigherPriority(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityMoveChildToHigherPriority(t)
+ })
+}
+
+func testPriorityMoveChildToHigherPriority(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -1120,18 +1190,6 @@
t.Fatal(err.Error())
}
- // Old subconn should be shut down.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc1 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
- }
-
addrs2 := <-cc.NewSubConnAddrsCh
if got, want := addrs2[0].Addr, testBackendAddrStrs[1]; got != want {
t.Fatalf("sc is created with addr %v, want %v", got, want)
@@ -1153,6 +1211,17 @@
//
// Init a(p0) and b(p1); a(p0) is down, use b; move b to p0, a to p1, use b.
func (s) TestPriority_MoveReadyChildToHigherPriority(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityMoveReadyChildToHigherPriority(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityMoveReadyChildToHigherPriority(t)
+ })
+}
+
+func testPriorityMoveReadyChildToHigherPriority(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -1227,18 +1296,6 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // Old subconn from child-0 should be removed.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc0 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc0, scToShutdown)
- }
-
// Because this was a ready child moved to a higher priority, no new subconn
// or picker should be updated.
select {
@@ -1255,6 +1312,17 @@
//
// Init a(p0) and b(p1); a(p0) is down, use b; move b to p0, a to p1, use b.
func (s) TestPriority_RemoveReadyLowestChild(t *testing.T) {
+ t.Run("caching_enabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
+ testPriorityRemoveReadyLowestChild(t)
+ })
+ t.Run("caching_disabled", func(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ testPriorityRemoveReadyLowestChild(t)
+ })
+}
+
+func testPriorityRemoveReadyLowestChild(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
@@ -1329,16 +1397,18 @@
t.Fatalf("failed to update ClientConn state: %v", err)
}
- // Old subconn from child-1 should be shut down.
- scToShutdown := <-cc.ShutdownSubConnCh
- // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
- // are closed. Remove duplicate events.
- // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
- // workaround once pickfirst is the only leaf policy and responsible for
- // shutting down SubConns.
- <-cc.ShutdownSubConnCh
- if scToShutdown != sc1 {
- t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+ if !envconfig.EnablePriorityLBChildPolicyCache {
+ // Old subconn from child-1 should be shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when they
+ // are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc1 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+ }
}
if err := cc.WaitForErrPicker(ctx); err != nil {
@@ -1358,19 +1428,11 @@
//
// Init 0; 0 is up, use 0; remove 0, only picker is updated, no subconn is
// removed; re-add 0, picker is updated.
-func (s) TestPriority_ReadyChildRemovedButInCache(t *testing.T) {
+func (s) TestPriority_RemoveOnlyReadyChild_CachingEnabled(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, true)
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
- const testChildCacheTimeout = time.Second
- defer func() func() {
- old := DefaultSubBalancerCloseTimeout
- DefaultSubBalancerCloseTimeout = testChildCacheTimeout
- return func() {
- DefaultSubBalancerCloseTimeout = old
- }
- }()()
-
cc := testutils.NewBalancerClientConn(t)
bb := balancer.Get(Name)
pb := bb.Build(cc, balancer.BuildOptions{})
@@ -1408,8 +1470,7 @@
t.Fatal(err.Error())
}
- // Remove the child, it shouldn't cause any conn changed, but picker should
- // be different.
+ // Remove the child.
if err := pb.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{},
BalancerConfig: &LBConfig{},
@@ -1430,7 +1491,7 @@
case <-time.After(time.Millisecond * 100):
}
- // Re-add the child, shouldn't create new connections.
+ // Re-add the child.
if err := pb.UpdateClientConnState(balancer.ClientConnState{
ResolverState: resolver.State{
Endpoints: []resolver.Endpoint{
@@ -1463,6 +1524,109 @@
}
}
+// When the only ready child is removed, it is deleted and subchannels are shut
+// down. Re-adding results in subchannels being recreated.
+//
+// Init 0; 0 is up, use 0; remove 0, re-add 0.
+func (s) TestPriority_RemoveOnlyReadyChild_CachingDisabled(t *testing.T) {
+ testutils.SetEnvConfig(t, &envconfig.EnablePriorityLBChildPolicyCache, false)
+ ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+ defer cancel()
+
+ cc := testutils.NewBalancerClientConn(t)
+ bb := balancer.Get(Name)
+ pb := bb.Build(cc, balancer.BuildOptions{})
+ defer pb.Close()
+
+ // One children, with priorities [0], with one backend.
+ if err := pb.UpdateClientConnState(balancer.ClientConnState{
+ ResolverState: resolver.State{
+ Endpoints: []resolver.Endpoint{
+ hierarchy.SetInEndpoint(resolver.Endpoint{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}}, []string{"child-0"}),
+ },
+ },
+ BalancerConfig: &LBConfig{
+ Children: map[string]*Child{
+ "child-0": {Config: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name}},
+ },
+ Priorities: []string{"child-0"},
+ },
+ }); err != nil {
+ t.Fatalf("failed to update ClientConn state: %v", err)
+ }
+
+ addrs1 := <-cc.NewSubConnAddrsCh
+ if got, want := addrs1[0].Addr, testBackendAddrStrs[0]; got != want {
+ t.Fatalf("sc is created with addr %v, want %v", got, want)
+ }
+ sc1 := <-cc.NewSubConnCh
+
+ // p0 is ready.
+ sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+ sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+
+ // Test roundrobin with only p0 subconns.
+ if err := cc.WaitForRoundRobinPicker(ctx, sc1); err != nil {
+ t.Fatal(err.Error())
+ }
+
+ // Remove the child.
+ if err := pb.UpdateClientConnState(balancer.ClientConnState{
+ ResolverState: resolver.State{},
+ BalancerConfig: &LBConfig{},
+ }); err != nil {
+ t.Fatalf("failed to update ClientConn state: %v", err)
+ }
+
+ // p0 SubConns are shut down.
+ scToShutdown := <-cc.ShutdownSubConnCh
+ // The same SubConn is closed by gracefulswitch and pickfirstleaf when
+ // they are closed. Remove duplicate events.
+ // TODO: https://github.com/grpc/grpc-go/issues/6472 - Remove this
+ // workaround once pickfirst is the only leaf policy and responsible for
+ // shutting down SubConns.
+ <-cc.ShutdownSubConnCh
+ if scToShutdown != sc1 {
+ t.Fatalf("ShutdownSubConn, want %v, got %v", sc1, scToShutdown)
+ }
+
+ if err := cc.WaitForPickerWithErr(ctx, ErrAllPrioritiesRemoved); err != nil {
+ t.Fatal(err.Error())
+ }
+
+ // Re-add the child.
+ if err := pb.UpdateClientConnState(balancer.ClientConnState{
+ ResolverState: resolver.State{
+ Endpoints: []resolver.Endpoint{
+ hierarchy.SetInEndpoint(resolver.Endpoint{Addresses: []resolver.Address{{Addr: testBackendAddrStrs[0]}}}, []string{"child-0"}),
+ },
+ },
+ BalancerConfig: &LBConfig{
+ Children: map[string]*Child{
+ "child-0": {Config: &internalserviceconfig.BalancerConfig{Name: roundrobin.Name}},
+ },
+ Priorities: []string{"child-0"},
+ },
+ }); err != nil {
+ t.Fatalf("failed to update ClientConn state: %v", err)
+ }
+
+ addrs1 = <-cc.NewSubConnAddrsCh
+ if got, want := addrs1[0].Addr, testBackendAddrStrs[0]; got != want {
+ t.Fatalf("sc is created with addr %v, want %v", got, want)
+ }
+ sc1 = <-cc.NewSubConnCh
+
+ // p0 is ready.
+ sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Connecting})
+ sc1.UpdateState(balancer.SubConnState{ConnectivityState: connectivity.Ready})
+
+ // Test roundrobin with only p0 subconns.
+ if err := cc.WaitForRoundRobinPicker(ctx, sc1); err != nil {
+ t.Fatal(err.Error())
+ }
+}
+
// When the policy of a child is changed.
//
// Init 0; 0 is up, use 0; change 0's policy, 0 is used.
@@ -1691,7 +1855,6 @@
case <-time.After(time.Second):
t.Fatalf("timeout waiting for ResolveNow()")
}
-
}
// TestPriority_IgnoreReresolutionRequestTwoChildren tests the case where the
@@ -1791,7 +1954,7 @@
}
}
-const initIdleBalancerName = "test-init-Idle-balancer"
+const initIdleBalancerName = "test-init-idle-balancer"
var errsTestInitIdle = []error{
fmt.Errorf("init Idle balancer error 0"),
