Merge branch 'grpc:master' into benchmark-client-context-cancellation

Author: wdsjk

internal/transport/http2_server.go

@@ -35,14 +35,15 @@ import (
 
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
+	"google.golang.org/protobuf/proto"
+
 	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpclog"
 	"google.golang.org/grpc/internal/grpcutil"
 	"google.golang.org/grpc/internal/pretty"
 	istatus "google.golang.org/grpc/internal/status"
 	"google.golang.org/grpc/internal/syscall"
 	"google.golang.org/grpc/mem"
-	"google.golang.org/protobuf/proto"
 
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -1304,15 +1305,16 @@ func (t *http2Server) Close(err error) {
 // deleteStream deletes the stream s from transport's active streams.
 func (t *http2Server) deleteStream(s *ServerStream, eosReceived bool) {
 	t.mu.Lock()
-	if _, ok := t.activeStreams[s.id]; ok {
+	_, isActive := t.activeStreams[s.id]
+	if isActive {
 		delete(t.activeStreams, s.id)
 		if len(t.activeStreams) == 0 {
 			t.idle = time.Now()
 		}
 	}
 	t.mu.Unlock()
 
-	if channelz.IsOn() {
+	if isActive && channelz.IsOn() {
 		if eosReceived {
 			t.channelz.SocketMetrics.StreamsSucceeded.Add(1)
 		} else {

internal/transport/transport_test.go

@@ -39,9 +39,11 @@ import (
 	"github.com/google/go-cmp/cmp"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/hpack"
+
 	"google.golang.org/grpc/attributes"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/leakcheck"
@@ -3260,3 +3262,125 @@ func (s) TestClientTransport_Handle1xxHeaders(t *testing.T) {
 		})
 	}
 }
+
+func (s) TestDeleteStreamMetricsIncrementedOnlyOnce(t *testing.T) {
+	// Enable channelz for metrics collection
+	defer internal.ChannelzTurnOffForTesting()
+	if !channelz.IsOn() {
+		channelz.TurnOn()
+	}
+
+	for _, test := range []struct {
+		name                string
+		eosReceived         bool
+		wantStreamSucceeded int64
+		wantStreamFailed    int64
+	}{
+		{
+			name:                "StreamsSucceeded",
+			eosReceived:         true,
+			wantStreamSucceeded: 1,
+			wantStreamFailed:    0,
+		},
+		{
+			name:                "StreamsFailed",
+			eosReceived:         false,
+			wantStreamSucceeded: 0,
+			wantStreamFailed:    1,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+
+			// Setup server configuration with channelz support
+			serverConfig := &ServerConfig{
+				ChannelzParent: channelz.RegisterServer(t.Name()),
+			}
+			defer channelz.RemoveEntry(serverConfig.ChannelzParent.ID)
+
+			// Create server and client with normal handler (not notifyCall)
+			server, client, cancel := setUpWithOptions(t, 0, serverConfig, normal, ConnectOptions{})
+			defer func() {
+				client.Close(fmt.Errorf("test cleanup"))
+				server.stop()
+				cancel()
+			}()
+
+			// Wait for connection to be established
+			waitWhileTrue(t, func() (bool, error) {
+				server.mu.Lock()
+				defer server.mu.Unlock()
+				if len(server.conns) == 0 {
+					return true, fmt.Errorf("timed-out while waiting for connection")
+				}
+				return false, nil
+			})
+
+			// Get the server transport
+			server.mu.Lock()
+			var serverTransport *http2Server
+			for st := range server.conns {
+				serverTransport = st.(*http2Server)
+				break
+			}
+			server.mu.Unlock()
+
+			if serverTransport == nil {
+				t.Fatal("Server transport not found")
+			}
+
+			clientStream, err := client.NewStream(ctx, &CallHdr{})
+			if err != nil {
+				t.Fatalf("Failed to create stream: %v", err)
+			}
+
+			// Wait for the stream to be created on the server side
+			var serverStream *ServerStream
+			waitWhileTrue(t, func() (bool, error) {
+				serverTransport.mu.Lock()
+				defer serverTransport.mu.Unlock()
+				for _, v := range serverTransport.activeStreams {
+					if v.id == clientStream.id {
+						serverStream = v
+						return false, nil
+					}
+				}
+				return true, nil
+			})
+
+			if serverStream == nil {
+				t.Fatalf("Server stream not found for client stream ID %d", clientStream.id)
+			}
+
+			// First call to deleteStream should remove the stream from activeStreams and update metrics
+			serverTransport.deleteStream(serverStream, test.eosReceived)
+
+			// Check metrics after first deleteStream call
+			streamsSucceeded := serverTransport.channelz.SocketMetrics.StreamsSucceeded.Load()
+			streamsFailed := serverTransport.channelz.SocketMetrics.StreamsFailed.Load()
+
+			if streamsSucceeded != test.wantStreamSucceeded {
+				t.Errorf("After first deleteStream - StreamsSucceeded: got %d, want %d", streamsSucceeded, test.wantStreamSucceeded)
+			}
+			if streamsFailed != test.wantStreamFailed {
+				t.Errorf("After first deleteStream - StreamsFailed: got %d, want %d", streamsFailed, test.wantStreamFailed)
+			}
+
+			// Additional calls to deleteStream should not change metrics (stream already deleted)
+			serverTransport.deleteStream(serverStream, test.eosReceived)
+			serverTransport.deleteStream(serverStream, test.eosReceived)
+
+			// Verify metrics haven't changed after subsequent calls
+			additionalStreamsSucceeded := serverTransport.channelz.SocketMetrics.StreamsSucceeded.Load()
+			additionalStreamsFailed := serverTransport.channelz.SocketMetrics.StreamsFailed.Load()
+
+			if additionalStreamsSucceeded != test.wantStreamSucceeded {
+				t.Errorf("After multiple deleteStream calls - StreamsSucceeded changed: got %d, want %d", additionalStreamsSucceeded, test.wantStreamSucceeded)
+			}
+			if additionalStreamsFailed != test.wantStreamFailed {
+				t.Errorf("After multiple deleteStream calls - StreamsFailed changed: got %d, want %d", additionalStreamsFailed, test.wantStreamFailed)
+			}
+		})
+	}
+}

internal/xds/balancer/clusterimpl/balancer_test.go

@@ -20,9 +20,7 @@ package clusterimpl
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
-	"strings"
 	"testing"
 	"time"
 
@@ -38,7 +36,6 @@ import (
 	"google.golang.org/grpc/internal/xds/testutils/fakeclient"
 	"google.golang.org/grpc/internal/xds/xdsclient"
 	"google.golang.org/grpc/resolver"
-	"google.golang.org/grpc/serviceconfig"
 )
 
 const (
@@ -70,9 +67,6 @@ func init() {
 // are handled in the run() goroutine, which exits before Close() returns, we
 // expect the above picker update to be dropped.
 func (s) TestPickerUpdateAfterClose(t *testing.T) {
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
 	builder := balancer.Get(Name)
 	cc := testutils.NewBalancerClientConn(t)
 	b := builder.Build(cc, balancer.BuildOptions{})
@@ -107,6 +101,7 @@ func (s) TestPickerUpdateAfterClose(t *testing.T) {
 	})
 
 	var maxRequest uint32 = 50
+	xdsC := fakeclient.NewClient()
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
 		BalancerConfig: &LBConfig{
@@ -143,14 +138,12 @@ func (s) TestClusterNameInAddressAttributes(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
 	builder := balancer.Get(Name)
 	cc := testutils.NewBalancerClientConn(t)
 	b := builder.Build(cc, balancer.BuildOptions{})
 	defer b.Close()
 
+	xdsC := fakeclient.NewClient()
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
 		BalancerConfig: &LBConfig{
@@ -212,108 +205,6 @@ func (s) TestClusterNameInAddressAttributes(t *testing.T) {
 	}
 }
 
-// Test verifies that child policies was updated on receipt of
-// configuration update.
-func (s) TestChildPolicyUpdatedOnConfigUpdate(t *testing.T) {
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	// Keep track of which child policy was updated
-	updatedChildPolicy := ""
-
-	// Create stub balancers to track config updates
-	const (
-		childPolicyName1 = "stubBalancer1"
-		childPolicyName2 = "stubBalancer2"
-	)
-
-	stub.Register(childPolicyName1, stub.BalancerFuncs{
-		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
-			updatedChildPolicy = childPolicyName1
-			return nil
-		},
-	})
-
-	stub.Register(childPolicyName2, stub.BalancerFuncs{
-		UpdateClientConnState: func(_ *stub.BalancerData, _ balancer.ClientConnState) error {
-			updatedChildPolicy = childPolicyName2
-			return nil
-		},
-	})
-
-	// Initial config update with childPolicyName1
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName1,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("Error updating the config: %v", err)
-	}
-
-	if updatedChildPolicy != childPolicyName1 {
-		t.Fatal("Child policy 1 was not updated on initial configuration update.")
-	}
-
-	// Second config update with childPolicyName2
-	if err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName2,
-			},
-		},
-	}); err != nil {
-		t.Fatalf("Error updating the config: %v", err)
-	}
-
-	if updatedChildPolicy != childPolicyName2 {
-		t.Fatal("Child policy 2 was not updated after child policy name change.")
-	}
-}
-
-// Test verifies that config update fails if child policy config
-// failed to parse.
-func (s) TestFailedToParseChildPolicyConfig(t *testing.T) {
-	xdsC := fakeclient.NewClient()
-
-	builder := balancer.Get(Name)
-	cc := testutils.NewBalancerClientConn(t)
-	b := builder.Build(cc, balancer.BuildOptions{})
-	defer b.Close()
-
-	// Create a stub balancer which fails to ParseConfig.
-	const parseConfigError = "failed to parse config"
-	const childPolicyName = "stubBalancer-FailedToParseChildPolicyConfig"
-	stub.Register(childPolicyName, stub.BalancerFuncs{
-		ParseConfig: func(_ json.RawMessage) (serviceconfig.LoadBalancingConfig, error) {
-			return nil, errors.New(parseConfigError)
-		},
-	})
-
-	err := b.UpdateClientConnState(balancer.ClientConnState{
-		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
-		BalancerConfig: &LBConfig{
-			Cluster: testClusterName,
-			ChildPolicy: &internalserviceconfig.BalancerConfig{
-				Name: childPolicyName,
-			},
-		},
-	})
-
-	if err == nil || !strings.Contains(err.Error(), parseConfigError) {
-		t.Fatalf("Got error: %v, want error: %s", err, parseConfigError)
-	}
-}
-
 // Test verify that the case picker is updated synchronously on receipt of
 // configuration update.
 func (s) TestPickerUpdatedSynchronouslyOnConfigUpdate(t *testing.T) {
@@ -342,9 +233,6 @@ func (s) TestPickerUpdatedSynchronouslyOnConfigUpdate(t *testing.T) {
 	}
 	defer func() { clientConnUpdateHook = origClientConnUpdateHook }()
 
-	defer xdsclient.ClearCounterForTesting(testClusterName, testServiceName)
-	xdsC := fakeclient.NewClient()
-
 	builder := balancer.Get(Name)
 	cc := testutils.NewBalancerClientConn(t)
 	b := builder.Build(cc, balancer.BuildOptions{})
@@ -362,6 +250,7 @@ func (s) TestPickerUpdatedSynchronouslyOnConfigUpdate(t *testing.T) {
 		},
 	})
 
+	xdsC := fakeclient.NewClient()
 	if err := b.UpdateClientConnState(balancer.ClientConnState{
 		ResolverState: xdsclient.SetClient(resolver.State{Endpoints: testBackendEndpoints}, xdsC),
 		BalancerConfig: &LBConfig{