Merge commit 'd66fc3a1efa1dfb33dfedf9760528f1ac2b923b6' into HEAD

Author: arjan-bal

.github/workflows/deps.yml

@@ -0,0 +1,43 @@
+name: Dependency Changes
+
+# Trigger on PRs.
+on:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  # Compare dependencies before and after this PR.
+  dependencies:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    strategy:
+      fail-fast: true
+
+    steps:
+      - name: Checkout repo
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Setup Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: stable
+          cache-dependency-path: "**/*go.sum"
+
+      # Run the commands to generate dependencies before and after and compare.
+      - name: Compare dependencies
+        run: |
+          BEFORE="$(mktemp -d)"
+          AFTER="$(mktemp -d)"
+
+          scripts/gen-deps.sh "${AFTER}"
+          git checkout origin/master
+          scripts/gen-deps.sh "${BEFORE}"
+
+          echo "Comparing dependencies..."
+          # Run grep in a sub-shell since bash does not support ! in the middle of a pipe
+          diff -u0 -r "${BEFORE}" "${AFTER}" | bash -c '! grep -v "@@"'
+          echo "No changes detected."

.github/workflows/testing.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v5
         with:
-          go-version: '1.22'
+          go-version: '1.23'
           cache-dependency-path: "**/go.sum"
 
       # Run the vet-proto checks.
@@ -46,32 +46,28 @@ jobs:
             goversion: '1.22'
 
           - type: extras
-            goversion: '1.22'
+            goversion: '1.23'
 
           - type: tests
-            goversion: '1.22'
+            goversion: '1.23'
 
           - type: tests
-            goversion: '1.22'
+            goversion: '1.23'
             testflags: -race
 
           - type: tests
-            goversion: '1.22'
-            testflags: '-race -tags=buffer_pooling'
-
-          - type: tests
-            goversion: '1.22'
+            goversion: '1.23'
             goarch: 386
 
           - type: tests
-            goversion: '1.22'
+            goversion: '1.23'
             goarch: arm64
 
           - type: tests
-            goversion: '1.21'
+            goversion: '1.22'
 
           - type: tests
-            goversion: '1.22'
+            goversion: '1.23'
             testflags: -race
             grpcenv: 'GRPC_EXPERIMENTAL_ENABLE_NEW_PICK_FIRST=true'
 

balancer/endpointsharding/endpointsharding.go

