Skip to content

Commit

Permalink
some more cleanups
Browse files Browse the repository at this point in the history
  • Loading branch information
@easwars
easwars committed Feb 11, 2022
1 parent a480380 commit 535e619
Show file tree
Hide file tree
Showing 5 changed files with 146 additions and 99 deletions.
8 changes: 7 additions & 1 deletion channelz/service/service_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"fmt"
"net"
"strconv"
"strings"
"testing"
"time"

Expand Down Expand Up @@ -477,13 +478,15 @@ func (s) TestGetServerSocketsNonZeroStartID(t *testing.T) {
func (s) TestGetChannel(t *testing.T) {
czCleanup := channelz.NewChannelzStorageForTesting()
defer cleanupWrapper(czCleanup, t)

refNames := []string{"top channel 1", "nested channel 1", "sub channel 2", "nested channel 3"}
ids := make([]*channelz.Identifier, 4)
ids[0] = channelz.RegisterChannel(&dummyChannel{}, nil, refNames[0])
channelz.AddTraceEvent(logger, ids[0], 0, &channelz.TraceEventDesc{
Desc: "Channel Created",
Severity: channelz.CtInfo,
})

ids[1] = channelz.RegisterChannel(&dummyChannel{}, ids[0], refNames[1])
channelz.AddTraceEvent(logger, ids[1], 0, &channelz.TraceEventDesc{
Desc: "Channel Created",
Expand All @@ -507,6 +510,7 @@ func (s) TestGetChannel(t *testing.T) {
Severity: channelz.CtInfo,
},
})

ids[3] = channelz.RegisterChannel(&dummyChannel{}, ids[1], refNames[3])
channelz.AddTraceEvent(logger, ids[3], 0, &channelz.TraceEventDesc{
Desc: "Channel Created",
Expand All @@ -524,9 +528,11 @@ func (s) TestGetChannel(t *testing.T) {
Desc: "Resolver returns an empty address list",
Severity: channelz.CtWarning,
})

for _, id := range ids {
defer channelz.RemoveEntry(id)
}

svr := newCZServer()
ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
defer cancel()
Expand Down Expand Up @@ -555,7 +561,7 @@ func (s) TestGetChannel(t *testing.T) {
}

for i, e := range trace.Events {
if e.GetDescription() != want[i].desc {
if !strings.Contains(e.GetDescription(), want[i].desc) {
t.Fatalf("trace: GetDescription want %#v, got %#v", want[i].desc, e.GetDescription())
}
if e.GetSeverity() != want[i].severity {
Expand Down
104 changes: 50 additions & 54 deletions clientconn.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -159,23 +159,20 @@ func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *
}
}()

if channelz.IsOn() {
if cc.dopts.channelzParentID != nil {
cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, cc.dopts.channelzParentID, target)
channelz.AddTraceEvent(logger, cc.channelzID, 0, &channelz.TraceEventDesc{
Desc: "Channel Created",
Severity: channelz.CtInfo,
Parent: &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Nested Channel(id:%d) created", cc.channelzID.Int()),
Severity: channelz.CtInfo,
},
})
} else {
cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, nil, target)
channelz.Info(logger, cc.channelzID, "Channel Created")
pid := cc.dopts.channelzParentID
cc.channelzID = channelz.RegisterChannel(&channelzChannel{cc}, pid, target)
ted := &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Channel(id:%d) created", cc.channelzID.Int()),
Severity: channelz.CtInfo,
}
if cc.dopts.channelzParentID != nil {
ted.Parent = &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Nested Channel(id:%d) created", cc.channelzID.Int()),
Severity: channelz.CtInfo,
}
cc.csMgr.channelzID = cc.channelzID
}
channelz.AddTraceEvent(logger, cc.channelzID, 1, ted)
cc.csMgr.channelzID = cc.channelzID

if cc.dopts.copts.TransportCredentials == nil && cc.dopts.copts.CredsBundle == nil {
return nil, errNoTransportSecurity
Expand Down Expand Up @@ -768,21 +765,21 @@ func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSub
cc.mu.Unlock()
return nil, ErrClientConnClosing
}
if channelz.IsOn() {
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,
},
})

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: fmt.Sprintf("Subchannel(id:%d) created", ac.channelzID.Int()),
Severity: channelz.CtInfo,
Parent: &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Subchannel(id:%d) created", ac.channelzID.Int()),
Severity: channelz.CtInfo,
},
})

