[FAB-17539] Always remember anchor peers in membership (hyperledger#1422

)

Gossip service removes a peer from its membership if the peer's alive message is expired.
However, it should always remember the anchor peers and bootstrap peers in order for
the peer to reconnect. Gossip already remembers bootstrap peers. This PR adds code to track
all anchor peers' endpoints and updates the expiration callback function to not delete
anchor peers.

Signed-off-by: Wenjian Qiao <wenjianq@gmail.com>

gossip/discovery/discovery.go

@@ -73,6 +73,11 @@ type CommService interface {
 	IdentitySwitch() <-chan common.PKIidType
 }
 
+// AnchorPeerTracker is an interface that is passed to discovery to check if an endpoint is an anchor peer
+type AnchorPeerTracker interface {
+	IsAnchorPeer(endpoint string) bool
+}
+
 // NetworkMember is a peer's representation
 type NetworkMember struct {
 	Endpoint         string

gossip/discovery/discovery_impl.go

@@ -27,15 +27,8 @@ const DefAliveTimeInterval = 5 * time.Second
 const DefAliveExpirationTimeout = 5 * DefAliveTimeInterval
 const DefAliveExpirationCheckInterval = DefAliveExpirationTimeout / 10
 const DefReconnectInterval = DefAliveExpirationTimeout
-const msgExpirationFactor = 20
-
-var maxConnectionAttempts = 120
-
-// SetMaxConnAttempts sets the maximum number of connection
-// attempts the peer would perform when invoking Connect()
-func SetMaxConnAttempts(attempts int) {
-	maxConnectionAttempts = attempts
-}
+const DefMsgExpirationFactor = 20
+const DefMaxConnectionAttempts = 120
 
 type timestamp struct {
 	incTime  time.Time
@@ -74,21 +67,26 @@ type gossipDiscoveryImpl struct {
 	aliveExpirationTimeout       time.Duration
 	aliveExpirationCheckInterval time.Duration
 	reconnectInterval            time.Duration
+	msgExpirationFactor          int
+	maxConnectionAttempts        int
 
-	bootstrapPeers []string
+	bootstrapPeers    []string
+	anchorPeerTracker AnchorPeerTracker
 }
 
 type DiscoveryConfig struct {
 	AliveTimeInterval            time.Duration
 	AliveExpirationTimeout       time.Duration
 	AliveExpirationCheckInterval time.Duration
 	ReconnectInterval            time.Duration
+	MaxConnectionAttempts        int
+	MsgExpirationFactor          int
 	BootstrapPeers               []string
 }
 
 // NewDiscoveryService returns a new discovery service with the comm module passed and the crypto service passed
 func NewDiscoveryService(self NetworkMember, comm CommService, crypt CryptoService, disPol DisclosurePolicy,
-	config DiscoveryConfig) Discovery {
+	config DiscoveryConfig, anchorPeerTracker AnchorPeerTracker) Discovery {
 	d := &gossipDiscoveryImpl{
 		self:             self,
 		incTime:          uint64(time.Now().UnixNano()),
@@ -110,8 +108,11 @@ func NewDiscoveryService(self NetworkMember, comm CommService, crypt CryptoServi
 		aliveExpirationTimeout:       config.AliveExpirationTimeout,
 		aliveExpirationCheckInterval: config.AliveExpirationCheckInterval,
 		reconnectInterval:            config.ReconnectInterval,
+		maxConnectionAttempts:        config.MaxConnectionAttempts,
+		msgExpirationFactor:          config.MsgExpirationFactor,
 
-		bootstrapPeers: config.BootstrapPeers,
+		bootstrapPeers:    config.BootstrapPeers,
+		anchorPeerTracker: anchorPeerTracker,
 	}
 
 	d.validateSelfConfig()
@@ -147,7 +148,7 @@ func (d *gossipDiscoveryImpl) Connect(member NetworkMember, id identifier) {
 	d.logger.Debug("Entering", member)
 	defer d.logger.Debug("Exiting")
 	go func() {
-		for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
+		for i := 0; i < d.maxConnectionAttempts && !d.toDie(); i++ {
 			id, err := id()
 			if err != nil {
 				if d.toDie() {
@@ -214,7 +215,7 @@ func (d *gossipDiscoveryImpl) validateSelfConfig() {
 
 func (d *gossipDiscoveryImpl) sendUntilAcked(peer *NetworkMember, message *protoext.SignedGossipMessage) {
 	nonce := message.Nonce
-	for i := 0; i < maxConnectionAttempts && !d.toDie(); i++ {
+	for i := 0; i < d.maxConnectionAttempts && !d.toDie(); i++ {
 		sub := d.pubsub.Subscribe(fmt.Sprintf("%d", nonce), time.Second*5)
 		d.comm.SendToPeer(peer, message)
 		if _, timeoutErr := sub.Listen(); timeoutErr == nil {
@@ -1040,7 +1041,7 @@ type aliveMsgStore struct {
 func newAliveMsgStore(d *gossipDiscoveryImpl) *aliveMsgStore {
 	policy := protoext.NewGossipMessageComparator(0)
 	trigger := func(m interface{}) {}
-	aliveMsgTTL := d.aliveExpirationTimeout * msgExpirationFactor
+	aliveMsgTTL := d.aliveExpirationTimeout * time.Duration(d.msgExpirationFactor)
 	externalLock := func() { d.lock.Lock() }
 	externalUnlock := func() { d.lock.Unlock() }
 	callback := func(m interface{}) {
@@ -1052,8 +1053,10 @@ func newAliveMsgStore(d *gossipDiscoveryImpl) *aliveMsgStore {
 		id := membership.PkiId
 		endpoint := membership.Endpoint
 		internalEndpoint := protoext.InternalEndpoint(msg.SecretEnvelope)
-		if util.Contains(endpoint, d.bootstrapPeers) || util.Contains(internalEndpoint, d.bootstrapPeers) {
-			// Never remove a bootstrap peer
+		if util.Contains(endpoint, d.bootstrapPeers) || util.Contains(internalEndpoint, d.bootstrapPeers) ||
+			d.anchorPeerTracker.IsAnchorPeer(endpoint) || d.anchorPeerTracker.IsAnchorPeer(internalEndpoint) {
+			// Never remove a bootstrap peer or an anchor peer
+			d.logger.Debugf("Do not remove bootstrap or anchor peer endpoint %s from membership", endpoint)
 			return
 		}
 		d.logger.Infof("Removing member: Endpoint: %s, InternalEndpoint: %s, PKIID: %x", endpoint, internalEndpoint, id)

gossip/discovery/discovery_test.go

@@ -44,11 +44,13 @@ var defaultTestConfig = DiscoveryConfig{
 	AliveExpirationTimeout:       10 * aliveTimeInterval,
 	AliveExpirationCheckInterval: aliveTimeInterval,
 	ReconnectInterval:            10 * aliveTimeInterval,
+	MaxConnectionAttempts:        DefMaxConnectionAttempts,
+	MsgExpirationFactor:          DefMsgExpirationFactor,
 }
 
 func init() {
 	util.SetupTestLogging()
-	maxConnectionAttempts = 10000
+	defaultTestConfig.MaxConnectionAttempts = 10000
 }
 
 type dummyReceivedMessage struct {
@@ -76,6 +78,15 @@ func (*dummyReceivedMessage) Ack(err error) {
 	panic("implement me")
 }
 
+// mockAnchorPeerTracker implements AnchorPeerTracker interface
+type mockAnchorPeerTracker struct {
+	apEndpoints []string
+}
+
+func (m *mockAnchorPeerTracker) IsAnchorPeer(endpoint string) bool {
+	return util.Contains(endpoint, m.apEndpoints)
+}
+
 type dummyCommModule struct {
 	validatedMessages chan *protoext.SignedGossipMessage
 	msgsReceived      uint32
@@ -378,6 +389,12 @@ func createDiscoveryInstanceThatGossips(port int, id string, bootstrapPeers []st
 }
 
 func createDiscoveryInstanceThatGossipsWithInterceptors(port int, id string, bootstrapPeers []string, shouldGossip bool, pol DisclosurePolicy, f func(*protoext.SignedGossipMessage), config DiscoveryConfig) *gossipInstance {
+	mockTracker := &mockAnchorPeerTracker{}
+	return createDiscoveryInstanceWithAnchorPeerTracker(port, id, bootstrapPeers, shouldGossip, pol, f, config, mockTracker)
+}
+
+func createDiscoveryInstanceWithAnchorPeerTracker(port int, id string, bootstrapPeers []string, shouldGossip bool, pol DisclosurePolicy,
+	f func(*protoext.SignedGossipMessage), config DiscoveryConfig, anchorPeerTracker AnchorPeerTracker) *gossipInstance {
 	comm := &dummyCommModule{
 		conns:          make(map[string]*grpc.ClientConn),
 		streams:        make(map[string]proto.Gossip_GossipStreamClient),
@@ -409,7 +426,8 @@ func createDiscoveryInstanceThatGossipsWithInterceptors(port int, id string, boo
 	s := grpc.NewServer()
 
 	config.BootstrapPeers = bootstrapPeers
-	discSvc := NewDiscoveryService(self, comm, comm, pol, config)
+
+	discSvc := NewDiscoveryService(self, comm, comm, pol, config, anchorPeerTracker)
 	for _, bootPeer := range bootstrapPeers {
 		bp := bootPeer
 		discSvc.Connect(NetworkMember{Endpoint: bp, InternalEndpoint: bootPeer}, func() (*PeerIdentification, error) {
@@ -1246,7 +1264,7 @@ func TestMsgStoreExpiration(t *testing.T) {
 		return true
 	}
 
-	waitUntilTimeoutOrFail(t, checkMessages, defaultTestConfig.AliveExpirationTimeout*(msgExpirationFactor+5))
+	waitUntilTimeoutOrFail(t, checkMessages, defaultTestConfig.AliveExpirationTimeout*(DefMsgExpirationFactor+5))
 
 	assertMembership(t, instances[:len(instances)-2], nodeNum-3)
 
@@ -1265,12 +1283,14 @@ func TestExpirationNoSecretEnvelope(t *testing.T) {
 		return nil
 	}))
 
+	mockTracker := &mockAnchorPeerTracker{}
 	msgStore := newAliveMsgStore(&gossipDiscoveryImpl{
 		aliveExpirationTimeout: time.Millisecond,
 		lock:                   &sync.RWMutex{},
 		aliveMembership:        util.NewMembershipStore(),
 		deadMembership:         util.NewMembershipStore(),
 		logger:                 logger,
+		anchorPeerTracker:      mockTracker,
 	})
 
 	msg := &proto.GossipMessage{
@@ -1436,7 +1456,7 @@ func TestMsgStoreExpirationWithMembershipMessages(t *testing.T) {
 	}
 
 	// Sleep until expire
-	time.Sleep(defaultTestConfig.AliveExpirationTimeout * (msgExpirationFactor + 5))
+	time.Sleep(defaultTestConfig.AliveExpirationTimeout * (DefMsgExpirationFactor + 5))
 
 	// Checking Alive expired
 	for i := 0; i < peersNum; i++ {
@@ -1672,7 +1692,7 @@ func TestPeerIsolation(t *testing.T) {
 
 	// Sleep the same amount of time as it takes to remove a message from the aliveMsgStore (aliveMsgTTL)
 	// Add a second as buffer
-	time.Sleep(config.AliveExpirationTimeout*msgExpirationFactor + time.Second)
+	time.Sleep(config.AliveExpirationTimeout*DefMsgExpirationFactor + time.Second)
 
 	// Start again the first 2 peers and wait for all the peers to get full membership.
 	// Especially, we want to test that peer2 won't be isolated
@@ -1684,6 +1704,110 @@ func TestPeerIsolation(t *testing.T) {
 	assertMembership(t, instances, peersNum-1)
 }
 
+func TestMembershipAfterExpiration(t *testing.T) {
+	// Scenario:
+	// Start 3 peers (peer0, peer1, peer2). Set peer0 as the anchor peer.
+	// Stop peer0 and peer1 for a while, start them again and test if peer2 still gets full membership
+
+	config := defaultTestConfig
+	// Use a smaller AliveExpirationTimeout than the default to reduce the running time of the test.
+	config.AliveExpirationTimeout = 2 * config.AliveTimeInterval
+	config.ReconnectInterval = config.AliveExpirationTimeout
+	config.MsgExpirationFactor = 5
+
+	peersNum := 3
+	ports := []int{9120, 9121, 9122}
+	anchorPeer := "localhost:9120"
+	bootPeers := []string{}
+	instances := []*gossipInstance{}
+	var inst *gossipInstance
+	mockTracker := &mockAnchorPeerTracker{[]string{anchorPeer}}
+
+	// use a custom logger to verify messages from expiration callback
+	expectedMsgs := []string{
+		"Do not remove bootstrap or anchor peer endpoint localhost:9120 from membership",
+		"Removing member: Endpoint: localhost:9121",
+	}
+	numMsgsFound := 0
+	l, err := zap.NewDevelopment()
+	assert.NoError(t, err)
+	expired := make(chan struct{})
+	logger := flogging.NewFabricLogger(l, zap.Hooks(func(entry zapcore.Entry) error {
+		// do nothing if we already found all the expectedMsgs
+		if numMsgsFound == len(expectedMsgs) {
+			return nil
+		}
+		for _, msg := range expectedMsgs {
+			if strings.Contains(entry.Message, msg) {
+				numMsgsFound++
+				if numMsgsFound == len(expectedMsgs) {
+					expired <- struct{}{}
+				}
+				break
+			}
+		}
+		return nil
+	}))
+
+	// Start all peers, connect to the anchor peer and verify full membership
+	for i := 0; i < peersNum; i++ {
+		id := fmt.Sprintf("d%d", i)
+		inst = createDiscoveryInstanceWithAnchorPeerTracker(ports[i], id, bootPeers, true, noopPolicy, func(_ *protoext.SignedGossipMessage) {}, config, mockTracker)
+		instances = append(instances, inst)
+	}
+	instances[peersNum-1].Discovery.(*gossipDiscoveryImpl).logger = logger
+	for i := 1; i < peersNum; i++ {
+		connect(instances[i], anchorPeer)
+	}
+	assertMembership(t, instances, peersNum-1)
+
+	// Stop peer0 and peer1 so that peer2 would stay alone
+	stopInstances(t, instances[0:peersNum-1])
+
+	// waitTime is the same amount of time as it takes to remove a message from the aliveMsgStore (aliveMsgTTL)
+	// Add a second as buffer
+	waitTime := config.AliveExpirationTimeout*time.Duration(config.MsgExpirationFactor) + time.Second
+	select {
+	case <-expired:
+	case <-time.After(waitTime):
+		t.Fatalf("timed out")
+	}
+	// peer2's deadMembership should contain the anchor peer
+	deadMemeberShip := instances[peersNum-1].discoveryImpl().deadMembership
+	assert.Equal(t, 1, deadMemeberShip.Size())
+	assertMembership(t, instances[peersNum-1:], 0)
+
+	// Start again peer0 and peer1 and wait for all the peers to get full membership.
+	// Especially, we want to test that peer2 won't be isolated
+	for i := 0; i < peersNum-1; i++ {
+		id := fmt.Sprintf("d%d", i)
+		inst = createDiscoveryInstanceWithAnchorPeerTracker(ports[i], id, bootPeers, true, noopPolicy, func(_ *protoext.SignedGossipMessage) {}, config, mockTracker)
+		instances[i] = inst
+	}
+	connect(instances[1], anchorPeer)
+	assertMembership(t, instances, peersNum-1)
+}
+
+func connect(inst *gossipInstance, endpoint string) {
+	inst.comm.lock.Lock()
+	inst.comm.mock = &mock.Mock{}
+	inst.comm.mock.On("SendToPeer", mock.Anything, mock.Anything).Run(func(arguments mock.Arguments) {
+		inst := inst
+		msg := arguments.Get(1).(*protoext.SignedGossipMessage)
+		if req := msg.GetMemReq(); req != nil {
+			inst.comm.lock.Lock()
+			inst.comm.mock = nil
+			inst.comm.lock.Unlock()
+		}
+	})
+	inst.comm.mock.On("Ping", mock.Anything)
+	inst.comm.lock.Unlock()
+	netMember2Connect2 := NetworkMember{Endpoint: endpoint, PKIid: []byte(endpoint)}
+	inst.Connect(netMember2Connect2, func() (identification *PeerIdentification, err error) {
+		return &PeerIdentification{SelfOrg: true, ID: nil}, nil
+	})
+}
+
 func waitUntilOrFail(t *testing.T, pred func() bool) {
 	waitUntilTimeoutOrFail(t, pred, timeout)
 }

gossip/gossip/config.go

@@ -93,6 +93,10 @@ type Config struct {
 	AliveExpirationCheckInterval time.Duration
 	// ReconnectInterval is the Reconnect interval.
 	ReconnectInterval time.Duration
+	// MsgExpirationFactor is the expiration factor for alive message TTL
+	MsgExpirationFactor int
+	// MaxConnectionAttempts is the max number of attempts to connect to a peer (wait for alive ack)
+	MaxConnectionAttempts int
 }
 
 // GlobalConfig builds a Config from the given endpoint, certificate and bootstrap peers.
@@ -142,6 +146,8 @@ func (c *Config) loadConfig(endpoint string, certs *common.TLSCertificates, boot
 	c.AliveExpirationTimeout = util.GetDurationOrDefault("peer.gossip.aliveExpirationTimeout", 5*c.AliveTimeInterval)
 	c.AliveExpirationCheckInterval = c.AliveExpirationTimeout / 10
 	c.ReconnectInterval = util.GetDurationOrDefault("peer.gossip.reconnectInterval", c.AliveExpirationTimeout)
+	c.MaxConnectionAttempts = util.GetIntOrDefault("peer.gossip.maxConnectionAttempts", discovery.DefMaxConnectionAttempts)
+	c.MsgExpirationFactor = util.GetIntOrDefault("peer.gossip.msgExpirationFactor", discovery.DefMsgExpirationFactor)
 
 	return nil
 }

gossip/gossip/config_test.go

@@ -55,6 +55,8 @@ func TestGlobalConfig(t *testing.T) {
 	viper.Set("peer.gossip.aliveTimeInterval", "20s")
 	viper.Set("peer.gossip.aliveExpirationTimeout", "21s")
 	viper.Set("peer.gossip.reconnectInterval", "22s")
+	viper.Set("peer.gossip.maxConnectionAttempts", "100")
+	viper.Set("peer.gossip.msgExpirationFactor", "10")
 
 	coreConfig, err := gossip.GlobalConfig(endpoint, nil, bootstrap...)
 	assert.NoError(t, err)
@@ -95,6 +97,8 @@ func TestGlobalConfig(t *testing.T) {
 		AliveExpirationTimeout:       21 * time.Second,
 		AliveExpirationCheckInterval: 21 * time.Second / 10, // AliveExpirationTimeout / 10
 		ReconnectInterval:            22 * time.Second,
+		MaxConnectionAttempts:        100,
+		MsgExpirationFactor:          10,
 	}
 
 	assert.Equal(t, expectedConfig, coreConfig)
@@ -148,6 +152,8 @@ func TestGlobalConfigDefaults(t *testing.T) {
 		AliveExpirationTimeout:       5 * discovery.DefAliveTimeInterval,
 		AliveExpirationCheckInterval: 5 * discovery.DefAliveTimeInterval / 10,
 		ReconnectInterval:            5 * discovery.DefAliveTimeInterval,
+		MaxConnectionAttempts:        120,
+		MsgExpirationFactor:          20,
 	}
 
 	assert.Equal(t, expectedConfig, coreConfig)

gossip/gossip/gossip_impl.go

@@ -71,7 +71,8 @@ type Node struct {
 // New creates a gossip instance attached to a gRPC server
 func New(conf *Config, s *grpc.Server, sa api.SecurityAdvisor,
 	mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType,
-	secureDialOpts api.PeerSecureDialOpts, gossipMetrics *metrics.GossipMetrics) *Node {
+	secureDialOpts api.PeerSecureDialOpts, gossipMetrics *metrics.GossipMetrics,
+	anchorPeerTracker discovery.AnchorPeerTracker) *Node {
 	var err error
 
 	lgr := util.GetLogger(util.GossipLogger, conf.ID)
@@ -126,10 +127,12 @@ func New(conf *Config, s *grpc.Server, sa api.SecurityAdvisor,
 		AliveExpirationTimeout:       conf.AliveExpirationTimeout,
 		AliveExpirationCheckInterval: conf.AliveExpirationCheckInterval,
 		ReconnectInterval:            conf.ReconnectInterval,
+		MaxConnectionAttempts:        conf.MaxConnectionAttempts,
+		MsgExpirationFactor:          conf.MsgExpirationFactor,
 		BootstrapPeers:               conf.BootstrapPeers,
 	}
 	g.disc = discovery.NewDiscoveryService(g.selfNetworkMember(), g.discAdapter, g.disSecAdap, g.disclosurePolicy,
-		discoveryConfig)
+		discoveryConfig, anchorPeerTracker)
 	g.logger.Infof("Creating gossip service with self membership of %s", g.selfNetworkMember())
 
 	g.certPuller = g.createCertStorePuller()