FAB-613 Add Kafka metrics for orderer

This CR adds a simple polling wrapper around the go-metrics metric
registry logic that Sarama already implements.  It converts all of the
exposed per broker/topic metrics to gauges and exposes those gauges
through the go-kit metrics.

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

Author: Jason Yellick

Gopkg.lock

@@ -61,7 +61,7 @@ type CounterOpts struct {
 	StatsdFormat string
 }
 
-// A Gauge is a memter that expresses the current value of some metric.
+// A Gauge is a meter that expresses the current value of some metric.
 type Gauge interface {
 	// With is used to provide label values when recording a Gauge value. This
 	// must be used to provide values for all LabelNames provided to GaugeOpts.
@@ -108,7 +108,7 @@ type GaugeOpts struct {
 	StatsdFormat string
 }
 
-// A Histogram is a memter that records an observed value into quantized
+// A Histogram is a meter that records an observed value into quantized
 // buckets.
 type Histogram interface {
 	// With is used to provide label values when recording a Histogram

common/metrics/provider.go

@@ -126,7 +126,7 @@ func Start(cmd string, conf *localconfig.TopLevel) {
 		}
 	}
 
-	manager := initializeMultichannelRegistrar(clusterType, clusterDialer, serverConfig, grpcServer, conf, signer, tlsCallback)
+	manager := initializeMultichannelRegistrar(clusterType, clusterDialer, serverConfig, grpcServer, conf, signer, metricsProvider, tlsCallback)
 	mutualTLS := serverConfig.SecOpts.UseTLS && serverConfig.SecOpts.RequireClientCert
 	server := NewServer(manager, metricsProvider, &conf.Debug, conf.General.Authentication.TimeWindow, mutualTLS)
 
@@ -397,6 +397,7 @@ func initializeMultichannelRegistrar(isClusterType bool,
 	srv *comm.GRPCServer,
 	conf *localconfig.TopLevel,
 	signer crypto.LocalSigner,
+	metricsProvider metrics.Provider,
 	callbacks ...func(bundle *channelconfig.Bundle)) *multichannel.Registrar {
 	lf, _ := createLedgerFactory(conf)
 	genesisBlock := extractGenesisBlock(conf)
@@ -412,7 +413,11 @@ func initializeMultichannelRegistrar(isClusterType bool,
 	registrar := multichannel.NewRegistrar(lf, signer, callbacks...)
 
 	consenters["solo"] = solo.New()
-	consenters["kafka"] = kafka.New(conf.Kafka)
+	var kafkaMetrics *kafka.Metrics
+	consenters["kafka"], kafkaMetrics = kafka.New(conf.Kafka, metricsProvider)
+	// Note, we pass a 'nil' channel here, we could pass a channel that
+	// closes if we wished to cleanup this routine on exit.
+	go kafkaMetrics.PollGoMetricsUntilStop(time.Minute, nil)
 	if isClusterType {
 		raftConsenter := etcdraft.New(clusterDialer, conf, srvConf, srv, registrar)
 		consenters["etcdraft"] = raftConsenter

orderer/common/server/main.go

@@ -259,7 +259,7 @@ func TestInitializeMultiChainManager(t *testing.T) {
 	conf := genesisConfig(t)
 	assert.NotPanics(t, func() {
 		initializeLocalMsp(conf)
-		initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, conf, localmsp.NewSigner())
+		initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, conf, localmsp.NewSigner(), &disabled.Provider{})
 	})
 }
 
@@ -320,7 +320,7 @@ func TestUpdateTrustedRoots(t *testing.T) {
 			updateTrustedRoots(grpcServer, caSupport, bundle)
 		}
 	}
-	initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, genesisConfig(t), localmsp.NewSigner(), callback)
+	initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, genesisConfig(t), localmsp.NewSigner(), &disabled.Provider{}, callback)
 	t.Logf("# app CAs: %d", len(caSupport.AppRootCAsByChain[genesisconfig.TestChainID]))
 	t.Logf("# orderer CAs: %d", len(caSupport.OrdererRootCAsByChain[genesisconfig.TestChainID]))
 	// mutual TLS not required so no updates should have occurred