cc.conns[ac] = struct{}{}
cc.mu.Unlock()
return ac, nil
Expand Down Expand Up @@ -1089,22 +1086,22 @@ func (cc *ClientConn) Close() error {
for ac := range conns {
ac.tearDown(ErrClientConnClosing)
}
if channelz.IsOn() {
ted := &channelz.TraceEventDesc{
Desc: "Channel Deleted",
ted := &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Channel(id:%d) deleted", cc.channelzID.Int()),
Severity: channelz.CtInfo,
}
if cc.dopts.channelzParentID != nil {
ted.Parent = &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Nested channel(id:%d) deleted", cc.channelzID.Int()),
Severity: channelz.CtInfo,
}
if cc.dopts.channelzParentID != nil {
ted.Parent = &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Nested channel(id:%d) deleted", cc.channelzID.Int()),
Severity: channelz.CtInfo,
}
}
channelz.AddTraceEvent(logger, cc.channelzID, 0, ted)
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
// the entity being deleted, and thus prevent it from being deleted right away.
channelz.RemoveEntry(cc.channelzID)
}
channelz.AddTraceEvent(logger, cc.channelzID, 0, ted)
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add
// trace reference to the entity being deleted, and thus prevent it from being
// deleted right away.
channelz.RemoveEntry(cc.channelzID)

return nil
}

Expand Down Expand Up @@ -1501,19 +1498,18 @@ func (ac *addrConn) tearDown(err error) {
curTr.GracefulClose()
ac.mu.Lock()
}
if channelz.IsOn() {
channelz.AddTraceEvent(logger, ac.channelzID, 0, &channelz.TraceEventDesc{
Desc: "Subchannel Deleted",
channelz.AddTraceEvent(logger, ac.channelzID, 0, &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Subchannel(id:%d) deleted", ac.channelzID.Int()),
Severity: channelz.CtInfo,
Parent: &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Subchannel(id:%d) deleted", ac.channelzID.Int()),
Severity: channelz.CtInfo,
Parent: &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Subchanel(id:%d) deleted", ac.channelzID.Int()),
Severity: channelz.CtInfo,
},
})
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add trace reference to
// the entity being deleted, and thus prevent it from being deleted right away.
channelz.RemoveEntry(ac.channelzID)
}
},
})
// TraceEvent needs to be called before RemoveEntry, as TraceEvent may add
// trace reference to the entity being deleted, and thus prevent it from
// being deleted right away.
channelz.RemoveEntry(ac.channelzID)
ac.mu.Unlock()
}

