FAB-16695 Re-use metrics for comm GRPCServer

Today, the comm.NewGRPCServer code creates and registers metrics, given
a supplied metrics provider.  This is problematic because we should only
have one metrics provider, and re-registering a metric causes a panic.

Thus, if the admin service, or the Raft cluster service are put onto
different ports when Prometheus is enabled, the result is a crash.  This
CR fixes both peer and orderer initialization to re-use the metric
components across servers.

Signed-off-by: Jason Yellick <jyellick@us.ibm.com>
Change-Id: I80a33c1a3bd4cb25b08ca09297b098b11963dbba

core/comm/config.go

@@ -60,8 +60,8 @@ type ServerConfig struct {
 	UnaryInterceptors []grpc.UnaryServerInterceptor
 	// Logger specifies the logger the server will use
 	Logger *flogging.FabricLogger
-	// Metrics Provider
-	MetricsProvider metrics.Provider
+	// ServerStatsHandler should be set if metrics on connections are to be reported.
+	ServerStatsHandler *ServerStatsHandler
 }
 
 // ClientConfig defines the parameters for configuring a GRPCClient instance

core/comm/server.go

@@ -156,9 +156,8 @@ func NewGRPCServerFromListener(listener net.Listener, serverConfig ServerConfig)
 		)
 	}
 
