channelz: include channelz identifier in logs (#5192)

Author: easwars

balancer/balancer.go

@@ -27,6 +27,7 @@ import (
 	"net"
 	"strings"
 
+	"google.golang.org/grpc/channelz"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/internal"
@@ -192,7 +193,7 @@ type BuildOptions struct {
 	// server can ignore this field.
 	Authority string
 	// ChannelzParentID is the parent ClientConn's channelz ID.
-	ChannelzParentID int64
+	ChannelzParentID *channelz.Identifier
 	// CustomUserAgent is the custom user agent set on the parent ClientConn.
 	// The balancer should set the same custom user agent if it creates a
 	// ClientConn.

balancer/grpclb/grpclb_remote_balancer.go

@@ -35,7 +35,6 @@ import (
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/backoff"
-	"google.golang.org/grpc/internal/channelz"
 	imetadata "google.golang.org/grpc/internal/metadata"
 	"google.golang.org/grpc/keepalive"
 	"google.golang.org/grpc/metadata"
@@ -240,9 +239,7 @@ func (lb *lbBalancer) newRemoteBalancerCCWrapper() {
 	// Explicitly set pickfirst as the balancer.
 	dopts = append(dopts, grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"pick_first"}`))
 	dopts = append(dopts, grpc.WithResolvers(lb.manualResolver))
-	if channelz.IsOn() {
-		dopts = append(dopts, grpc.WithChannelzParentID(lb.opt.ChannelzParentID))
-	}
+	dopts = append(dopts, grpc.WithChannelzParentID(lb.opt.ChannelzParentID))
 
 	// Enable Keepalive for grpclb client.
 	dopts = append(dopts, grpc.WithKeepaliveParams(keepalive.ClientParameters{

balancer_conn_wrappers.go

@@ -184,6 +184,7 @@ func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer
 	}
 	ac, err := ccb.cc.newAddrConn(addrs, opts)
 	if err != nil {
+		channelz.Warningf(logger, ccb.cc.channelzID, "acBalancerWrapper: NewSubConn: failed to newAddrConn: %v", err)
 		return nil, err
 	}
 	acbw := &acBalancerWrapper{ac: ac}

call_test.go

@@ -31,6 +31,7 @@ import (
 	"time"
 
 	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/internal/channelz"
 	"google.golang.org/grpc/internal/transport"
 	"google.golang.org/grpc/status"
 )
@@ -123,12 +124,16 @@ type server struct {
 	startedErr chan error // sent nil or an error after server starts
 	mu         sync.Mutex
 	conns      map[transport.ServerTransport]bool
+	channelzID *channelz.Identifier
 }
 
 type ctxKey string
 
 func newTestServer() *server {
-	return &server{startedErr: make(chan error, 1)}
+	return &server{
+		startedErr: make(chan error, 1),
+		channelzID: channelz.NewIdentifierForTesting(channelz.RefServer, time.Now().Unix(), nil),
+	}
 }
 
 // start starts server. Other goroutines should block on s.startedErr for further operations.
@@ -158,10 +163,12 @@ func (s *server) start(t *testing.T, port int, maxStreams uint32) {
 			return
 		}
 		config := &transport.ServerConfig{
-			MaxStreams: maxStreams,
+			MaxStreams:       maxStreams,
+			ChannelzParentID: s.channelzID,
 		}
 		st, err := transport.NewServerTransport(conn, config)
 		if err != nil {
+			t.Errorf("failed to create server transport: %v", err)
 			continue
 		}
 		s.mu.Lock()

channelz/channelz.go

@@ -0,0 +1,36 @@
+/*
+ *
+ * Copyright 2020 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+// Package channelz exports internals of the channelz implementation as required
+// by other gRPC packages.
+//
+// The implementation of the channelz spec as defined in
+// https://github.com/grpc/proposal/blob/master/A14-channelz.md, is provided by
+// the `internal/channelz` package.
+//
+// Experimental
+//
+// Notice: All APIs in this package are experimental and may be removed in a
+// later release.
+package channelz
+
+import "google.golang.org/grpc/internal/channelz"
+
+// Identifier is an opaque identifier which uniquely identifies an entity in the
+// channelz database.
+type Identifier = channelz.Identifier

channelz/service/service_sktopt_test.go

@@ -28,15 +28,17 @@ package service
 
 import (
 	"context"
-	"reflect"
 	"strconv"
 	"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"
 )
 
 func init() {
@@ -139,20 +141,27 @@ func (s) TestGetSocketOptions(t *testing.T) {
 		},
 	}
 	svr := newCZServer()
-	ids := make([]int64, len(ss))
+	ids := make([]*channelz.Identifier, len(ss))
 	svrID := channelz.RegisterServer(&dummyServer{}, "")
 	defer channelz.RemoveEntry(svrID)
 	for i, s := range ss {
-		ids[i] = channelz.RegisterNormalSocket(s, svrID, strconv.Itoa(i))
+		ids[i], _ = channelz.RegisterNormalSocket(s, svrID, strconv.Itoa(i))
 		defer channelz.RemoveEntry(ids[i])
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 	for i, s := range ss {
-		resp, _ := svr.GetSocket(ctx, &channelzpb.GetSocketRequest{SocketId: ids[i]})
-		metrics := resp.GetSocket()
-		if !reflect.DeepEqual(metrics.GetRef(), &channelzpb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}) || !reflect.DeepEqual(socketProtoToStruct(metrics), s) {
-			t.Fatalf("resp.GetSocket() want: metrics.GetRef() = %#v and %#v, got: metrics.GetRef() = %#v and %#v", &channelzpb.SocketRef{SocketId: ids[i], Name: strconv.Itoa(i)}, s, metrics.GetRef(), socketProtoToStruct(metrics))
+		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{})) {
+			t.Fatalf("resp.GetSocket() returned metrics.GetRef() = %#v, want %#v", got, want)
+		}
+		socket, err := socketProtoToStruct(resp.GetSocket())
+		if err != nil {
+			t.Fatal(err)
+		}
+		if diff := cmp.Diff(s, socket, protocmp.Transform(), cmp.AllowUnexported(dummySocket{})); diff != "" {
+			t.Fatalf("unexpected socket, diff (-want +got):\n%s", diff)
 		}
 	}
 }