Merge remote-tracking branch 'origin/master' into ignore-status-for-grpc-content-type-8486

Author: mbissa

.github/pull_request_template.md

@@ -1,4 +1,4 @@
-Thank you for your PR.  Please read and follow
+Thank you for your PR. Please read and follow
 https://github.com/grpc/grpc-go/blob/master/CONTRIBUTING.md, especially the
 "Guidelines for Pull Requests" section, and then delete this text before
 entering your PR description.

internal/envconfig/xds.go

@@ -68,4 +68,10 @@ var (
 	// trust.  For more details, see:
 	// https://github.com/grpc/proposal/blob/master/A87-mtls-spiffe-support.md
 	XDSSPIFFEEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_MTLS_SPIFFE", false)
+
+	// XDSHTTPConnectEnabled is true if gRPC should parse custom Metadata
+	// configuring use of an HTTP CONNECT proxy via xDS from cluster resources.
+	// For more details, see:
+	// https://github.com/grpc/proposal/blob/master/A86-xds-http-connect.md
+	XDSHTTPConnectEnabled = boolFromEnv("GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT", false)
 )

internal/transport/http2_client.go

@@ -1469,7 +1469,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		rawStatusCode  = codes.Internal
 		// headerError is set if an error is encountered while parsing the headers
 		headerError        string
-		receivedHttpStatus string
+		receivedHTTPStatus string
 	)
 
 	for _, hf := range frame.Fields {
@@ -1496,7 +1496,7 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 			grpcMessage = decodeGrpcMessage(hf.Value)
 		case ":status":
 			if !isGRPC {
-				receivedHttpStatus = hf.Value
+				receivedHTTPStatus = hf.Value
 			}
 		default:
 			if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
@@ -1518,20 +1518,25 @@ func (t *http2Client) operateHeaders(frame *http2.MetaHeadersFrame) {
 		var grpcErrorCode = codes.Internal // when header does not include HTTP status, return INTERNAL
 		var errs []string
 
-		switch receivedHttpStatus {
+		switch receivedHTTPStatus {
 		case "":
 			httpStatusErr = "malformed header: missing HTTP status"
 		case "200":
 			grpcErrorCode = codes.Unknown
 		default:
 			// Any other status code (e.g., "404", "503"). We must parse it.
-			c, err := strconv.ParseInt(receivedHttpStatus, 10, 32)
+			c, err := strconv.ParseInt(receivedHTTPStatus, 10, 32)
 			if err != nil {
 				se := status.New(grpcErrorCode, fmt.Sprintf("transport: malformed http-status: %v", err))
 				t.closeStream(s, se.Err(), true, http2.ErrCodeProtocol, se, nil, endStream)
 				return
 			}
 			statusCode := int(c)
+			if statusCode >= 100 && statusCode < 200 {
+				//In case of informational headers return
+				//For trailers, since we are already in gRPC mode, we will ignore all http statuses and not enter this block
+				return
+			}
 			httpStatusErr = fmt.Sprintf(
 				"unexpected HTTP status code received from server: %d (%s)",
 				statusCode,

internal/transport/http2_server.go

@@ -1353,10 +1353,10 @@ func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCo
 	// called to interrupt the potential blocking on other goroutines.
 	s.cancel()
 
-	oldState := s.swapState(streamDone)
-	if oldState == streamDone {
-		return
-	}
+	// We can't return early even if the stream's state is "done" as the state
+	// might have been set by the `finishStream` method. Deleting the stream via
+	// `finishStream` can get blocked on flow control.
+	s.swapState(streamDone)
 	t.deleteStream(s, eosReceived)
 
 	t.controlBuf.put(&cleanupStream{

internal/transport/transport_test.go

@@ -3183,3 +3183,79 @@ func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) {
 		t.Fatalf("RST frame received earlier than expected by duration: %v", want-got)
 	}
 }
+
+// TestClientTransport_Handle1xxHeaders validates that 1xx HTTP status headers
+// are ignored and treated as a protocol error if END_STREAM is set.
+func (s) TestClientTransport_Handle1xxHeaders(t *testing.T) {
+	testStream := func() *ClientStream {
+		return &ClientStream{
+			Stream: &Stream{
+				buf: &recvBuffer{
+					c:  make(chan recvMsg),
+					mu: sync.Mutex{},
+				},
+			},
+			done:       make(chan struct{}),
+			headerChan: make(chan struct{}),
+		}
+	}
+
+	testClient := func(ts *ClientStream) *http2Client {
+		return &http2Client{
+			mu: sync.Mutex{},
+			activeStreams: map[uint32]*ClientStream{
+				0: ts,
+			},
+			controlBuf: newControlBuffer(make(<-chan struct{})),
+		}
+	}
+
+	for _, test := range []struct {
+		name            string
+		metaHeaderFrame *http2.MetaHeadersFrame
+		httpFlags       http2.Flags
+		wantStatus      *status.Status
+	}{
+		{
+			name: "1xx with END_STREAM will be ignored",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: ":status", Value: "100"},
+				},
+			},
+			httpFlags:  http2.FlagHeadersEndStream,
+			wantStatus: nil,
+		},
+		{
+			name: "1xx without END_STREAM is ignored",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: ":status", Value: "100"},
+				},
+			},
+			httpFlags:  0,
+			wantStatus: nil,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ts := testStream()
+			s := testClient(ts)
+
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
+					Flags:    test.httpFlags,
+				},
+			}
+
+			s.operateHeaders(test.metaHeaderFrame)
+
+			got := ts.status
+			want := test.wantStatus
+
+			if got.Code() != want.Code() || got.Message() != want.Message() {
+				t.Fatalf("operateHeaders(%v); status = %v, want %v", test.metaHeaderFrame, got, want)
+			}
+		})
+	}
+}