@@ -357,7 +357,7 @@ func TestUpdateTrustedRoots(t *testing.T) {
 			updateClusterDialer(caSupport, predDialer, clusterConf.SecOpts.ServerRootCAs)
 		}
 	}
-	initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, genesisConfig(t), localmsp.NewSigner(), callback)
+	initializeMultichannelRegistrar(false, &cluster.PredicateDialer{}, comm.ServerConfig{}, nil, genesisConfig(t), localmsp.NewSigner(), &disabled.Provider{}, callback)
 	t.Logf("# app CAs: %d", len(caSupport.AppRootCAsByChain[genesisconfig.TestChainID]))
 	t.Logf("# orderer CAs: %d", len(caSupport.OrdererRootCAsByChain[genesisconfig.TestChainID]))
 	// mutual TLS is required so updates should have occurred

orderer/common/server/main_test.go

@@ -11,18 +11,20 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Shopify/sarama"
-	"github.com/Shopify/sarama/mocks"
-	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/channelconfig"
 	mockconfig "github.com/hyperledger/fabric/common/mocks/config"
 	"github.com/hyperledger/fabric/orderer/common/blockcutter"
 	"github.com/hyperledger/fabric/orderer/common/msgprocessor"
+	lmock "github.com/hyperledger/fabric/orderer/consensus/kafka/mock"
 	mockblockcutter "github.com/hyperledger/fabric/orderer/mocks/common/blockcutter"
 	mockmultichannel "github.com/hyperledger/fabric/orderer/mocks/common/multichannel"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/Shopify/sarama"
