Merge branch 'grpc:master' into subchannel-disconnection-unknown-reason

Author: mbissa

.github/workflows/testing.yml

@@ -27,7 +27,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v5
         with:
-          go-version: '1.25'
+          go-version: '1.26'
           cache-dependency-path: "**/go.sum"
 
       # Run the vet-proto checks.
@@ -45,29 +45,29 @@ jobs:
       matrix:
         include:
           - type: vet
-            goversion: '1.24'
+            goversion: '1.25'
 
           - type: extras
-            goversion: '1.25'
+            goversion: '1.26'
 
           - type: tests
-            goversion: '1.25'
+            goversion: '1.26'
 
           - type: tests
-            goversion: '1.25'
+            goversion: '1.26'
             testflags: -race
 
           - type: tests
-            goversion: '1.25'
+            goversion: '1.26'
             goarch: 386
 
           - type: tests
-            goversion: '1.25'
+            goversion: '1.26'
             goarch: arm64
             runner: ubuntu-24.04-arm
 
           - type: tests
-            goversion: '1.24'
+            goversion: '1.25'
 
     steps:
       # Setup the environment.

balancer/rls/balancer.go

@@ -79,14 +79,14 @@ var (
 	dataCachePurgeHook   = func() {}
 	resetBackoffHook     = func() {}
 
-	cacheEntriesMetric = estats.RegisterInt64Gauge(estats.MetricDescriptor{
+	cacheEntriesMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{
 		Name:        "grpc.lb.rls.cache_entries",
 		Description: "EXPERIMENTAL. Number of entries in the RLS cache.",
 		Unit:        "{entry}",
 		Labels:      []string{"grpc.target", "grpc.lb.rls.server_target", "grpc.lb.rls.instance_uuid"},
 		Default:     false,
 	})
-	cacheSizeMetric = estats.RegisterInt64Gauge(estats.MetricDescriptor{
+	cacheSizeMetric = estats.RegisterInt64AsyncGauge(estats.MetricDescriptor{
 		Name:        "grpc.lb.rls.cache_size",
 		Description: "EXPERIMENTAL. The current size of the RLS cache.",
 		Unit:        "By",
@@ -140,7 +140,9 @@ func (rlsBB) Build(cc balancer.ClientConn, opts balancer.BuildOptions) balancer.
 		updateCh:           buffer.NewUnbounded(),
 	}
 	lb.logger = internalgrpclog.NewPrefixLogger(logger, fmt.Sprintf("[rls-experimental-lb %p] ", lb))
-	lb.dataCache = newDataCache(maxCacheSize, lb.logger, cc.MetricsRecorder(), opts.Target.String())
+	lb.dataCache = newDataCache(maxCacheSize, lb.logger, opts.Target.String())
+	metricsRecorder := cc.MetricsRecorder()
+	lb.unregisterMetricHandler = metricsRecorder.RegisterAsyncReporter(lb, cacheEntriesMetric, cacheSizeMetric)
 	lb.bg = balancergroup.New(balancergroup.Options{
 		CC:                      cc,
 		BuildOpts:               opts,
@@ -162,6 +164,9 @@ type rlsBalancer struct {
 	dataCachePurgeHook func()
 	logger             *internalgrpclog.PrefixLogger
 
+	// unregisterMetricHandler is the function to deregister the async metric reporter.
+	unregisterMetricHandler func()
+
 	// If both cacheMu and stateMu need to be acquired, the former must be
 	// acquired first to prevent a deadlock. This order restriction is due to the
 	// fact that in places where we need to acquire both the locks, we always
@@ -488,6 +493,7 @@ func (b *rlsBalancer) Close() {
 	if b.ctrlCh != nil {
 		b.ctrlCh.close()
 	}
+	b.unregisterMetricHandler()
 	b.bg.Close()
 	b.stateMu.Unlock()
 
@@ -702,3 +708,23 @@ func (b *rlsBalancer) releaseChildPolicyReferences(targets []string) {
 	}
 	b.stateMu.Unlock()
 }
+
+// Report reports the metrics data to the provided recorder.
+func (b *rlsBalancer) Report(r estats.AsyncMetricsRecorder) error {
+	b.cacheMu.Lock()
+	currentSize := b.dataCache.currentSize
+	entriesLen := int64(len(b.dataCache.entries))
+	rlsServerTarget := b.dataCache.rlsServerTarget
+	grpcTarget := b.dataCache.grpcTarget
+	uuid := b.dataCache.uuid
+	shutdown := b.dataCache.shutdown.HasFired()
+	b.cacheMu.Unlock()
+
+	if shutdown {
+		return nil
+	}
+
+	cacheSizeMetric.Record(r, currentSize, grpcTarget, rlsServerTarget, uuid)
+	cacheEntriesMetric.Record(r, entriesLen, grpcTarget, rlsServerTarget, uuid)
+	return nil
+}

balancer/rls/cache.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	estats "google.golang.org/grpc/experimental/stats"
 	"google.golang.org/grpc/internal/backoff"
 	internalgrpclog "google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcsync"
@@ -174,21 +173,19 @@ type dataCache struct {
 	rlsServerTarget string
 
 	// Read only after initialization.
-	grpcTarget      string
-	uuid            string
-	metricsRecorder estats.MetricsRecorder
+	grpcTarget string
+	uuid       string
 }
 
-func newDataCache(size int64, logger *internalgrpclog.PrefixLogger, metricsRecorder estats.MetricsRecorder, grpcTarget string) *dataCache {
+func newDataCache(size int64, logger *internalgrpclog.PrefixLogger, grpcTarget string) *dataCache {
 	return &dataCache{
-		maxSize:         size,
-		keys:            newLRU(),
-		entries:         make(map[cacheKey]*cacheEntry),
-		logger:          logger,
-		shutdown:        grpcsync.NewEvent(),
-		grpcTarget:      grpcTarget,
-		uuid:            uuid.New().String(),
-		metricsRecorder: metricsRecorder,
+		maxSize:    size,
+		keys:       newLRU(),
+		entries:    make(map[cacheKey]*cacheEntry),
+		logger:     logger,
+		shutdown:   grpcsync.NewEvent(),
+		grpcTarget: grpcTarget,
+		uuid:       uuid.New().String(),
 	}
 }
 
@@ -327,8 +324,7 @@ func (dc *dataCache) addEntry(key cacheKey, entry *cacheEntry) (backoffCancelled
 	if dc.currentSize > dc.maxSize {
 		backoffCancelled = dc.resize(dc.maxSize)
 	}
-	cacheSizeMetric.Record(dc.metricsRecorder, dc.currentSize, dc.grpcTarget, dc.rlsServerTarget, dc.uuid)
-	cacheEntriesMetric.Record(dc.metricsRecorder, int64(len(dc.entries)), dc.grpcTarget, dc.rlsServerTarget, dc.uuid)
+
 	return backoffCancelled, true
 }
 
@@ -338,7 +334,7 @@ func (dc *dataCache) updateEntrySize(entry *cacheEntry, newSize int64) {
 	dc.currentSize -= entry.size
 	entry.size = newSize
 	dc.currentSize += entry.size
-	cacheSizeMetric.Record(dc.metricsRecorder, dc.currentSize, dc.grpcTarget, dc.rlsServerTarget, dc.uuid)
+
 }
 
 func (dc *dataCache) getEntry(key cacheKey) *cacheEntry {
@@ -371,8 +367,7 @@ func (dc *dataCache) deleteAndCleanup(key cacheKey, entry *cacheEntry) {
 	delete(dc.entries, key)
 	dc.currentSize -= entry.size
 	dc.keys.removeEntry(key)
-	cacheSizeMetric.Record(dc.metricsRecorder, dc.currentSize, dc.grpcTarget, dc.rlsServerTarget, dc.uuid)
-	cacheEntriesMetric.Record(dc.metricsRecorder, int64(len(dc.entries)), dc.grpcTarget, dc.rlsServerTarget, dc.uuid)
+
 }
 
 func (dc *dataCache) stop() {

balancer/rls/cache_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/grpc/internal/backoff"
-	"google.golang.org/grpc/internal/testutils/stats"
 )
 
 var (
@@ -120,7 +119,7 @@ func (s) TestLRU_BasicOperations(t *testing.T) {
 
 func (s) TestDataCache_BasicOperations(t *testing.T) {
 	initCacheEntries()
-	dc := newDataCache(5, nil, &stats.NoopMetricsRecorder{}, "")
+	dc := newDataCache(5, nil, "")
 	for i, k := range cacheKeys {
 		dc.addEntry(k, cacheEntries[i])
 	}
@@ -134,7 +133,7 @@ func (s) TestDataCache_BasicOperations(t *testing.T) {
 
 func (s) TestDataCache_AddForcesResize(t *testing.T) {
 	initCacheEntries()
-	dc := newDataCache(1, nil, &stats.NoopMetricsRecorder{}, "")
+	dc := newDataCache(1, nil, "")
 
 	// The first entry in cacheEntries has a minimum expiry time in the future.
 	// This entry would stop the resize operation since we do not evict entries
@@ -163,7 +162,7 @@ func (s) TestDataCache_AddForcesResize(t *testing.T) {
 
 func (s) TestDataCache_Resize(t *testing.T) {
 	initCacheEntries()
-	dc := newDataCache(5, nil, &stats.NoopMetricsRecorder{}, "")
+	dc := newDataCache(5, nil, "")
 	for i, k := range cacheKeys {
 		dc.addEntry(k, cacheEntries[i])
 	}
@@ -194,7 +193,7 @@ func (s) TestDataCache_Resize(t *testing.T) {
 
 func (s) TestDataCache_EvictExpiredEntries(t *testing.T) {
 	initCacheEntries()
-	dc := newDataCache(5, nil, &stats.NoopMetricsRecorder{}, "")
+	dc := newDataCache(5, nil, "")
 	for i, k := range cacheKeys {
 		dc.addEntry(k, cacheEntries[i])
 	}
@@ -221,7 +220,7 @@ func (s) TestDataCache_ResetBackoffState(t *testing.T) {
 	}
 
 	initCacheEntries()
-	dc := newDataCache(5, nil, &stats.NoopMetricsRecorder{}, "")
+	dc := newDataCache(5, nil, "")
 	for i, k := range cacheKeys {
 		dc.addEntry(k, cacheEntries[i])
 	}
@@ -242,61 +241,3 @@ func (s) TestDataCache_ResetBackoffState(t *testing.T) {
 		t.Fatalf("unexpected diff in backoffState for cache entry after dataCache.resetBackoffState(): %s", diff)
 	}
 }
-
-func (s) TestDataCache_Metrics(t *testing.T) {
-	cacheEntriesMetricsTests := []*cacheEntry{
-		{size: 1},
-		{size: 2},
-		{size: 3},
-		{size: 4},
-		{size: 5},
-	}
-	tmr := stats.NewTestMetricsRecorder()
-	dc := newDataCache(50, nil, tmr, "")
-
-	dc.updateRLSServerTarget("rls-server-target")
-	for i, k := range cacheKeys {
-		dc.addEntry(k, cacheEntriesMetricsTests[i])
-	}
-
-	const cacheEntriesKey = "grpc.lb.rls.cache_entries"
-	const cacheSizeKey = "grpc.lb.rls.cache_size"
-	// 5 total entries which add up to 15 size, so should record that.
-	if got, _ := tmr.Metric(cacheEntriesKey); got != 5 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheEntriesKey, got, 5)
-	}
-	if got, _ := tmr.Metric(cacheSizeKey); got != 15 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheSizeKey, got, 15)
-	}
-
-	// Resize down the cache to 2 entries (deterministic as based of LRU).
-	dc.resize(9)
-	if got, _ := tmr.Metric(cacheEntriesKey); got != 2 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheEntriesKey, got, 2)
-	}
-	if got, _ := tmr.Metric(cacheSizeKey); got != 9 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheSizeKey, got, 9)
-	}
-
-	// Update an entry to have size 6. This should reflect in the size metrics,
-	// which will increase by 1 to 11, while the number of cache entries should
-	// stay same. This write is deterministic and writes to the last one.
-	dc.updateEntrySize(cacheEntriesMetricsTests[4], 6)
-
-	if got, _ := tmr.Metric(cacheEntriesKey); got != 2 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheEntriesKey, got, 2)
-	}
-	if got, _ := tmr.Metric(cacheSizeKey); got != 10 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheSizeKey, got, 10)
-	}
-
-	// Delete this scaled up cache key. This should scale down the cache to 1
-	// entries, and remove 6 size so cache size should be 4.
-	dc.deleteAndCleanup(cacheKeys[4], cacheEntriesMetricsTests[4])
-	if got, _ := tmr.Metric(cacheEntriesKey); got != 1 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheEntriesKey, got, 1)
-	}
-	if got, _ := tmr.Metric(cacheSizeKey); got != 4 {
-		t.Fatalf("Unexpected data for metric %v, got: %v, want: %v", cacheSizeKey, got, 4)
-	}
-}

benchmark/benchmain/main.go

@@ -117,7 +117,7 @@ var (
 	sleepBetweenRPCs      = flags.DurationSlice("sleepBetweenRPCs", []time.Duration{0}, "Configures the maximum amount of time the client should sleep between consecutive RPCs - may be a comma-separated list")
 	connections           = flag.Int("connections", 1, "The number of connections. Each connection will handle maxConcurrentCalls RPC streams")
 	recvBufferPool        = flags.StringWithAllowedValues("recvBufferPool", recvBufferPoolSimple, "Configures the shared receive buffer pool. One of: nil, simple, all", allRecvBufferPools)
-	sharedWriteBuffer     = flags.StringWithAllowedValues("sharedWriteBuffer", toggleModeOff,
+	sharedWriteBuffer     = flags.StringWithAllowedValues("sharedWriteBuffer", toggleModeOn,
 		fmt.Sprintf("Configures both client and server to share write buffer - One of: %v", strings.Join(allToggleModes, ", ")), allToggleModes)
 
 	logger = grpclog.Component("benchmark")

cmd/protoc-gen-go-grpc/go.mod

@@ -1,6 +1,6 @@
 module google.golang.org/grpc/cmd/protoc-gen-go-grpc
 
-go 1.24.0
+go 1.25.0
 
 require (
 	google.golang.org/grpc v1.70.0

credentials/xds/xds.go

@@ -25,7 +25,6 @@ import (
 	"crypto/tls"
 	"errors"
 	"net"
-	"sync/atomic"
 	"time"
 
 	"google.golang.org/grpc/credentials"
@@ -114,8 +113,10 @@ func (c *credsImpl) ClientHandshake(ctx context.Context, authority string, rawCo
 		return c.fallback.ClientHandshake(ctx, authority, rawConn)
 	}
 
-	uPtr := xdsinternal.GetHandshakeInfo(chi.Attributes)
-	hi := (*xdsinternal.HandshakeInfo)(atomic.LoadPointer(uPtr))
+	hi := xdsinternal.HandshakeInfoFromAttributes(chi.Attributes).Load()
+	if hi == nil {
+		return c.fallback.ClientHandshake(ctx, authority, rawConn)
+	}
 	if hi.UseFallbackCreds() {
 		return c.fallback.ClientHandshake(ctx, authority, rawConn)
 	}

credentials/xds/xds_client_test.go

@@ -30,7 +30,6 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
-	"unsafe"
 
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/tls/certprovider"
@@ -228,9 +227,10 @@ func newTestContextWithHandshakeInfo(parent context.Context, root, identity cert
 	if sanExactMatch != "" {
 		sms = []matcher.StringMatcher{matcher.NewExactStringMatcher(sanExactMatch, false)}
 	}
+	var hiPtr atomic.Pointer[xdsinternal.HandshakeInfo]
 	info := xdsinternal.NewHandshakeInfo(root, identity, sms, false)
-	uPtr := unsafe.Pointer(info)
-	addr := xdsinternal.SetHandshakeInfo(resolver.Address{}, &uPtr)
+	hiPtr.Store(info)
+	addr := xdsinternal.SetHandshakeInfo(resolver.Address{}, &hiPtr)
 
 	// Moving the attributes from the resolver.Address to the context passed to
 	// the handshaker is done in the transport layer. Since we directly call the
@@ -546,8 +546,9 @@ func (s) TestClientCredsProviderSwitch(t *testing.T) {
 	// We need to repeat most of what newTestContextWithHandshakeInfo() does
 	// here because we need access to the underlying HandshakeInfo so that we
 	// can update it before the next call to ClientHandshake().
-	uPtr := unsafe.Pointer(handshakeInfo)
-	addr := xdsinternal.SetHandshakeInfo(resolver.Address{}, &uPtr)
+	var hiPtr atomic.Pointer[xdsinternal.HandshakeInfo]
+	hiPtr.Store(handshakeInfo)
+	addr := xdsinternal.SetHandshakeInfo(resolver.Address{}, &hiPtr)
 	ctx = icredentials.NewClientHandshakeInfoContext(ctx, credentials.ClientHandshakeInfo{Attributes: addr.Attributes})
 	if _, _, err := creds.ClientHandshake(ctx, authority, conn); err == nil {
 		t.Fatal("ClientHandshake() succeeded when expected to fail")
@@ -571,7 +572,7 @@ func (s) TestClientCredsProviderSwitch(t *testing.T) {
 	handshakeInfo = xdsinternal.NewHandshakeInfo(root2, nil, []matcher.StringMatcher{matcher.NewExactStringMatcher(defaultTestCertSAN, false)}, false)
 	// Update the existing pointer, which address attribute will continue to
 	// point to.
-	atomic.StorePointer(&uPtr, unsafe.Pointer(handshakeInfo))
+	hiPtr.Store(handshakeInfo)
 	_, ai, err := creds.ClientHandshake(ctx, authority, conn)
 	if err != nil {
 		t.Fatalf("ClientHandshake() returned failed: %q", err)

dialoptions.go

@@ -705,10 +705,11 @@ func WithDisableHealthCheck() DialOption {
 func defaultDialOptions() dialOptions {
 	return dialOptions{
 		copts: transport.ConnectOptions{
-			ReadBufferSize:  defaultReadBufSize,
-			WriteBufferSize: defaultWriteBufSize,
-			UserAgent:       grpcUA,
-			BufferPool:      mem.DefaultBufferPool(),
+			ReadBufferSize:    defaultReadBufSize,
+			WriteBufferSize:   defaultWriteBufSize,
+			SharedWriteBuffer: true,
+			UserAgent:         grpcUA,
+			BufferPool:        mem.DefaultBufferPool(),
 		},
 		bs:                       internalbackoff.DefaultExponential,
 		idleTimeout:              30 * time.Minute,