[FAB-12915] gossip leader election metrics

This change set introduces metrics for gossip leader election.

- Leader declaration event - gauge, 0 or 1 according
to whether the peer is the leader or not

Change-Id: I14e3ffbc77c425fbf9a5575e1a07802eda586dda
Signed-off-by: Hagar Meir <hagar.meir@ibm.com>
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: HagarMeir

docs/source/metrics_reference.rst

@@ -119,6 +119,8 @@ The following metrics are currently exported for consumption by Prometheus.
 +-----------------------------------------------------+-----------+------------------------------------------------------------+--------------------+
 | fabric_version                                      | gauge     | The active version of Fabric.                              | version            |
 +-----------------------------------------------------+-----------+------------------------------------------------------------+--------------------+
+| gossip_leader_election_leader                       | gauge     | Peer is leader (1) or follower (0)                         | channel            |
++-----------------------------------------------------+-----------+------------------------------------------------------------+--------------------+
 | gossip_payload_buffer_size                          | gauge     | Size of the payload buffer                                 | channel            |
 +-----------------------------------------------------+-----------+------------------------------------------------------------+--------------------+
 | gossip_state_commit_duration                        | histogram | Time it takes to commit a block in seconds                 | channel            |
@@ -278,6 +280,8 @@ associated with the metric.
 +-----------------------------------------------------------------------------------------+-----------+------------------------------------------------------------+
 | fabric_version.%{version}                                                               | gauge     | The active version of Fabric.                              |
 +-----------------------------------------------------------------------------------------+-----------+------------------------------------------------------------+
+| gossip.leader_election.leader.%{channel}                                                | gauge     | Peer is leader (1) or follower (0)                         |
++-----------------------------------------------------------------------------------------+-----------+------------------------------------------------------------+
 | gossip.payload_buffer.size.%{channel}                                                   | gauge     | Size of the payload buffer                                 |
 +-----------------------------------------------------------------------------------------+-----------+------------------------------------------------------------+
 | gossip.state.commit_duration.%{channel}                                                 | histogram | Time it takes to commit a block in seconds                 |

gossip/election/adapter.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
+	"github.com/hyperledger/fabric/gossip/metrics"
 	"github.com/hyperledger/fabric/gossip/util"
 	proto "github.com/hyperledger/fabric/protos/gossip"
 )
@@ -68,10 +69,12 @@ type adapterImpl struct {
 
 	doneCh   chan struct{}
 	stopOnce *sync.Once
+	metrics  *metrics.ElectionMetrics
 }
 
 // NewAdapter creates new leader election adapter
