Split evictionSuspector from util.go

This CR splits evictionSuspector and PeriodicCheck from util.go
by creating eviction.go file. Also moves related tests to
eviction_test.go.

Signed-off-by: jongkwon.lee <jongkwon.lee@navercorp.com>
Change-Id: Ib0770d63ae495a54987a3b8dfdbcd208248e5e42

Author: jongkwon.lee

orderer/consensus/etcdraft/eviction.go

@@ -0,0 +1,142 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package etcdraft
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric/common/flogging"
+	"github.com/hyperledger/fabric/orderer/common/cluster"
+	"github.com/hyperledger/fabric/protoutil"
+	"go.etcd.io/etcd/raft/raftpb"
+)
+
+// PeriodicCheck checks periodically a condition, and reports
+// the cumulative consecutive period the condition was fulfilled.
+type PeriodicCheck struct {
+	Logger              *flogging.FabricLogger
+	CheckInterval       time.Duration
+	Condition           func() bool
+	Report              func(cumulativePeriod time.Duration)
+	conditionHoldsSince time.Time
+	once                sync.Once // Used to prevent double initialization
+	stopped             uint32
+}
+
+// Run runs the PeriodicCheck
+func (pc *PeriodicCheck) Run() {
+	pc.once.Do(pc.check)
+}
+
+// Stop stops the periodic checks
+func (pc *PeriodicCheck) Stop() {
+	pc.Logger.Info("Periodic check is stopping.")
+	atomic.AddUint32(&pc.stopped, 1)
+}
+
+func (pc *PeriodicCheck) shouldRun() bool {
+	return atomic.LoadUint32(&pc.stopped) == 0
+}
+
+func (pc *PeriodicCheck) check() {
+	if pc.Condition() {
+		pc.conditionFulfilled()
+	} else {
+		pc.conditionNotFulfilled()
+	}
+
+	if !pc.shouldRun() {
+		return
+	}
+	time.AfterFunc(pc.CheckInterval, pc.check)
+}
+
+func (pc *PeriodicCheck) conditionNotFulfilled() {
+	pc.conditionHoldsSince = time.Time{}
+}
+
+func (pc *PeriodicCheck) conditionFulfilled() {
+	if pc.conditionHoldsSince.IsZero() {
+		pc.conditionHoldsSince = time.Now()
+	}
+
+	pc.Report(time.Since(pc.conditionHoldsSince))
+}
+
+type evictionSuspector struct {
+	evictionSuspicionThreshold time.Duration
+	logger                     *flogging.FabricLogger
+	createPuller               CreateBlockPuller
+	height                     func() uint64
+	amIInChannel               cluster.SelfMembershipPredicate
+	halt                       func()
+	writeBlock                 func(block *common.Block) error
+	triggerCatchUp             func(sn *raftpb.Snapshot)
+	halted                     bool
+}
+
+func (es *evictionSuspector) confirmSuspicion(cumulativeSuspicion time.Duration) {
+	if es.evictionSuspicionThreshold > cumulativeSuspicion || es.halted {
+		return
+	}
+	es.logger.Infof("Suspecting our own eviction from the channel for %v", cumulativeSuspicion)
+	puller, err := es.createPuller()
+	if err != nil {
+		es.logger.Panicf("Failed creating a block puller: %v", err)
+	}
+
+	lastConfigBlock, err := cluster.PullLastConfigBlock(puller)
+	if err != nil {
+		es.logger.Errorf("Failed pulling the last config block: %v", err)
+		return
+	}
+
+	es.logger.Infof("Last config block was found to be block [%d]", lastConfigBlock.Header.Number)
+
+	height := es.height()
+
+	if lastConfigBlock.Header.Number+1 <= height {
+		es.logger.Infof("Our height is higher or equal than the height of the orderer we pulled the last block from, aborting.")
+		return
+	}
+
+	err = es.amIInChannel(lastConfigBlock)
+	if err != cluster.ErrNotInChannel && err != cluster.ErrForbidden {
+		details := fmt.Sprintf(", our certificate was found in config block with sequence %d", lastConfigBlock.Header.Number)
+		if err != nil {
+			details = fmt.Sprintf(": %s", err.Error())
+		}
+		es.logger.Infof("Cannot confirm our own eviction from the channel%s", details)
+
+		es.triggerCatchUp(&raftpb.Snapshot{Data: protoutil.MarshalOrPanic(lastConfigBlock)})
+		return
+	}
+
+	es.logger.Warningf("Detected our own eviction from the channel in block [%d]", lastConfigBlock.Header.Number)
+
+	es.logger.Infof("Waiting for chain to halt")
+	es.halt()
+	es.halted = true
+	es.logger.Infof("Chain has been halted, pulling remaining blocks up to (and including) eviction block.")
+
+	nextBlock := height
+	es.logger.Infof("Will now pull blocks %d to %d", nextBlock, lastConfigBlock.Header.Number)
+	for seq := nextBlock; seq <= lastConfigBlock.Header.Number; seq++ {
+		es.logger.Infof("Pulling block [%d]", seq)
+		block := puller.PullBlock(seq)
+		err := es.writeBlock(block)
+		if err != nil {
+			es.logger.Panicf("Failed writing block [%d] to the ledger: %v", block.Header.Number, err)
+		}
+	}
+
+	es.logger.Infof("Pulled all blocks up to eviction block.")
+}

