client: make channelz a feature of a SubConn, not an addrConn

grpc · May 19, 2023 · 92d9bd9 · 92d9bd9
1 parent 098b2d0
commit 92d9bd9
Show file tree

Hide file tree

Showing 2 changed files with 50 additions and 35 deletions.
diff --git a/balancer_conn_wrappers.go b/balancer_conn_wrappers.go
@@ -23,6 +23,8 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"google.golang.org/grpc/balancer"
 	"google.golang.org/grpc/connectivity"
@@ -187,15 +189,31 @@ func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer
 	if len(addrs) <= 0 {
 		return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
 	}
-	ac, err := ccb.cc.newAddrConn(addrs, opts)
+
+	acbw := &acBalancerWrapper{producers: make(map[balancer.ProducerBuilder]*refCountedProducer)}
+	var err error
+	acbw.channelzID, err = channelz.RegisterSubChannel(acbw, ccb.cc.channelzID, "")
 	if err != nil {
+		// An error here should only happen if there is no ClientConn ID.
+		return nil, fmt.Errorf("Unexpected error creating channelz ID for subchannel: %v", err)
+	}
+
+	acbw.ac, err = ccb.cc.newAddrConn(addrs, acbw, opts)
+	if err != nil {
+		// An error here should only happen if the ClientConn is shutting down.
 		channelz.Warningf(logger, ccb.cc.channelzID, "acBalancerWrapper: NewSubConn: failed to newAddrConn: %v", err)
 		return nil, err
 	}
-	acbw := &acBalancerWrapper{ac: ac, producers: make(map[balancer.ProducerBuilder]*refCountedProducer)}
-	acbw.ac.mu.Lock()
-	ac.acbw = acbw
-	acbw.ac.mu.Unlock()
+
+	channelz.AddTraceEvent(logger, acbw.channelzID, 0, &channelz.TraceEventDesc{
+		Desc:     "Subchannel created",
+		Severity: channelz.CtInfo,
+		Parent: &channelz.TraceEventDesc{
+			Desc:     fmt.Sprintf("Subchannel(id:%d) created", acbw.channelzID.Int()),
+			Severity: channelz.CtInfo,
+		},
+	})
+
 	return acbw, nil
 }
 
@@ -236,11 +254,34 @@ func (ccb *ccBalancerWrapper) Target() string {
 // acBalancerWrapper is a wrapper on top of ac for balancers.
 // It implements balancer.SubConn interface.
 type acBalancerWrapper struct {
+	channelzID *channelz.Identifier
+
 	mu        sync.Mutex
 	ac        *addrConn
 	producers map[balancer.ProducerBuilder]*refCountedProducer
 }
 
+func (acbw *acBalancerWrapper) String() string {
+	return fmt.Sprintf("SubConn(id:%d)", acbw.channelzID.Int())
+}
+
+func (acbw *acBalancerWrapper) ChannelzMetric() *channelz.ChannelInternalMetric {
+	acbw.mu.Lock()
+	defer acbw.mu.Unlock()
+	acbw.ac.mu.Lock()
+	addr := acbw.ac.curAddr.Addr
+	acbw.ac.mu.Unlock()
+	ac := acbw.ac
+	return &channelz.ChannelInternalMetric{
+		State:                    ac.getState(),
+		Target:                   addr,
+		CallsStarted:             atomic.LoadInt64(&ac.czData.callsStarted),
+		CallsSucceeded:           atomic.LoadInt64(&ac.czData.callsSucceeded),
+		CallsFailed:              atomic.LoadInt64(&ac.czData.callsFailed),
+		LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&ac.czData.lastCallStartedTime)),
+	}
+}
+
 func (acbw *acBalancerWrapper) UpdateAddresses(addrs []resolver.Address) {
 	acbw.mu.Lock()
 	defer acbw.mu.Unlock()
@@ -266,7 +307,7 @@ func (acbw *acBalancerWrapper) UpdateAddresses(addrs []resolver.Address) {
 			return
 		}
 
-		newAC, err := cc.newAddrConn(addrs, opts)
+		newAC, err := cc.newAddrConn(addrs, acbw, opts)
 		if err != nil {
 			channelz.Warningf(logger, acbw.ac.channelzID, "acBalancerWrapper: UpdateAddresses: failed to newAddrConn: %v", err)
 			return

diff --git a/clientconn.go b/clientconn.go
@@ -733,7 +733,7 @@ func (cc *ClientConn) handleSubConnStateChange(sc balancer.SubConn, s connectivi
 // newAddrConn creates an addrConn for addrs and adds it to cc.conns.
 //
 // Caller needs to make sure len(addrs) > 0.
-func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
+func (cc *ClientConn) newAddrConn(addrs []resolver.Address, acbw *acBalancerWrapper, opts balancer.NewSubConnOptions) (*addrConn, error) {
 	ac := &addrConn{
 		state:        connectivity.Idle,
 		cc:           cc,
@@ -743,6 +743,8 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
 		czData:       new(channelzData),
 		resetBackoff: make(chan struct{}),
 		stateChan:    make(chan struct{}),
+		channelzID:   acbw.channelzID,
+		acbw:         acbw,
 	}
 	ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
 	// Track ac in cc. This needs to be done before any getTransport(...) is called.
@@ -752,20 +754,6 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
 		return nil, ErrClientConnClosing
 	}
 
-	var err error
-	ac.channelzID, err = channelz.RegisterSubChannel(ac, cc.channelzID, "")
-	if err != nil {
-		return nil, err
-	}
-	channelz.AddTraceEvent(logger, ac.channelzID, 0, &channelz.TraceEventDesc{
-		Desc:     "Subchannel created",
-		Severity: channelz.CtInfo,
-		Parent: &channelz.TraceEventDesc{
-			Desc:     fmt.Sprintf("Subchannel(id:%d) created", ac.channelzID.Int()),
-			Severity: channelz.CtInfo,
-		},
-	})
-
 	cc.conns[ac] = struct{}{}
 	return ac, nil
 }
@@ -1514,20 +1502,6 @@ func (ac *addrConn) getState() connectivity.State {
 	return ac.state
 }
 
-func (ac *addrConn) ChannelzMetric() *channelz.ChannelInternalMetric {
-	ac.mu.Lock()
-	addr := ac.curAddr.Addr
-	ac.mu.Unlock()
-	return &channelz.ChannelInternalMetric{
-		State:                    ac.getState(),
-		Target:                   addr,
-		CallsStarted:             atomic.LoadInt64(&ac.czData.callsStarted),
-		CallsSucceeded:           atomic.LoadInt64(&ac.czData.callsSucceeded),
-		CallsFailed:              atomic.LoadInt64(&ac.czData.callsFailed),
-		LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&ac.czData.lastCallStartedTime)),
-	}
-}
-
 func (ac *addrConn) incrCallsStarted() {
 	atomic.AddInt64(&ac.czData.callsStarted, 1)
 	atomic.StoreInt64(&ac.czData.lastCallStartedTime, time.Now().UnixNano())