Merge branch 'master' into add-close-to-unbounded-buff

CONTRIBUTING.md

@@ -20,6 +20,15 @@ How to get your contributions merged smoothly and quickly.
   both author's & review's time is wasted. Create more PRs to address different
   concerns and everyone will be happy.
 
+- If you are searching for features to work on, issues labeled [Status: Help
+  Wanted](https://github.com/grpc/grpc-go/issues?q=is%3Aissue+is%3Aopen+sort%3Aupdated-desc+label%3A%22Status%3A+Help+Wanted%22)
+  is a great place to start. These issues are well-documented and usually can be
+  resolved with a single pull request.
+
+- If you are adding a new file, make sure it has the copyright message template 
+  at the top as a comment. You can copy over the message from an existing file 
+  and update the year.
+
 - The grpc package should only depend on standard Go packages and a small number
   of exceptions. If your contribution introduces new dependencies which are NOT
   in the [list](https://godoc.org/google.golang.org/grpc?imports), you need a
@@ -32,14 +41,18 @@ How to get your contributions merged smoothly and quickly.
 - Provide a good **PR description** as a record of **what** change is being made
   and **why** it was made. Link to a github issue if it exists.
 
-- Don't fix code style and formatting unless you are already changing that line
-  to address an issue. PRs with irrelevant changes won't be merged. If you do
-  want to fix formatting or style, do that in a separate PR.
+- If you want to fix formatting or style, consider whether your changes are an 
+  obvious improvement or might be considered a personal preference. If a style 
+  change is based on preference, it likely will not be accepted. If it corrects 
+  widely agreed-upon anti-patterns, then please do create a PR and explain the 
+  benefits of the change.
 
 - Unless your PR is trivial, you should expect there will be reviewer comments
-  that you'll need to address before merging. We expect you to be reasonably
-  responsive to those comments, otherwise the PR will be closed after 2-3 weeks
-  of inactivity.
+  that you'll need to address before merging. We'll mark it as `Status: Requires
+  Reporter Clarification` if we expect you to respond to these comments in a
+  timely manner. If the PR remains inactive for 6 days, it will be marked as
+  `stale` and automatically close 7 days after that if we don't hear back from
+  you.
 
 - Maintain **clean commit history** and use **meaningful commit messages**. PRs
   with messy commit history are difficult to review and won't be merged. Use

admin/test/utils.go

@@ -36,6 +36,7 @@ import (
 
 	v3statusgrpc "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
 	v3statuspb "github.com/envoyproxy/go-control-plane/envoy/service/status/v3"
+	channelzgrpc "google.golang.org/grpc/channelz/grpc_channelz_v1"
 	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 )
 
@@ -98,7 +99,7 @@ func RunRegisterTests(t *testing.T, ec ExpectedStatusCodes) {
 
 // RunChannelz makes a channelz RPC.
 func RunChannelz(conn *grpc.ClientConn) error {
-	c := channelzpb.NewChannelzClient(conn)
+	c := channelzgrpc.NewChannelzClient(conn)
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	_, err := c.GetTopChannels(ctx, &channelzpb.GetTopChannelsRequest{}, grpc.WaitForReady(true))

channelz/service/func_linux.go

@@ -25,7 +25,7 @@ import (
 	durpb "github.com/golang/protobuf/ptypes/duration"
 	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 	"google.golang.org/grpc/internal/channelz"
-	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 func convertToPtypesDuration(sec int64, usec int64) *durpb.Duration {
@@ -35,32 +35,47 @@ func convertToPtypesDuration(sec int64, usec int64) *durpb.Duration {
 func sockoptToProto(skopts *channelz.SocketOptionData) []*channelzpb.SocketOption {
 	var opts []*channelzpb.SocketOption
 	if skopts.Linger != nil {
-		opts = append(opts, &channelzpb.SocketOption{
-			Name: "SO_LINGER",
-			Additional: testutils.MarshalAny(&channelzpb.SocketOptionLinger{
-				Active:   skopts.Linger.Onoff != 0,
-				Duration: convertToPtypesDuration(int64(skopts.Linger.Linger), 0),
-			}),
+		additional, err := anypb.New(&channelzpb.SocketOptionLinger{
+			Active:   skopts.Linger.Onoff != 0,
+			Duration: convertToPtypesDuration(int64(skopts.Linger.Linger), 0),
 		})
+		if err == nil {
+			opts = append(opts, &channelzpb.SocketOption{
+				Name:       "SO_LINGER",
+				Additional: additional,
+			})
+		} else {
+			logger.Warningf("Failed to marshal socket options linger %+v: %v", skopts.Linger, err)
+		}
 	}
 	if skopts.RecvTimeout != nil {
-		opts = append(opts, &channelzpb.SocketOption{
-			Name: "SO_RCVTIMEO",
-			Additional: testutils.MarshalAny(&channelzpb.SocketOptionTimeout{
-				Duration: convertToPtypesDuration(int64(skopts.RecvTimeout.Sec), int64(skopts.RecvTimeout.Usec)),
-			}),
+		additional, err := anypb.New(&channelzpb.SocketOptionTimeout{
+			Duration: convertToPtypesDuration(int64(skopts.RecvTimeout.Sec), int64(skopts.RecvTimeout.Usec)),
 		})
+		if err == nil {
+			opts = append(opts, &channelzpb.SocketOption{
+				Name:       "SO_RCVTIMEO",
+				Additional: additional,
+			})
+		} else {
+			logger.Warningf("Failed to marshal socket options receive timeout %+v: %v", skopts.RecvTimeout, err)
+		}
 	}
 	if skopts.SendTimeout != nil {
-		opts = append(opts, &channelzpb.SocketOption{
-			Name: "SO_SNDTIMEO",
-			Additional: testutils.MarshalAny(&channelzpb.SocketOptionTimeout{
-				Duration: convertToPtypesDuration(int64(skopts.SendTimeout.Sec), int64(skopts.SendTimeout.Usec)),
-			}),
+		additional, err := anypb.New(&channelzpb.SocketOptionTimeout{
+			Duration: convertToPtypesDuration(int64(skopts.SendTimeout.Sec), int64(skopts.SendTimeout.Usec)),
 		})
+		if err == nil {
+			opts = append(opts, &channelzpb.SocketOption{
+				Name:       "SO_SNDTIMEO",
+				Additional: additional,
+			})
+		} else {
+			logger.Warningf("Failed to marshal socket options send timeout %+v: %v", skopts.SendTimeout, err)
+		}
 	}
 	if skopts.TCPInfo != nil {
-		additional := testutils.MarshalAny(&channelzpb.SocketOptionTcpInfo{
+		additional, err := anypb.New(&channelzpb.SocketOptionTcpInfo{
 			TcpiState:       uint32(skopts.TCPInfo.State),
 			TcpiCaState:     uint32(skopts.TCPInfo.Ca_state),
 			TcpiRetransmits: uint32(skopts.TCPInfo.Retransmits),
@@ -91,10 +106,14 @@ func sockoptToProto(skopts *channelz.SocketOptionData) []*channelzpb.SocketOptio
 			TcpiAdvmss:       skopts.TCPInfo.Advmss,
 			TcpiReordering:   skopts.TCPInfo.Reordering,
 		})
-		opts = append(opts, &channelzpb.SocketOption{
-			Name:       "TCP_INFO",
-			Additional: additional,
-		})
+		if err == nil {
+			opts = append(opts, &channelzpb.SocketOption{
+				Name:       "TCP_INFO",
+				Additional: additional,
+			})
+		} else {
+			logger.Warningf("Failed to marshal socket options TCP info %+v: %v", skopts.TCPInfo, err)
+		}
 	}
 	return opts
 }

channelz/service/service.go

@@ -25,15 +25,18 @@ import (
 
 	"github.com/golang/protobuf/ptypes"
 	wrpb "github.com/golang/protobuf/ptypes/wrappers"
-	"google.golang.org/grpc"
 	channelzgrpc "google.golang.org/grpc/channelz/grpc_channelz_v1"
 	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
+
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/grpclog"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/protoadapt"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 func init() {
@@ -187,7 +190,7 @@ func securityToProto(se credentials.ChannelzSecurityValue) *channelzpb.Security
 		otherSecurity := &channelzpb.Security_OtherSecurity{
 			Name: v.Name,
 		}
-		if anyval, err := ptypes.MarshalAny(v.Value); err == nil {
+		if anyval, err := anypb.New(protoadapt.MessageV2Of(v.Value)); err == nil {
 			otherSecurity.Value = anyval
 		}
 		return &channelzpb.Security{Model: &channelzpb.Security_Other{Other: otherSecurity}}

channelz/service/service_sktopt_test.go

@@ -32,13 +32,13 @@ import (
 	"testing"
 
 	"github.com/golang/protobuf/ptypes"
-	durpb "github.com/golang/protobuf/ptypes/duration"
 	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
 	"golang.org/x/sys/unix"
-	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/protobuf/testing/protocmp"
+
+	durpb "github.com/golang/protobuf/ptypes/duration"
+	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 )
 
 func init() {
@@ -153,7 +153,7 @@ func (s) TestGetSocketOptions(t *testing.T) {
 	for i, s := range ss {
 		resp, _ := svr.GetSocket(ctx, &channelzpb.GetSocketRequest{SocketId: ids[i].Int()})
 		got, want := resp.GetSocket().GetRef(), &channelzpb.SocketRef{SocketId: ids[i].Int(), Name: strconv.Itoa(i)}
-		if !cmp.Equal(got, want, cmpopts.IgnoreUnexported(channelzpb.SocketRef{})) {
+		if !cmp.Equal(got, want, protocmp.Transform()) {
 			t.Fatalf("resp.GetSocket() returned metrics.GetRef() = %#v, want %#v", got, want)
 		}
 		socket, err := socketProtoToStruct(resp.GetSocket())

channelz/service/service_test.go

@@ -30,13 +30,13 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/google/go-cmp/cmp"
-	"github.com/google/go-cmp/cmp/cmpopts"
-	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/protobuf/testing/protocmp"
+
+	channelzpb "google.golang.org/grpc/channelz/grpc_channelz_v1"
 )
 
 func init() {
@@ -748,7 +748,7 @@ func (s) TestGetSocket(t *testing.T) {
 	for i, s := range ss {
 		resp, _ := svr.GetSocket(ctx, &channelzpb.GetSocketRequest{SocketId: ids[i].Int()})
 		got, want := resp.GetSocket().GetRef(), &channelzpb.SocketRef{SocketId: ids[i].Int(), Name: strconv.Itoa(i)}
-		if !cmp.Equal(got, want, cmpopts.IgnoreUnexported(channelzpb.SocketRef{})) {
+		if !cmp.Equal(got, want, protocmp.Transform()) {
 			t.Fatalf("resp.GetSocket() returned metrics.GetRef() = %#v, want %#v", got, want)
 		}
 		socket, err := socketProtoToStruct(resp.GetSocket())

clientconn.go

@@ -1113,7 +1113,11 @@ func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error)
 		return
 	}
 	ac.state = s
-	channelz.Infof(logger, ac.channelzID, "Subchannel Connectivity change to %v", s)
+	if lastErr == nil {
+		channelz.Infof(logger, ac.channelzID, "Subchannel Connectivity change to %v", s)
+	} else {
+		channelz.Infof(logger, ac.channelzID, "Subchannel Connectivity change to %v, last error: %s", s, lastErr)
+	}
 	ac.cc.handleSubConnStateChange(ac.acbw, s, lastErr)
 }
 

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

@@ -2,4 +2,4 @@ module google.golang.org/grpc/cmd/protoc-gen-go-grpc
 
 go 1.17
 
-require google.golang.org/protobuf v1.28.1
+require google.golang.org/protobuf v1.30.0

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

@@ -4,5 +4,5 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
-google.golang.org/protobuf v1.28.1 h1:d0NfwRgPtno5B1Wa6L2DAG+KivqkdutMf1UhdNx175w=
-google.golang.org/protobuf v1.28.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=

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

@@ -41,7 +41,7 @@ import (
 	"google.golang.org/protobuf/types/pluginpb"
 )
 
-const version = "1.2.0"
+const version = "1.3.0"
 
 var requireUnimplemented *bool
 

credentials/xds/xds.go

@@ -162,8 +162,10 @@ func (c *credsImpl) ClientHandshake(ctx context.Context, authority string, rawCo
 		}
 		// The SANs sent by the MeshCA are encoded as SPIFFE IDs. We need to
 		// only look at the SANs on the leaf cert.
-		if !hi.MatchingSANExists(certs[0]) {
-			return fmt.Errorf("SANs received in leaf certificate %+v does not match any of the accepted SANs", certs[0])
+		if cert := certs[0]; !hi.MatchingSANExists(cert) {
+			// TODO: Print the complete certificate once the x509 package
+			// supports a String() method on the Certificate type.
+			return fmt.Errorf("xds: received SANs {DNSNames: %v, EmailAddresses: %v, IPAddresses: %v, URIs: %v} do not match any of the accepted SANs", cert.DNSNames, cert.EmailAddresses, cert.IPAddresses, cert.URIs)
 		}
 		return nil
 	}

credentials/xds/xds_client_test.go

@@ -476,7 +476,7 @@ func (s) TestClientCredsHandshakeFailure(t *testing.T) {
 			handshakeFunc: testServerTLSHandshake,
 			rootProvider:  makeRootProvider(t, "x509/server_ca_cert.pem"),
 			san:           "bad-san",
-			wantErr:       "does not match any of the accepted SANs",
+			wantErr:       "do not match any of the accepted SANs",
 		},
 	}