orderer/consensus/etcdraft/eviction_test.go

@@ -0,0 +1,250 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package etcdraft
+
+import (
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/hyperledger/fabric-protos-go/common"
+	"github.com/hyperledger/fabric/common/flogging"
+	"github.com/hyperledger/fabric/orderer/common/cluster"
+	"github.com/hyperledger/fabric/orderer/common/cluster/mocks"
+	"github.com/hyperledger/fabric/protoutil"
+	"github.com/onsi/gomega"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"go.etcd.io/etcd/raft/raftpb"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+func TestPeriodicCheck(t *testing.T) {
+	t.Parallel()
+
+	g := gomega.NewGomegaWithT(t)
+
+	var cond uint32
+	var checkNum uint32
+
+	fiveChecks := func() bool {
+		return atomic.LoadUint32(&checkNum) > uint32(5)
+	}
+
+	condition := func() bool {
+		atomic.AddUint32(&checkNum, 1)
+		return atomic.LoadUint32(&cond) == uint32(1)
+	}
+
+	reports := make(chan time.Duration, 1000)
+
+	report := func(duration time.Duration) {
+		reports <- duration
+	}
+
+	check := &PeriodicCheck{
+		Logger:        flogging.MustGetLogger("test"),
+		Condition:     condition,
+		CheckInterval: time.Millisecond,
+		Report:        report,
+	}
+
+	go check.Run()
+
+	g.Eventually(fiveChecks, time.Minute, time.Millisecond).Should(gomega.BeTrue())
+	// trigger condition to be true
+	atomic.StoreUint32(&cond, 1)
+	g.Eventually(reports, time.Minute, time.Millisecond).Should(gomega.Not(gomega.BeEmpty()))
+	// read first report
+	firstReport := <-reports
+	g.Eventually(reports, time.Minute, time.Millisecond).Should(gomega.Not(gomega.BeEmpty()))
+	// read second report
+	secondReport := <-reports
+	// time increases between reports
+	g.Expect(secondReport).To(gomega.BeNumerically(">", firstReport))
+	// wait for the reports channel to be full
+	g.Eventually(func() int { return len(reports) }, time.Minute, time.Millisecond).Should(gomega.BeNumerically("==", 1000))
+
+	// trigger condition to be false
+	atomic.StoreUint32(&cond, 0)
+
+	var lastReport time.Duration
+	// drain the reports channel
+	for len(reports) > 0 {
+		select {
+		case report := <-reports:
+			lastReport = report
+		default:
+			break
+		}
+	}
+
+	// ensure the checks have been made
+	checksDoneSoFar := atomic.LoadUint32(&checkNum)
+	g.Consistently(reports, time.Second*2, time.Millisecond).Should(gomega.BeEmpty())
+	checksDoneAfter := atomic.LoadUint32(&checkNum)
+	g.Expect(checksDoneAfter).To(gomega.BeNumerically(">", checksDoneSoFar))
+	// but nothing has been reported
+	g.Expect(reports).To(gomega.BeEmpty())
+
+	// trigger the condition again
+	atomic.StoreUint32(&cond, 1)
+	g.Eventually(reports, time.Minute, time.Millisecond).Should(gomega.Not(gomega.BeEmpty()))
+	// The first report is smaller than the last report,
+	// so the countdown has been reset when the condition was reset
+	firstReport = <-reports
+	g.Expect(lastReport).To(gomega.BeNumerically(">", firstReport))
+	// Stop the periodic check.
+	check.Stop()
+	checkCountAfterStop := atomic.LoadUint32(&checkNum)
+	// Wait 50 times the check interval.
+	time.Sleep(check.CheckInterval * 50)
+	// Ensure that we cease checking the condition, hence the PeriodicCheck is stopped.
+	g.Expect(atomic.LoadUint32(&checkNum)).To(gomega.BeNumerically("<", checkCountAfterStop+2))
+}
+
+func TestEvictionSuspector(t *testing.T) {
+	configBlock := &common.Block{
+		Header: &common.BlockHeader{Number: 9},
+		Metadata: &common.BlockMetadata{
+			Metadata: [][]byte{{}, {}, {}, {}},
+		},
+	}
+	configBlock.Metadata.Metadata[common.BlockMetadataIndex_LAST_CONFIG] = protoutil.MarshalOrPanic(&common.Metadata{
+		Value: protoutil.MarshalOrPanic(&common.LastConfig{Index: 9}),
+	})
+
+	puller := &mocks.ChainPuller{}
+	puller.On("Close")
+	puller.On("HeightsByEndpoints").Return(map[string]uint64{"foo": 10}, nil)
+	puller.On("PullBlock", uint64(9)).Return(configBlock)
+
+	for _, testCase := range []struct {
+		description                 string
+		expectedPanic               string
+		expectedLog                 string
+		expectedCommittedBlockCount int
+		amIInChannelReturns         error
+		evictionSuspicionThreshold  time.Duration
+		blockPuller                 BlockPuller
+		blockPullerErr              error
+		height                      uint64
+		halt                        func()
+	}{
+		{
+			description:                "suspected time is lower than threshold",
+			evictionSuspicionThreshold: 11 * time.Minute,
+			halt:                       t.Fail,
+		},
+		{
+			description:                "puller creation fails",
+			evictionSuspicionThreshold: 10*time.Minute - time.Second,
+			blockPullerErr:             errors.New("oops"),
+			expectedPanic:              "Failed creating a block puller: oops",
+			halt:                       t.Fail,
+		},
+		{
+			description:                "our height is the highest",
+			expectedLog:                "Our height is higher or equal than the height of the orderer we pulled the last block from, aborting",
+			evictionSuspicionThreshold: 10*time.Minute - time.Second,
+			blockPuller:                puller,
+			height:                     10,
+			halt:                       t.Fail,
+		},
+		{
+			description:                "failed pulling the block",
+			expectedLog:                "Cannot confirm our own eviction from the channel: bad block",
+			evictionSuspicionThreshold: 10*time.Minute - time.Second,
+			amIInChannelReturns:        errors.New("bad block"),
+			blockPuller:                puller,
+			height:                     9,
+			halt:                       t.Fail,
+		},
+		{
+			description:                "we are still in the channel",
+			expectedLog:                "Cannot confirm our own eviction from the channel, our certificate was found in config block with sequence 9",
+			evictionSuspicionThreshold: 10*time.Minute - time.Second,
+			amIInChannelReturns:        nil,
+			blockPuller:                puller,
+			height:                     9,
+			halt:                       t.Fail,
+		},
+		{
+			description:                 "we are not in the channel",
+			expectedLog:                 "Detected our own eviction from the channel in block [9]",
+			evictionSuspicionThreshold:  10*time.Minute - time.Second,
+			amIInChannelReturns:         cluster.ErrNotInChannel,
+			blockPuller:                 puller,
+			height:                      8,
+			expectedCommittedBlockCount: 2,
+			halt: func() {
+				puller.On("PullBlock", uint64(8)).Return(&common.Block{
+					Header: &common.BlockHeader{Number: 8},
+					Metadata: &common.BlockMetadata{
+						Metadata: [][]byte{{}, {}, {}, {}},
+					},
+				})
+			},
+		},
+	} {
+		testCase := testCase
+		t.Run(testCase.description, func(t *testing.T) {
+			committedBlocks := make(chan *common.Block, 2)
+
+			commitBlock := func(block *common.Block) error {
+				committedBlocks <- block
+				return nil
+			}
+
+			es := &evictionSuspector{
+				halt: testCase.halt,
+				amIInChannel: func(_ *common.Block) error {
+					return testCase.amIInChannelReturns
+				},
+				evictionSuspicionThreshold: testCase.evictionSuspicionThreshold,
+				createPuller: func() (BlockPuller, error) {
+					return testCase.blockPuller, testCase.blockPullerErr
+				},
+				writeBlock: commitBlock,
+				height: func() uint64 {
+					return testCase.height
+				},
+				logger:         flogging.MustGetLogger("test"),
+				triggerCatchUp: func(sn *raftpb.Snapshot) { return },
+			}
+
+			foundExpectedLog := testCase.expectedLog == ""
+			es.logger = es.logger.WithOptions(zap.Hooks(func(entry zapcore.Entry) error {
+				if strings.Contains(entry.Message, testCase.expectedLog) {
+					foundExpectedLog = true
+				}
+				return nil
+			}))
+
+			runTestCase := func() {
+				es.confirmSuspicion(time.Minute * 10)
+			}
+
+			if testCase.expectedPanic != "" {
+				assert.PanicsWithValue(t, testCase.expectedPanic, runTestCase)
+			} else {
+				runTestCase()
+				// Run the test case again.
+				// Conditions that do not lead to a conclusion of a chain eviction
+				// should be idempotent.
+				// Conditions that do lead to conclusion of a chain eviction
+				// in the second time - should result in a no-op.
+				runTestCase()
+			}
+
+			assert.True(t, foundExpectedLog, "expected to find %s but didn't", testCase.expectedLog)
+			assert.Equal(t, testCase.expectedCommittedBlockCount, len(committedBlocks))
+		})
+	}
+}