-func NewAdapter(gossip gossip, pkiid common.PKIidType, channel common.ChainID) LeaderElectionAdapter {
+func NewAdapter(gossip gossip, pkiid common.PKIidType, channel common.ChainID,
+	metrics *metrics.ElectionMetrics) LeaderElectionAdapter {
 	return &adapterImpl{
 		gossip:    gossip,
 		selfPKIid: pkiid,
@@ -85,6 +88,7 @@ func NewAdapter(gossip gossip, pkiid common.PKIidType, channel common.ChainID) L
 
 		doneCh:   make(chan struct{}),
 		stopOnce: &sync.Once{},
+		metrics:  metrics,
 	}
 }
 
@@ -152,6 +156,14 @@ func (ai *adapterImpl) Peers() []Peer {
 	return res
 }
 
+func (ai *adapterImpl) ReportMetrics(isLeader bool) {
+	var leadershipBit float64
+	if isLeader {
+		leadershipBit = 1
+	}
+	ai.metrics.Declaration.With("channel", string(ai.channel)).Set(leadershipBit)
+}
+
 func (ai *adapterImpl) Stop() {
 	stopFunc := func() {
 		close(ai.doneCh)

gossip/election/adapter_test.go

@@ -14,10 +14,15 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hyperledger/fabric/common/metrics"
+	"github.com/hyperledger/fabric/common/metrics/disabled"
+	"github.com/hyperledger/fabric/common/metrics/metricsfakes"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
+	gossipMetrics "github.com/hyperledger/fabric/gossip/metrics"
 	"github.com/hyperledger/fabric/gossip/util"
 	proto "github.com/hyperledger/fabric/protos/gossip"
+	"github.com/stretchr/testify/assert"
 )
 
 func init() {
@@ -35,7 +40,8 @@ func TestNewAdapter(t *testing.T) {
 	peersCluster := newClusterOfPeers("0")
 	peersCluster.addPeer("peer0", mockGossip)
 
-	NewAdapter(mockGossip, selfNetworkMember.PKIid, []byte("channel0"))
+	NewAdapter(mockGossip, selfNetworkMember.PKIid, []byte("channel0"),
+		gossipMetrics.NewGossipMetrics(&disabled.Provider{}).ElectionMetrics)
 }
 
 func TestAdapterImpl_CreateMessage(t *testing.T) {
@@ -46,7 +52,8 @@ func TestAdapterImpl_CreateMessage(t *testing.T) {
 	}
 	mockGossip := newGossip("peer0", selfNetworkMember)
 
-	adapter := NewAdapter(mockGossip, selfNetworkMember.PKIid, []byte("channel0"))
+	adapter := NewAdapter(mockGossip, selfNetworkMember.PKIid, []byte("channel0"),
+		gossipMetrics.NewGossipMetrics(&disabled.Provider{}).ElectionMetrics)
 	msg := adapter.CreateMessage(true)
 
 	if !msg.(*msgImpl).msg.IsLeadershipMsg() {
@@ -281,10 +288,78 @@ func createCluster(peers ...int) (*clusterOfPeers, map[string]*adapterImpl) {
 		}
 
 		mockGossip := newGossip(peerEndpoint, peerMember)
-		adapter := NewAdapter(mockGossip, peerMember.PKIid, []byte("channel0"))
+		adapter := NewAdapter(mockGossip, peerMember.PKIid, []byte("channel0"),
+			gossipMetrics.NewGossipMetrics(&disabled.Provider{}).ElectionMetrics)
 		adapters[peerEndpoint] = adapter.(*adapterImpl)
 		cluster.addPeer(peerEndpoint, mockGossip)
 	}
 
 	return cluster, adapters
 }
+
+type testMetricProvider struct {
+	fakeProvider         *metricsfakes.Provider
+	fakeDeclarationGauge *metricsfakes.Gauge
+}
+
+func testUtilConstructMetricProvider() *testMetricProvider {
+	fakeProvider := &metricsfakes.Provider{}
+	fakeDeclarationGauge := testUtilConstructGauge()
+
+	fakeProvider.NewCounterStub = func(opts metrics.CounterOpts) metrics.Counter {
+		return nil
+	}
+	fakeProvider.NewHistogramStub = func(opts metrics.HistogramOpts) metrics.Histogram {
+		return nil
+	}
+	fakeProvider.NewGaugeStub = func(opts metrics.GaugeOpts) metrics.Gauge {
+		switch opts.Name {
+		case gossipMetrics.LeaderDeclerationOpts.Name:
+			return fakeDeclarationGauge
+		}
+		return nil
+	}
+
+	return &testMetricProvider{
+		fakeProvider,
+		fakeDeclarationGauge,
+	}
+}
+
+func testUtilConstructGauge() *metricsfakes.Gauge {
+	fakeGauge := &metricsfakes.Gauge{}
+	fakeGauge.WithReturns(fakeGauge)
+	return fakeGauge
+}
+
+func TestReportMetrics(t *testing.T) {
+
+	testMetricProvider := testUtilConstructMetricProvider()
+	electionMetrics := gossipMetrics.NewGossipMetrics(testMetricProvider.fakeProvider).ElectionMetrics
+
+	mockGossip := newGossip("", &discovery.NetworkMember{})
+	adapter := NewAdapter(mockGossip, nil, []byte("channel0"), electionMetrics)
+
+	adapter.ReportMetrics(true)
+
+	assert.Equal(t,
+		[]string{"channel", "channel0"},
+		testMetricProvider.fakeDeclarationGauge.WithArgsForCall(0),
+	)
+	assert.EqualValues(t,
+		1,
+		testMetricProvider.fakeDeclarationGauge.SetArgsForCall(0),
+	)
+
+	adapter.ReportMetrics(false)
+
+	assert.Equal(t,
+		[]string{"channel", "channel0"},
+		testMetricProvider.fakeDeclarationGauge.WithArgsForCall(1),
+	)
+	assert.EqualValues(t,
+		0,
+		testMetricProvider.fakeDeclarationGauge.SetArgsForCall(1),
+	)
+
+}

gossip/election/election.go

@@ -82,6 +82,9 @@ type LeaderElectionAdapter interface {
 
 	// Peers returns a list of peers considered alive
 	Peers() []Peer
+
+	// ReportMetrics sends a report to the metrics server about a leadership status
+	ReportMetrics(isLeader bool)
 }
 
 type leadershipCallback func(isLeader bool)
@@ -319,6 +322,7 @@ func (le *leaderElectionSvcImpl) follower() {
 
 	le.proposals.Clear()
 	atomic.StoreInt32(&le.leaderExists, int32(0))
+	le.adapter.ReportMetrics(false)
 	select {
 	case <-time.After(getLeaderAliveThreshold()):
 	case <-le.stopChan:
@@ -329,6 +333,7 @@ func (le *leaderElectionSvcImpl) follower() {
 func (le *leaderElectionSvcImpl) leader() {
 	leaderDeclaration := le.adapter.CreateMessage(true)
 	le.adapter.Gossip(leaderDeclaration)
+	le.adapter.ReportMetrics(true)
 	le.waitForInterrupt(getLeadershipDeclarationInterval())
 }
 

gossip/election/election_test.go

@@ -123,6 +123,10 @@ func (p *peer) Peers() []Peer {
 	return peers
 }
 
+func (p *peer) ReportMetrics(isLeader bool) {
+	p.Mock.Called(isLeader)
+}
+
 func (p *peer) leaderCallback(isLeader bool) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
@@ -160,6 +164,7 @@ func createPeer(id int, peerMap map[string]*peer, l *sync.RWMutex) *peer {
 	idStr := fmt.Sprintf("p%d", id)
 	c := make(chan Msg, 100)
 	p := &peer{id: idStr, peers: peerMap, sharedLock: l, msgChan: c, mockedMethods: make(map[string]struct{}), leaderFromCallback: false, callbackInvoked: false}
+	p.On("ReportMetrics", mock.Anything, mock.Anything)
 	p.LeaderElectionService = NewLeaderElectionService(p, idStr, p.leaderCallback)
 	l.Lock()
 	peerMap[idStr] = p
@@ -190,6 +195,21 @@ func waitForLeaderElection(t *testing.T, peers []*peer) []string {
 	return waitForMultipleLeadersElection(t, peers, 1)
 }
 
+func TestMetrics(t *testing.T) {
+	t.Parallel()
+	// Scenario: spawn a single peer and ensure it reports being a leader after some time.
+	// Then, make it relinquish its leadership and then ensure it reports not being a leader.
+	p := createPeer(0, make(map[string]*peer), &sync.RWMutex{})
+	waitForLeaderElection(t, []*peer{p})
+	// Ensure we sent a leadership declaration during the time of leadership acquisition.
+	p.AssertCalled(t, "ReportMetrics", true)
+	p.Yield()
+	assert.False(t, p.IsLeader())
+	waitForLeaderElection(t, []*peer{p})
+	// Ensure declaration for not being a leader was sent
+	p.AssertCalled(t, "ReportMetrics", false)
+}
+
 func TestInitPeersAtSameTime(t *testing.T) {
 	t.Parallel()
 	// Scenario: Peers are spawned at the same time

gossip/metrics/metrics.go

@@ -10,12 +10,14 @@ import "github.com/hyperledger/fabric/common/metrics"
 
 // GossipMetrics encapsulates all of gossip metrics
 type GossipMetrics struct {
-	StateMetrics *StateMetrics
+	StateMetrics    *StateMetrics
+	ElectionMetrics *ElectionMetrics
 }
 
 func NewGossipMetrics(p metrics.Provider) *GossipMetrics {
 	return &GossipMetrics{
-		StateMetrics: newStateMetrics(p),
+		StateMetrics:    newStateMetrics(p),
+		ElectionMetrics: newElectionMetrics(p),
 	}
 }
 
@@ -62,3 +64,25 @@ var (
 		StatsdFormat: "%{#fqname}.%{channel}",
 	}
 )
+
+// ElectionMetrics encapsulates gossip leader election related metrics
+type ElectionMetrics struct {
+	Declaration metrics.Gauge
+}
+
+func newElectionMetrics(p metrics.Provider) *ElectionMetrics {
+	return &ElectionMetrics{
+		Declaration: p.NewGauge(LeaderDeclerationOpts),
+	}
+}
+
+var (
+	LeaderDeclerationOpts = metrics.GaugeOpts{
+		Namespace:    "gossip",
+		Subsystem:    "leader_election",
+		Name:         "leader",
+		Help:         "Peer is leader (1) or follower (0)",
+		LabelNames:   []string{"channel"},
+		StatsdFormat: "%{#fqname}.%{channel}",
+	}
+)

gossip/metrics/metrics_test.go

@@ -29,9 +29,13 @@ func TestMetrics(t *testing.T) {
 
 	// make sure all metrics were created
 	assert.NotNil(t, gossipMetrics)
+
 	assert.NotNil(t, gossipMetrics.StateMetrics)
 	assert.NotNil(t, gossipMetrics.StateMetrics.Height)
 	assert.NotNil(t, gossipMetrics.StateMetrics.CommitDuration)
 	assert.NotNil(t, gossipMetrics.StateMetrics.PayloadBufferSize)
 
+	assert.NotNil(t, gossipMetrics.ElectionMetrics)
+	assert.NotNil(t, gossipMetrics.ElectionMetrics.Declaration)
+
 }

gossip/service/gossip_service.go

@@ -294,7 +294,8 @@ func (g *gossipServiceImpl) InitializeChannel(chainID string, endpoints []string
 
 		if leaderElection {
 			logger.Debug("Delivery uses dynamic leader election mechanism, channel", chainID)
-			g.leaderElection[chainID] = g.newLeaderElectionComponent(chainID, g.onStatusChangeFactory(chainID, support.Committer))
+			g.leaderElection[chainID] = g.newLeaderElectionComponent(chainID, g.onStatusChangeFactory(chainID,
+				support.Committer), g.metrics.ElectionMetrics)
 		} else if isStaticOrgLeader {
 			logger.Debug("This peer is configured to connect to ordering service for blocks delivery, channel", chainID)
 			g.deliveryService[chainID].StartDeliverForChannel(chainID, support.Committer, func() {})
@@ -384,9 +385,10 @@ func (g *gossipServiceImpl) Stop() {
 	g.gossipSvc.Stop()
 }
 
-func (g *gossipServiceImpl) newLeaderElectionComponent(chainID string, callback func(bool)) election.LeaderElectionService {
+func (g *gossipServiceImpl) newLeaderElectionComponent(chainID string, callback func(bool),
+	electionMetrics *gossipMetrics.ElectionMetrics) election.LeaderElectionService {
 	PKIid := g.mcs.GetPKIidOfCert(g.peerIdentity)
-	adapter := election.NewAdapter(g, PKIid, gossipCommon.ChainID(chainID))
+	adapter := election.NewAdapter(g, PKIid, gossipCommon.ChainID(chainID), electionMetrics)
 	return election.NewLeaderElectionService(adapter, string(PKIid), callback)
 }
 

gossip/service/gossip_service_test.go

@@ -409,9 +409,12 @@ func TestLeaderElectionWithRealGossip(t *testing.T) {
 	//Starting leader election services
 	services := make([]*electionService, n)
 
+	electionMetrics := gossipMetrics.NewGossipMetrics(&disabled.Provider{}).ElectionMetrics
+
 	for i := 0; i < n; i++ {
 		services[i] = &electionService{nil, false, 0}
-		services[i].LeaderElectionService = gossips[i].(*gossipServiceImpl).newLeaderElectionComponent(channelName, services[i].callback)
+		services[i].LeaderElectionService = gossips[i].(*gossipServiceImpl).newLeaderElectionComponent(channelName,
+			services[i].callback, electionMetrics)
 	}
 
 	logger.Warning("Waiting for leader election")
@@ -437,7 +440,9 @@ func TestLeaderElectionWithRealGossip(t *testing.T) {
 
 	for idx, i := range secondChannelPeerIndexes {
 		secondChannelServices[idx] = &electionService{nil, false, 0}
-		secondChannelServices[idx].LeaderElectionService = gossips[i].(*gossipServiceImpl).newLeaderElectionComponent(secondChannelName, secondChannelServices[idx].callback)
+		secondChannelServices[idx].LeaderElectionService =
+			gossips[i].(*gossipServiceImpl).newLeaderElectionComponent(secondChannelName,
+				secondChannelServices[idx].callback, electionMetrics)
 	}
 
 	assert.True(t, waitForLeaderElection(t, secondChannelServices, time.Second*30, time.Second*2), "One leader should be selected for chanB")

gossip/state/metrics_test.go

@@ -29,22 +29,20 @@ type testMetricProvider struct {
 	fakePayloadBufferSizeGauge *metricsfakes.Gauge
 }
 
-func testUtilConstructMetricProvider(t *testing.T) *testMetricProvider {
+func testUtilConstructMetricProvider() *testMetricProvider {
 	fakeProvider := &metricsfakes.Provider{}
 	fakeHeightGauge := testUtilConstructGuage()
 	fakeCommitDurationHist := testUtilConstructHist()
 	fakePayloadBufferSizeGauge := testUtilConstructGuage()
 
 	fakeProvider.NewCounterStub = func(opts metrics.CounterOpts) metrics.Counter {
-		assert.Fail(t, "Unknown counter name: ", opts.Name)
 		return nil
 	}
 	fakeProvider.NewHistogramStub = func(opts metrics.HistogramOpts) metrics.Histogram {
 		switch opts.Name {
 		case gossipMetrics.CommitDurationOpts.Name:
 			return fakeCommitDurationHist
 		}
-		assert.Fail(t, "Unknown histogram name: ", opts.Name)
 		return nil
 	}
 	fakeProvider.NewGaugeStub = func(opts metrics.GaugeOpts) metrics.Gauge {
@@ -54,7 +52,6 @@ func testUtilConstructMetricProvider(t *testing.T) *testMetricProvider {
 		case gossipMetrics.HeightOpts.Name:
 			return fakeHeightGauge
 		}
-		assert.Fail(t, "Unknown gauge name: ", opts.Name)
 		return nil
 	}
 
@@ -93,7 +90,7 @@ func TestMetrics(t *testing.T) {
 	committedDurationWG := sync.WaitGroup{}
 	committedDurationWG.Add(1)
 
-	testMetricProvider := testUtilConstructMetricProvider(t)
+	testMetricProvider := testUtilConstructMetricProvider()
 
 	testMetricProvider.fakeHeightGauge.SetStub = func(delta float64) {
 		heightWG.Done()