Expand Down
54 changes: 47 additions & 7 deletions internal/channelz/funcs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -190,9 +190,14 @@ func GetServer(id int64) *ServerMetric {
// (identified by pid). pid == nil means no parent.
//
// Returns a unique channelz identifier assigned to this channel.
//
// If channelz is not turned ON, this function is a no-op.
func RegisterChannel(c Channel, pid *Identifier, ref string) *Identifier {
id := idGen.genID()
if !IsOn() {
return nil
}

id := idGen.genID()
parent := int64(0)
isTopChannel := true
if pid != nil {
Expand All @@ -218,7 +223,13 @@ func RegisterChannel(c Channel, pid *Identifier, ref string) *Identifier {
// (identified by pid).
//
// Returns a unique channelz identifier assigned to this subChannel.
//
// If channelz is not turned ON, this function is a no-op.
func RegisterSubChannel(c Channel, pid *Identifier, ref string) (*Identifier, error) {
if !IsOn() {
return nil, nil
}

if pid == nil {
return nil, errors.New("a SubChannel's parent id cannot be nil")
}
Expand All @@ -237,7 +248,13 @@ func RegisterSubChannel(c Channel, pid *Identifier, ref string) (*Identifier, er

// RegisterServer registers the given server s in channelz database. It returns
// the unique channelz tracking id assigned to this server.
//
// If channelz is not turned ON, this function is a no-op.
func RegisterServer(s Server, ref string) *Identifier {
if !IsOn() {
return nil
}

id := idGen.genID()
svr := &server{
refName: ref,
Expand All @@ -254,7 +271,13 @@ func RegisterServer(s Server, ref string) *Identifier {
// with ref as its reference name, and add it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this listen socket.
//
// If channelz is not turned ON, this function is a no-op.
func RegisterListenSocket(s Socket, pid *Identifier, ref string) (*Identifier, error) {
if !IsOn() {
return nil, nil
}

if pid == nil {
return nil, errors.New("a ListenSocket's parent id cannot be 0")
}
Expand All @@ -268,7 +291,13 @@ func RegisterListenSocket(s Socket, pid *Identifier, ref string) (*Identifier, e
// with ref as its reference name, and adds it to the child list of its parent
// (identified by pid). It returns the unique channelz tracking id assigned to
// this normal socket.
//
// If channelz is not turned ON, this function is a no-op.
func RegisterNormalSocket(s Socket, pid *Identifier, ref string) (*Identifier, error) {
if !IsOn() {
return nil, nil
}

if pid == nil {
return nil, errors.New("a NormalSocket's parent id cannot be 0")
}
Expand All @@ -280,21 +309,30 @@ func RegisterNormalSocket(s Socket, pid *Identifier, ref string) (*Identifier, e

// RemoveEntry removes an entry with unique channelz tracking id to be id from
// channelz database.
//
// If channelz is not turned ON, this function is a no-op.
func RemoveEntry(id *Identifier) {
if !IsOn() {
return
}
db.get().removeEntry(id.Int())
}

// TraceEventDesc is what the caller of AddTraceEvent should provide to describe the event to be added
// to the channel trace.
// The Parent field is optional. It is used for event that will be recorded in the entity's parent
// trace also.
// TraceEventDesc is what the caller of AddTraceEvent should provide to describe
// the event to be added to the channel trace.
//
// The Parent field is optional. It is used for an event that will be recorded
// in the entity's parent trace.
type TraceEventDesc struct {
Desc string
Severity Severity
Parent *TraceEventDesc
}

// AddTraceEvent adds trace related to the entity with specified id, using the provided TraceEventDesc.
// AddTraceEvent adds trace related to the entity with specified id, using the
// provided TraceEventDesc.
//
// If channelz is not turned ON, this will simply log the event descriptions.
func AddTraceEvent(l grpclog.DepthLoggerV2, id *Identifier, depth int, desc *TraceEventDesc) {
for d := desc; d != nil; d = d.Parent {
switch d.Severity {
Expand All @@ -309,7 +347,9 @@ func AddTraceEvent(l grpclog.DepthLoggerV2, id *Identifier, depth int, desc *Tra
if getMaxTraceEntry() == 0 {
return
}
db.get().traceEvent(id.Int(), desc)
if IsOn() {
db.get().traceEvent(id.Int(), desc)
}
}

// channelMap is the storage data structure for channelz.
Expand Down
23 changes: 6 additions & 17 deletions resolver_conn_wrapper.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,14 +19,14 @@
package grpc

import (
"fmt"
"strings"
"sync"

"google.golang.org/grpc/balancer"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/internal/channelz"
"google.golang.org/grpc/internal/grpcsync"
"google.golang.org/grpc/internal/pretty"
"google.golang.org/grpc/resolver"
"google.golang.org/grpc/serviceconfig"
)
Expand Down Expand Up @@ -97,10 +97,7 @@ func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
if ccr.done.HasFired() {
return nil
}
channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s)
if channelz.IsOn() {
ccr.addChannelzTraceEvent(s)
}
ccr.addChannelzTraceEvent(s)
ccr.curState = s
if err := ccr.cc.updateResolverState(ccr.curState, nil); err == balancer.ErrBadResolverState {
return balancer.ErrBadResolverState
Expand All @@ -125,10 +122,7 @@ func (ccr *ccResolverWrapper) NewAddress(addrs []resolver.Address) {
if ccr.done.HasFired() {
return
}
channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending new addresses to cc: %v", addrs)
if channelz.IsOn() {
ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
}
ccr.addChannelzTraceEvent(resolver.State{Addresses: addrs, ServiceConfig: ccr.curState.ServiceConfig})
ccr.curState.Addresses = addrs
ccr.cc.updateResolverState(ccr.curState, nil)
}
Expand All @@ -141,7 +135,7 @@ func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
if ccr.done.HasFired() {
return
}
channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: got new service config: %v", sc)
channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: got new service config: %s", sc)
if ccr.cc.dopts.disableServiceConfig {
channelz.Info(logger, ccr.cc.channelzID, "Service config lookups disabled; ignoring config")
return
Expand All @@ -151,9 +145,7 @@ func (ccr *ccResolverWrapper) NewServiceConfig(sc string) {
channelz.Warningf(logger, ccr.cc.channelzID, "ccResolverWrapper: error parsing service config: %v", scpr.Err)
return
}
if channelz.IsOn() {
ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr})
}
ccr.addChannelzTraceEvent(resolver.State{Addresses: ccr.curState.Addresses, ServiceConfig: scpr})
ccr.curState.ServiceConfig = scpr
ccr.cc.updateResolverState(ccr.curState, nil)
}
Expand All @@ -180,8 +172,5 @@ func (ccr *ccResolverWrapper) addChannelzTraceEvent(s resolver.State) {
} else if len(ccr.curState.Addresses) == 0 && len(s.Addresses) > 0 {
updates = append(updates, "resolver returned new addresses")
}
channelz.AddTraceEvent(logger, ccr.cc.channelzID, 0, &channelz.TraceEventDesc{
Desc: fmt.Sprintf("Resolver state updated: %+v (%v)", s, strings.Join(updates, "; ")),
Severity: channelz.CtInfo,
})
channelz.Infof(logger, ccr.cc.channelzID, "Resolver state updated: %s (%v)", pretty.ToJSON(s), strings.Join(updates, "; "))
}

0 comments on commit 535e619

Please sign in to comment.