-	if serverConfig.MetricsProvider != nil {
-		sh := NewServerStatsHandler(serverConfig.MetricsProvider)
-		serverOpts = append(serverOpts, grpc.StatsHandler(sh))
+	if serverConfig.ServerStatsHandler != nil {
+		serverOpts = append(serverOpts, grpc.StatsHandler(serverConfig.ServerStatsHandler))
 	}
 
 	grpcServer.server = grpc.NewServer(serverOpts...)

core/comm/serverstatshandler_test.go

@@ -66,8 +66,8 @@ func TestConnMetricsGRPCServer(t *testing.T) {
 	srv, err := comm.NewGRPCServerFromListener(
 		listener,
 		comm.ServerConfig{
-			SecOpts:         &comm.SecureOptions{UseTLS: false},
-			MetricsProvider: fakeProvider,
+			SecOpts:            &comm.SecureOptions{UseTLS: false},
+			ServerStatsHandler: comm.NewServerStatsHandler(fakeProvider),
 		},
 	)
 	gt.Expect(err).NotTo(HaveOccurred())

orderer/common/server/main.go

@@ -361,7 +361,7 @@ func configureClusterListener(conf *localconfig.TopLevel, generalConf comm.Serve
 		StreamInterceptors: generalConf.StreamInterceptors,
 		UnaryInterceptors:  generalConf.UnaryInterceptors,
 		ConnectionTimeout:  generalConf.ConnectionTimeout,
-		MetricsProvider:    generalConf.MetricsProvider,
+		ServerStatsHandler: generalConf.ServerStatsHandler,
 		Logger:             generalConf.Logger,
 		KaOpts:             generalConf.KaOpts,
 		SecOpts: &comm.SecureOptions{
@@ -488,16 +488,17 @@ func initializeServerConfig(conf *localconfig.TopLevel, metricsProvider metrics.
 	kaOpts.ServerTimeout = conf.General.Keepalive.ServerTimeout
 
 	commLogger := flogging.MustGetLogger("core.comm").With("server", "Orderer")
+
 	if metricsProvider == nil {
 		metricsProvider = &disabled.Provider{}
 	}
 
 	return comm.ServerConfig{
-		SecOpts:           secureOpts,
-		KaOpts:            kaOpts,
-		Logger:            commLogger,
-		MetricsProvider:   metricsProvider,
-		ConnectionTimeout: conf.General.ConnectionTimeout,
+		SecOpts:            secureOpts,
+		KaOpts:             kaOpts,
+		Logger:             commLogger,
+		ServerStatsHandler: comm.NewServerStatsHandler(metricsProvider),
+		ConnectionTimeout:  conf.General.ConnectionTimeout,
 		StreamInterceptors: []grpc.StreamServerInterceptor{
 			grpcmetrics.StreamServerInterceptor(grpcmetrics.NewStreamMetrics(metricsProvider)),
 			grpclogging.StreamServerInterceptor(flogging.MustGetLogger("comm.grpc.server").Zap()),

orderer/common/server/main_test.go

@@ -120,12 +120,12 @@ func TestInitializeServerConfig(t *testing.T) {
 
 	sc = initializeServerConfig(conf, nil)
 	assert.NotNil(t, sc.Logger)
-	assert.Equal(t, &disabled.Provider{}, sc.MetricsProvider)
+	assert.Equal(t, comm.NewServerStatsHandler(&disabled.Provider{}), sc.ServerStatsHandler)
 	assert.Len(t, sc.UnaryInterceptors, 2)
 	assert.Len(t, sc.StreamInterceptors, 2)
 
 	sc = initializeServerConfig(conf, &prometheus.Provider{})
-	assert.Equal(t, &prometheus.Provider{}, sc.MetricsProvider)
+	assert.NotNil(t, sc.ServerStatsHandler)
 
 	goodFile := "main.go"
 	badFile := "does_not_exist"

peer/node/start.go

@@ -220,7 +220,7 @@ func serve(args []string) error {
 
 	throttle := comm.NewThrottle(grpcMaxConcurrency)
 	serverConfig.Logger = flogging.MustGetLogger("core.comm").With("server", "PeerServer")
-	serverConfig.MetricsProvider = metricsProvider
+	serverConfig.ServerStatsHandler = comm.NewServerStatsHandler(metricsProvider)
 	serverConfig.UnaryInterceptors = append(
 		serverConfig.UnaryInterceptors,
 		grpcmetrics.UnaryServerInterceptor(grpcmetrics.NewUnaryMetrics(metricsProvider)),
@@ -273,7 +273,7 @@ func serve(args []string) error {
 	logger.Debugf("Running peer")
 
 	// Start the Admin server
-	startAdminServer(listenAddr, peerServer.Server(), metricsProvider)
+	startAdminServer(listenAddr, peerServer.Server(), serverConfig)
 
 	privDataDist := func(channel string, txID string, privateData *transientstore.TxPvtReadWriteSetWithConfigInfo, blkHt uint64) error {
 		return service.GetGossipService().DistributePrivateData(channel, txID, privateData, blkHt)
@@ -798,7 +798,7 @@ func adminHasSeparateListener(peerListenAddr string, adminListenAddress string)
 	return adminPort != peerPort
 }
 
-func startAdminServer(peerListenAddr string, peerServer *grpc.Server, metricsProvider metrics.Provider) {
+func startAdminServer(peerListenAddr string, peerServer *grpc.Server, baseServerConfig comm.ServerConfig) {
 	adminListenAddress := viper.GetString("peer.adminService.listenAddress")
 	separateLsnrForAdmin := adminHasSeparateListener(peerListenAddr, adminListenAddress)
 	mspID := viper.GetString("peer.localMspId")
@@ -810,21 +810,11 @@ func startAdminServer(peerListenAddr string, peerServer *grpc.Server, metricsPro
 		if err != nil {
 			logger.Fatalf("Error loading secure config for admin service (%s)", err)
 		}
-		throttle := comm.NewThrottle(grpcMaxConcurrency)
 		serverConfig.Logger = flogging.MustGetLogger("core.comm").With("server", "AdminServer")
-		serverConfig.MetricsProvider = metricsProvider
-		serverConfig.UnaryInterceptors = append(
-			serverConfig.UnaryInterceptors,
-			grpcmetrics.UnaryServerInterceptor(grpcmetrics.NewUnaryMetrics(metricsProvider)),
-			grpclogging.UnaryServerInterceptor(flogging.MustGetLogger("comm.grpc.server").Zap()),
-			throttle.UnaryServerIntercptor,
-		)
-		serverConfig.StreamInterceptors = append(
-			serverConfig.StreamInterceptors,
-			grpcmetrics.StreamServerInterceptor(grpcmetrics.NewStreamMetrics(metricsProvider)),
-			grpclogging.StreamServerInterceptor(flogging.MustGetLogger("comm.grpc.server").Zap()),
-			throttle.StreamServerInterceptor,
-		)
+		serverConfig.ServerStatsHandler = baseServerConfig.ServerStatsHandler
+		serverConfig.UnaryInterceptors = baseServerConfig.UnaryInterceptors
+		serverConfig.StreamInterceptors = baseServerConfig.StreamInterceptors
+
 		adminServer, err := peer.NewPeerServer(adminListenAddress, serverConfig)
 		if err != nil {
 			logger.Fatalf("Failed to create admin server (%s)", err)