@@ -28,11 +28,11 @@ package endpointsharding
 import (
 	"encoding/json"
 	"errors"
-	"fmt"
-	"math/rand"
 	"sync"
 	"sync/atomic"
 
+	rand "math/rand/v2"
+
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/balancer/base"
 	"google.golang.org/grpc/connectivity"
@@ -66,7 +66,9 @@ type endpointSharding struct {
 	cc    balancer.ClientConn
 	bOpts balancer.BuildOptions
 
+	childMu  sync.Mutex // syncs balancer.Balancer calls into children
 	children atomic.Pointer[resolver.EndpointMap]
+	closed   bool
 
 	// inhibitChildUpdates is set during UpdateClientConnState/ResolverError
 	// calls (calls to children will each produce an update, only want one
@@ -80,19 +82,11 @@ type endpointSharding struct {
 // for endpoints that are no longer present. It also updates all the children,
 // and sends a single synchronous update of the childrens' aggregated state at
 // the end of the UpdateClientConnState operation. If any endpoint has no
-// addresses, returns error without forwarding any updates. Otherwise returns
-// first error found from a child, but fully processes the new update.
+// addresses it will ignore that endpoint. Otherwise, returns first error found
+// from a child, but fully processes the new update.
 func (es *endpointSharding) UpdateClientConnState(state balancer.ClientConnState) error {
-	if len(state.ResolverState.Endpoints) == 0 {
-		return errors.New("endpoints list is empty")
-	}
-	// Check/return early if any endpoints have no addresses.
-	// TODO: make this configurable if needed.
-	for i, endpoint := range state.ResolverState.Endpoints {
-		if len(endpoint.Addresses) == 0 {
-			return fmt.Errorf("endpoint %d has empty addresses", i)
-		}
-	}
+	es.childMu.Lock()
+	defer es.childMu.Unlock()
 
 	es.inhibitChildUpdates.Store(true)
 	defer func() {
@@ -106,6 +100,9 @@ func (es *endpointSharding) UpdateClientConnState(state balancer.ClientConnState
 
 	// Update/Create new children.
 	for _, endpoint := range state.ResolverState.Endpoints {
+		if len(endpoint.Addresses) == 0 {
+			continue
+		}
 		if _, ok := newChildren.Get(endpoint); ok {
 			// Endpoint child was already created, continue to avoid duplicate
 			// update.
@@ -153,6 +150,8 @@ func (es *endpointSharding) UpdateClientConnState(state balancer.ClientConnState
 // children and sends a single synchronous update of the childStates at the end
 // of the ResolverError operation.
 func (es *endpointSharding) ResolverError(err error) {
+	es.childMu.Lock()
+	defer es.childMu.Unlock()
 	es.inhibitChildUpdates.Store(true)
 	defer func() {
 		es.inhibitChildUpdates.Store(false)
@@ -170,11 +169,14 @@ func (es *endpointSharding) UpdateSubConnState(balancer.SubConn, balancer.SubCon
 }
 
 func (es *endpointSharding) Close() {
+	es.childMu.Lock()
+	defer es.childMu.Unlock()
 	children := es.children.Load()
 	for _, child := range children.Values() {
 		bal := child.(balancer.Balancer)
 		bal.Close()
 	}
+	es.closed = true
 }
 
 // updateState updates this component's state. It sends the aggregated state,
@@ -234,7 +236,7 @@ func (es *endpointSharding) updateState() {
 	p := &pickerWithChildStates{
 		pickers:     pickers,
 		childStates: childStates,
-		next:        uint32(rand.Intn(len(pickers))),
+		next:        uint32(rand.IntN(len(pickers))),
 	}
 	es.cc.UpdateState(balancer.State{
 		ConnectivityState: aggState,
@@ -282,6 +284,17 @@ func (bw *balancerWrapper) UpdateState(state balancer.State) {
 	bw.es.mu.Lock()
 	bw.childState.State = state
 	bw.es.mu.Unlock()
+	// When a child balancer says it's IDLE, ping it to exit idle and reconnect.
+	// TODO: In the future, perhaps make this a knob in configuration.
+	if ei, ok := bw.Balancer.(balancer.ExitIdler); state.ConnectivityState == connectivity.Idle && ok {
+		go func() {
+			bw.es.childMu.Lock()
+			if !bw.es.closed {
+				ei.ExitIdle()
+			}
+			bw.es.childMu.Unlock()
+		}()
+	}
 	bw.es.updateState()
 }
 

balancer/grpclb/grpclb_picker.go

@@ -19,7 +19,7 @@
 package grpclb
 
 import (
-	"math/rand"
+	rand "math/rand/v2"
 	"sync"
 	"sync/atomic"
 
@@ -112,7 +112,7 @@ type rrPicker struct {
 func newRRPicker(readySCs []balancer.SubConn) *rrPicker {
 	return &rrPicker{
 		subConns:     readySCs,
-		subConnsNext: rand.Intn(len(readySCs)),
+		subConnsNext: rand.IntN(len(readySCs)),
 	}
 }
 
@@ -147,7 +147,7 @@ func newLBPicker(serverList []*lbpb.Server, readySCs []balancer.SubConn, stats *
 	return &lbPicker{
 		serverList:   serverList,
 		subConns:     readySCs,
-		subConnsNext: rand.Intn(len(readySCs)),
+		subConnsNext: rand.IntN(len(readySCs)),
 		stats:        stats,
 	}
 }

balancer/leastrequest/leastrequest.go

@@ -22,7 +22,7 @@ package leastrequest
 import (
 	"encoding/json"
 	"fmt"
-	"math/rand"
+	rand "math/rand/v2"
 	"sync/atomic"
 
 	"google.golang.org/grpc/balancer"

balancer/pickfirst/pickfirst.go

@@ -23,7 +23,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"math/rand"
+	rand "math/rand/v2"
 
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/balancer/pickfirst/internal"

balancer/rls/config.go

@@ -143,7 +143,10 @@ type lbConfigJSON struct {
 //	- childPolicyConfigTargetFieldName:
 //	  - must be set and non-empty
 func (rlsBB) ParseConfig(c json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
-	logger.Infof("Received JSON service config: %v", pretty.ToJSON(c))
+	if logger.V(2) {
+		logger.Infof("Received JSON service config: %v", pretty.ToJSON(c))
+	}
+
 	cfgJSON := &lbConfigJSON{}
 	if err := json.Unmarshal(c, cfgJSON); err != nil {
 		return nil, fmt.Errorf("rls: json unmarshal failed for service config %+v: %v", string(c), err)

balancer/rls/control_channel.go

@@ -209,7 +209,9 @@ func (cc *controlChannel) lookup(reqKeys map[string]string, reason rlspb.RouteLo
 			Reason:          reason,
 			StaleHeaderData: staleHeaders,
 		}
-		cc.logger.Infof("Sending RLS request %+v", pretty.ToJSON(req))
+		if cc.logger.V(2) {
+			cc.logger.Infof("Sending RLS request %+v", pretty.ToJSON(req))
+		}
 
 		ctx, cancel := context.WithTimeout(context.Background(), cc.rpcTimeout)
 		defer cancel()

balancer/rls/internal/adaptive/adaptive.go

@@ -20,15 +20,15 @@
 package adaptive
 
 import (
-	"math/rand"
+	rand "math/rand/v2"
 	"sync"
 	"time"
 )
 
 // For overriding in unittests.
 var (
-	timeNowFunc = func() time.Time { return time.Now() }
-	randFunc    = func() float64 { return rand.Float64() }
+	timeNowFunc = time.Now
+	randFunc    = rand.Float64
 )
 
 const (