internal/xds/balancer/priority/ignore_resolve_now.go

@@ -29,29 +29,21 @@ import (
 // ResolveNow() method to ignore those calls if the ignoreResolveNow bit is set.
 type ignoreResolveNowClientConn struct {
 	balancer.ClientConn
-	ignoreResolveNow *uint32
+	ignoreResolveNow atomic.Bool
 }
 
 func newIgnoreResolveNowClientConn(cc balancer.ClientConn, ignore bool) *ignoreResolveNowClientConn {
-	ret := &ignoreResolveNowClientConn{
-		ClientConn:       cc,
-		ignoreResolveNow: new(uint32),
-	}
+	ret := &ignoreResolveNowClientConn{ClientConn: cc}
 	ret.updateIgnoreResolveNow(ignore)
 	return ret
 }
 
 func (i *ignoreResolveNowClientConn) updateIgnoreResolveNow(b bool) {
-	if b {
-		atomic.StoreUint32(i.ignoreResolveNow, 1)
-		return
-	}
-	atomic.StoreUint32(i.ignoreResolveNow, 0)
-
+	i.ignoreResolveNow.Store(b)
 }
 
-func (i ignoreResolveNowClientConn) ResolveNow(o resolver.ResolveNowOptions) {
-	if atomic.LoadUint32(i.ignoreResolveNow) != 0 {
+func (i *ignoreResolveNowClientConn) ResolveNow(o resolver.ResolveNowOptions) {
+	if i.ignoreResolveNow.Load() {
 		return
 	}
 	i.ClientConn.ResolveNow(o)

internal/xds/clients/xdsclient/internal/internal.go

@@ -21,10 +21,6 @@ package internal
 import "time"
 
 var (
-	// WatchExpiryTimeout is the watch expiry timeout for xDS client. It can be
-	// overridden by tests to change the default watch expiry timeout.
-	WatchExpiryTimeout time.Duration
-
 	// StreamBackoff is the stream backoff for xDS client. It can be overridden
 	// by tests to change the default backoff strategy.
 	StreamBackoff func(int) time.Duration

internal/xds/clients/xdsclient/test/ads_stream_watch_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/google/uuid"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/xds/clients"
 	"google.golang.org/grpc/internal/xds/clients/grpctransport"
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient"
@@ -175,9 +176,34 @@ func (s) TestADS_WatchState_TimerFires(t *testing.T) {
 	// short resource expiry timeout.
 	nodeID := uuid.New().String()
 	configs := map[string]grpctransport.Config{"insecure": {Credentials: insecure.NewBundle()}}
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
-	client := createXDSClient(t, mgmtServer.Address, nodeID, grpctransport.NewBuilder(configs))
-
+	resourceTypes := map[string]xdsclient.ResourceType{xdsresource.V3ListenerURL: listenerType}
+	si := clients.ServerIdentifier{
+		ServerURI:  mgmtServer.Address,
+		Extensions: grpctransport.ServerIdentifierExtension{ConfigName: "insecure"},
+	}
+
+	xdsClientConfig := xdsclient.Config{
+		Servers:          []xdsclient.ServerConfig{{ServerIdentifier: si}},
+		Node:             clients.Node{ID: nodeID, UserAgentName: "user-agent", UserAgentVersion: "0.0.0.0"},
+		TransportBuilder: grpctransport.NewBuilder(configs),
+		ResourceTypes:    resourceTypes,
+		// Xdstp resource names used in this test do not specify an
+		// authority. These will end up looking up an entry with the
+		// empty key in the authorities map. Having an entry with an
+		// empty key and empty configuration, results in these
+		// resources also using the top-level configuration.
+		Authorities: map[string]xdsclient.Authority{
+			"": {XDSServers: []xdsclient.ServerConfig{}},
+		},
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
+	}
+
+	// Create an xDS client with the above config.
+	client, err := xdsclient.New(xdsClientConfig)
+	if err != nil {
+		t.Fatalf("Failed to create xDS client: %v", err)
+	}
+	t.Cleanup(func() { client.Close() })
 	// Create a watch for the first listener resource and verify that the timer
 	// is running and the watch state is `requested`.
 	const listenerName = "listener"

internal/xds/clients/xdsclient/test/authority_test.go

@@ -105,10 +105,10 @@ func setupForAuthorityTests(ctx context.Context, t *testing.T) (*testutils.Liste
 			testAuthority2: {XDSServers: []xdsclient.ServerConfig{}},
 			testAuthority3: {XDSServers: []xdsclient.ServerConfig{{ServerIdentifier: siNonDefault}}},
 		},
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
 	// Create an xDS client with the above config.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)

internal/xds/clients/xdsclient/test/lds_watchers_test.go

@@ -35,7 +35,6 @@ import (
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils"
 	"google.golang.org/grpc/internal/xds/clients/internal/testutils/e2e"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient"
-	xdsclientinternal "google.golang.org/grpc/internal/xds/clients/xdsclient/internal"
 	"google.golang.org/grpc/internal/xds/clients/xdsclient/internal/xdsresource"
 
 	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
@@ -115,12 +114,6 @@ func badListenerResource(t *testing.T, name string) *v3listenerpb.Listener {
 	}
 }
 
-func overrideWatchExpiryTimeout(t *testing.T, watchExpiryTimeout time.Duration) {
-	originalWatchExpiryTimeout := xdsclientinternal.WatchExpiryTimeout
-	xdsclientinternal.WatchExpiryTimeout = watchExpiryTimeout
-	t.Cleanup(func() { xdsclientinternal.WatchExpiryTimeout = originalWatchExpiryTimeout })
-}
-
 // verifyNoListenerUpdate verifies that no listener update is received on the
 // provided update channel, and returns an error if an update is received.
 //
@@ -726,11 +719,10 @@ func (s) TestLDSWatch_ExpiryTimerFiresBeforeResponse(t *testing.T) {
 		Node:             clients.Node{ID: nodeID},
 		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
+		// Override the default watch expiry timeout.
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
-	// Create an xDS client with the above config and override the default
-	// watch expiry timeout.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)
@@ -777,11 +769,11 @@ func (s) TestLDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) {
 		Node:             clients.Node{ID: nodeID},
 		TransportBuilder: grpctransport.NewBuilder(configs),
 		ResourceTypes:    resourceTypes,
+		// Override the default watch expiry timeout.
+		WatchExpiryTimeout: defaultTestWatchExpiryTimeout,
 	}
 
-	// Create an xDS client with the above config and override the default
-	// watch expiry timeout.
-	overrideWatchExpiryTimeout(t, defaultTestWatchExpiryTimeout)
+	// Create an xDS client with the above config.
 	client, err := xdsclient.New(xdsClientConfig)
 	if err != nil {
 		t.Fatalf("Failed to create xDS client: %v", err)