+	"github.com/Shopify/sarama/mocks"
+	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -3402,7 +3404,7 @@ func TestDeliverSession(t *testing.T) {
 		defer env.broker2.Close()
 
 		// initialize consenter
-		consenter := New(mockLocalConfig.Kafka)
+		consenter, _ := New(mockLocalConfig.Kafka, &lmock.MetricsProvider{})
 
 		// initialize chain
 		metadata := &cb.Metadata{Value: utils.MarshalOrPanic(&ab.KafkaMetadata{LastOffsetPersisted: env.height})}
@@ -3491,7 +3493,7 @@ func TestDeliverSession(t *testing.T) {
 		defer env.broker0.Close()
 
 		// initialize consenter
-		consenter := New(mockLocalConfig.Kafka)
+		consenter, _ := New(mockLocalConfig.Kafka, &lmock.MetricsProvider{})
 
 		// initialize chain
 		metadata := &cb.Metadata{Value: utils.MarshalOrPanic(&ab.KafkaMetadata{LastOffsetPersisted: env.height})}
@@ -3553,7 +3555,7 @@ func TestDeliverSession(t *testing.T) {
 		defer env.broker0.Close()
 
 		// initialize consenter
-		consenter := New(mockLocalConfig.Kafka)
+		consenter, _ := New(mockLocalConfig.Kafka, &lmock.MetricsProvider{})
 
 		// initialize chain
 		metadata := &cb.Metadata{Value: utils.MarshalOrPanic(&ab.KafkaMetadata{LastOffsetPersisted: env.height})}

orderer/consensus/kafka/chain_test.go

@@ -10,8 +10,9 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 
-	"github.com/Shopify/sarama"
 	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
+
+	"github.com/Shopify/sarama"
 )
 
 func newBrokerConfig(

orderer/consensus/kafka/config.go

@@ -7,24 +7,28 @@ SPDX-License-Identifier: Apache-2.0
 package kafka
 
 import (
-	"github.com/Shopify/sarama"
+	"github.com/hyperledger/fabric/common/metrics"
 	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
 	"github.com/hyperledger/fabric/orderer/consensus"
 	cb "github.com/hyperledger/fabric/protos/common"
+
+	"github.com/Shopify/sarama"
 	logging "github.com/op/go-logging"
 )
 
 // New creates a Kafka-based consenter. Called by orderer's main.go.
-func New(config localconfig.Kafka) consensus.Consenter {
+func New(config localconfig.Kafka, metricsProvider metrics.Provider) (consensus.Consenter, *Metrics) {
 	if config.Verbose {
 		logging.SetLevel(logging.DEBUG, "orderer.consensus.kafka.sarama")
 	}
+
 	brokerConfig := newBrokerConfig(
 		config.TLS,
 		config.SASLPlain,
 		config.Retry,
 		config.Version,
 		defaultPartition)
+
 	return &consenterImpl{
 		brokerConfigVal: brokerConfig,
 		tlsConfigVal:    config.TLS,
@@ -34,7 +38,7 @@ func New(config localconfig.Kafka) consensus.Consenter {
 			NumPartitions:     1,
 			ReplicationFactor: config.Topic.ReplicationFactor,
 		},
-	}
+	}, NewMetrics(metricsProvider, brokerConfig.MetricRegistry)
 }
 
 // consenterImpl holds the implementation of type that satisfies the
@@ -46,6 +50,7 @@ type consenterImpl struct {
 	retryOptionsVal localconfig.Retry
 	kafkaVersionVal sarama.KafkaVersion
 	topicDetailVal  *sarama.TopicDetail
+	metricsProvider metrics.Provider
 }
 
 // HandleChain creates/returns a reference to a consensus.Chain object for the

orderer/consensus/kafka/consenter.go

@@ -12,16 +12,18 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Shopify/sarama"
-	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/flogging"
 	mockconfig "github.com/hyperledger/fabric/common/mocks/config"
 	localconfig "github.com/hyperledger/fabric/orderer/common/localconfig"
 	"github.com/hyperledger/fabric/orderer/consensus"
+	"github.com/hyperledger/fabric/orderer/consensus/kafka/mock"
 	mockmultichannel "github.com/hyperledger/fabric/orderer/mocks/common/multichannel"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 	"github.com/hyperledger/fabric/protos/utils"
+
+	"github.com/Shopify/sarama"
+	"github.com/golang/protobuf/proto"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -69,11 +71,12 @@ func init() {
 }
 
 func TestNew(t *testing.T) {
-	_ = consensus.Consenter(New(mockLocalConfig.Kafka))
+	c, _ := New(mockLocalConfig.Kafka, &mock.MetricsProvider{})
+	_ = consensus.Consenter(c)
 }
 
 func TestHandleChain(t *testing.T) {
-	consenter := consensus.Consenter(New(mockLocalConfig.Kafka))
+	consenter, _ := New(mockLocalConfig.Kafka, &mock.MetricsProvider{})
 
 	oldestOffset := int64(0)
 	newestOffset := int64(5)
@@ -103,7 +106,6 @@ func TestHandleChain(t *testing.T) {
 	}
 
 	mockMetadata := &cb.Metadata{Value: utils.MarshalOrPanic(&ab.KafkaMetadata{LastOffsetPersisted: newestOffset - 1})}
-
 	_, err := consenter.HandleChain(mockSupport, mockMetadata)
 	assert.NoError(t, err, "Expected the HandleChain call to return without errors")
 }

orderer/consensus/kafka/consenter_test.go

@@ -0,0 +1,48 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package kafka_test
+
+import (
+	"testing"
+
+	"github.com/hyperledger/fabric/common/metrics"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	gometrics "github.com/rcrowley/go-metrics"
+)
+
+//go:generate counterfeiter -o mock/metrics_registry.go --fake-name MetricsRegistry . metricsRegistry
+type metricsRegistry interface {
+	gometrics.Registry
+}
+
+//go:generate counterfeiter -o mock/metrics_meter.go --fake-name MetricsMeter . metricsMeter
+type metricsMeter interface {
+	gometrics.Meter
+}
+
+//go:generate counterfeiter -o mock/metrics_histogram.go --fake-name MetricsHistogram . metricsHistogram
+type metricsHistogram interface {
+	gometrics.Histogram
+}
+
+//go:generate counterfeiter -o mock/metrics_provider.go --fake-name MetricsProvider . metricsProvider
+type metricsProvider interface {
+	metrics.Provider
+}
+
+//go:generate counterfeiter -o mock/metrics_gauge.go --fake-name MetricsGauge . metricsGauge
+type metricsGauge interface {
+	metrics.Gauge
+}
+
+func TestKafka(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Kafka Suite")
+}