Add a Log for Newly Connected gRPC Clients in the Beacon Node (#6233)

* interceptor disable logging via feature flag
* Merge branch 'master' into validator-connection-logs

beacon-chain/rpc/BUILD.bazel

@@ -40,6 +40,7 @@ go_library(
         "@io_opencensus_go//plugin/ocgrpc:go_default_library",
         "@org_golang_google_grpc//:go_default_library",
         "@org_golang_google_grpc//credentials:go_default_library",
+        "@org_golang_google_grpc//peer:go_default_library",
         "@org_golang_google_grpc//reflection:go_default_library",
     ],
 )

beacon-chain/rpc/service.go

@@ -42,6 +42,7 @@ import (
 	"go.opencensus.io/plugin/ocgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/reflection"
 )
 
@@ -95,6 +96,7 @@ type Service struct {
 	slasherCredentialError  error
 	slasherClient           slashpb.SlasherClient
 	stateGen                *stategen.State
+	connectedRPCClients     map[net.Addr]bool
 }
 
 // Config options for the beacon node RPC server.
@@ -172,6 +174,7 @@ func NewService(ctx context.Context, cfg *Config) *Service {
 		slasherCert:             cfg.SlasherCert,
 		stateGen:                cfg.StateGen,
 		enableDebugRPCEndpoints: cfg.EnableDebugRPCEndpoints,
+		connectedRPCClients:     make(map[net.Addr]bool),
 	}
 }
 
@@ -193,6 +196,7 @@ func (s *Service) Start() {
 			),
 			grpc_prometheus.StreamServerInterceptor,
 			grpc_opentracing.StreamServerInterceptor(),
+			s.validatorStreamConnectionInterceptor,
 		)),
 		grpc.UnaryInterceptor(middleware.ChainUnaryServer(
 			recovery.UnaryServerInterceptor(
@@ -200,6 +204,7 @@ func (s *Service) Start() {
 			),
 			grpc_prometheus.UnaryServerInterceptor,
 			grpc_opentracing.UnaryServerInterceptor(),
+			s.validatorUnaryConnectionInterceptor,
 		)),
 	}
 	grpc_prometheus.EnableHandlingTimeHistogram()
@@ -362,3 +367,41 @@ func (s *Service) Status() error {
 	}
 	return nil
 }
+
+// Stream interceptor for new validator client connections to the beacon node.
+func (s *Service) validatorStreamConnectionInterceptor(
+	srv interface{},
+	ss grpc.ServerStream,
+	_ *grpc.StreamServerInfo,
+	handler grpc.StreamHandler,
+) error {
+	s.logNewClientConnection(ss.Context())
+	return handler(srv, ss)
+}
+
+// Unary interceptor for new validator client connections to the beacon node.
+func (s *Service) validatorUnaryConnectionInterceptor(
+	ctx context.Context,
+	req interface{},
+	_ *grpc.UnaryServerInfo,
+	handler grpc.UnaryHandler,
+) (interface{}, error) {
+	s.logNewClientConnection(ctx)
+	return handler(ctx, req)
+}
+
+func (s *Service) logNewClientConnection(ctx context.Context) {
+	if featureconfig.Get().DisableGRPCConnectionLogs {
+		return
+	}
+	if clientInfo, ok := peer.FromContext(ctx); ok {
+		// Check if we have not yet observed this grpc client connection
+		// in the running beacon node.
+		if !s.connectedRPCClients[clientInfo.Addr] {
+			log.WithFields(logrus.Fields{
+				"addr": clientInfo.Addr.String(),
+			}).Infof("New gRPC client connected to beacon node")
+			s.connectedRPCClients[clientInfo.Addr] = true
+		}
+	}
+}

go.sum

@@ -1491,6 +1491,7 @@ gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
 gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo=
 gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw=
+grpc.go4.org v0.0.0-20170609214715-11d0a25b4919 h1:tmXTu+dfa+d9Evp8NpJdgOy6+rt8/x4yG7qPBrtNfLY=
 grpc.go4.org v0.0.0-20170609214715-11d0a25b4919/go.mod h1:77eQGdRu53HpSqPFJFmuJdjuHRquDANNeA4x7B8WQ9o=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

shared/featureconfig/config.go

@@ -66,6 +66,9 @@ type Flags struct {
 	// as the chain head. UNSAFE, use with caution.
 	DisableForkChoice bool
 
+	// Logging related toggles.
+	DisableGRPCConnectionLogs bool // Disables logging when a new grpc client has connected.
+
 	// Slasher toggles.
 	DisableBroadcastSlashings bool // DisableBroadcastSlashings disables p2p broadcasting of proposer and attester slashings.
 	EnableHistoricalDetection bool // EnableHistoricalDetection disables historical attestation detection and performs detection on the chain head immediately.
@@ -240,6 +243,9 @@ func ConfigureBeaconChain(ctx *cli.Context) {
 		log.Warn("Disabling reducing attester state copy")
 		cfg.ReduceAttesterStateCopy = false
 	}
+	if ctx.IsSet(disableGRPCConnectionLogging.Name) {
+		cfg.DisableGRPCConnectionLogs = true
+	}
 	Init(cfg)
 }
 

shared/featureconfig/flags.go

@@ -165,6 +165,10 @@ var (
 		Name:  "disable-init-sync-wrr",
 		Usage: "Disables weighted round robin fetching optimization",
 	}
+	disableGRPCConnectionLogging = &cli.BoolFlag{
+		Name:  "disable-grpc-connection-logging",
+		Usage: "Disables displaying logs for newly connected grpc clients",
+	}
 )
 
 // devModeFlags holds list of flags that are set when development mode is on.
@@ -531,6 +535,7 @@ var BeaconChainFlags = append(deprecatedFlags, []cli.Flag{
 	disableNewStateMgmt,
 	enableKadDht,
 	disableReduceAttesterStateCopy,
+	disableGRPCConnectionLogging,
 }...)
 
 // E2EBeaconChainFlags contains a list of the beacon chain feature flags to be tested in E2E.