Clean up Channel lifecycle, add ID per channel (#548)

Add a unique channel ID, and have an onClosed method for when a channel
is actually closed. This is useful for:
 - Stopping related objects that are no longer needed on shutdown
 - Raising an event when a channel is closed

channel.go

@@ -32,6 +32,7 @@ import (
 	"github.com/uber/tchannel-go/tnet"
 
 	"github.com/opentracing/opentracing-go"
+	"github.com/uber-go/atomic"
 	"golang.org/x/net/context"
 )
 
@@ -113,6 +114,7 @@ const (
 type Channel struct {
 	channelConnectionCommon
 
+	chID              uint32
 	createdStack      string
 	commonStatsTags   map[string]string
 	connectionOptions ConnectionOptions
@@ -141,6 +143,9 @@ type channelConnectionCommon struct {
 	timeNow       func() time.Time
 }
 
+// _nextChID is used to allocate unique IDs to every channel for debugging purposes.
+var _nextChID atomic.Uint32
+
 // Tracer returns the OpenTracing Tracer for this channel. If no tracer was provided
 // in the configuration, returns opentracing.GlobalTracer(). Note that this approach
 // allows opentracing.GlobalTracer() to be initialized _after_ the channel is created.
@@ -187,16 +192,20 @@ func NewChannel(serviceName string, opts *ChannelOptions) (*Channel, error) {
 		timeNow = time.Now
 	}
 
+	chID := _nextChID.Inc()
 	ch := &Channel{
 		channelConnectionCommon: channelConnectionCommon{
-			log:           logger,
+			log: logger.WithFields(
+				LogField{"chID", chID},
+				LogField{"service", serviceName},
+				LogField{"process", processName}),
 			relayLocal:    toStringSet(opts.RelayLocalHandlers),
 			statsReporter: statsReporter,
 			subChannels:   &subChannelMap{},
 			timeNow:       timeNow,
 			tracer:        opts.Tracer,
 		},
-
+		chID:              chID,
 		connectionOptions: opts.DefaultConnectionOptions,
 		relayHost:         opts.RelayHost,
 		relayMaxTimeout:   validateRelayMaxTimeout(opts.RelayMaxTimeout, logger),
@@ -639,19 +648,30 @@ func (ch *Channel) connectionCloseStateChange(c *Connection) {
 		updateTo = ChannelInboundClosed
 	}
 
+	var updatedToState ChannelState
 	if updateTo > 0 {
 		ch.mutable.Lock()
 		// Recheck the state as it's possible another goroutine changed the state
 		// from what we expected, and so we might make a stale change.
 		if ch.mutable.state == chState {
 			ch.mutable.state = updateTo
+			updatedToState = updateTo
 		}
 		ch.mutable.Unlock()
 		chState = updateTo
 	}
 
 	c.log.Debugf("ConnectionCloseStateChange channel state = %v connection minState = %v",
 		chState, minState)
+
+	if updatedToState == ChannelClosed {
+		ch.onClosed()
+	}
+}
+
+func (ch *Channel) onClosed() {
+	removeClosedChannel(ch)
+	ch.log.Infof("Channel closed.")
 }
 
 // Closed returns whether this channel has been closed with .Close()
@@ -675,6 +695,7 @@ func (ch *Channel) State() ChannelState {
 func (ch *Channel) Close() {
 	ch.Logger().Info("Channel.Close called.")
 	var connections []*Connection
+	var channelClosed bool
 	ch.mutable.Lock()
 
 	if ch.mutable.l != nil {
@@ -684,6 +705,7 @@ func (ch *Channel) Close() {
 	ch.mutable.state = ChannelStartClose
 	if len(ch.mutable.conns) == 0 {
 		ch.mutable.state = ChannelClosed
+		channelClosed = true
 	}
 	for _, c := range ch.mutable.conns {
 		connections = append(connections, c)
@@ -693,7 +715,10 @@ func (ch *Channel) Close() {
 	for _, c := range connections {
 		c.Close()
 	}
-	removeClosedChannel(ch)
+
+	if channelClosed {
+		ch.onClosed()
+	}
 }
 
 // RelayHost returns the channel's RelayHost, if any.

connection.go

@@ -191,8 +191,8 @@ type Connection struct {
 	}
 }
 
-// nextConnID gives an ID for each connection for debugging purposes.
-var nextConnID atomic.Uint32
+// _nextConnID is used to allocate unique IDs to every connection for debugging purposes.
+var _nextConnID atomic.Uint32
 
 type connectionState int
 
@@ -290,7 +290,7 @@ func (ch *Channel) newConnection(conn net.Conn, outboundHP string, initialState
 		framePool = DefaultFramePool
 	}
 
-	connID := nextConnID.Inc()
+	connID := _nextConnID.Inc()
 	log := ch.log.WithFields(LogFields{
 		{"connID", connID},
 		{"localAddr", conn.LocalAddr()},

introspection.go

@@ -55,6 +55,8 @@ type RuntimeVersion struct {
 
 // RuntimeState is a snapshot of the runtime state for a channel.
 type RuntimeState struct {
+	ID uint32 `json:"id"`
+
 	// CreatedStack is the stack for how this channel was created.
 	CreatedStack string `json:"createdStack"`
 
@@ -91,6 +93,7 @@ type GoRuntimeStateOptions struct {
 
 // ChannelInfo is the state of other channels in the same process.
 type ChannelInfo struct {
+	ID           uint32        `json:"id"`
 	CreatedStack string        `json:"createdStack"`
 	LocalPeer    LocalPeerInfo `json:"localPeer"`
 }
@@ -209,6 +212,7 @@ func (ch *Channel) IntrospectState(opts *IntrospectionOptions) *RuntimeState {
 	ch.mutable.RUnlock()
 
 	return &RuntimeState{
+		ID:             ch.chID,
 		CreatedStack:   ch.createdStack,
 		LocalPeer:      ch.PeerInfo(),
 		SubChannels:    ch.subChannels.IntrospectState(opts),
@@ -248,6 +252,7 @@ func (ch *Channel) IntrospectOthers(opts *IntrospectionOptions) map[string][]Cha
 // ReportInfo returns ChannelInfo for a channel.
 func (ch *Channel) ReportInfo(opts *IntrospectionOptions) ChannelInfo {
 	return ChannelInfo{
+		ID:           ch.chID,
 		CreatedStack: ch.createdStack,
 		LocalPeer:    ch.PeerInfo(),
 	}