etcdserver: add initial hash check

Signed-off-by: Gyu-Ho Lee <[email protected]>

etcdserver/corrupt.go

@@ -16,6 +16,7 @@ package etcdserver
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/coreos/etcd/clientv3"
@@ -29,6 +30,27 @@ func (s *EtcdServer) monitorKVHash() {
 	if t == 0 {
 		return
 	}
+
+	plog.Infof("corruption checking on %s", s.ID().String())
+	if s.isLeader() {
+		if err := s.checkHashKV(); err != nil {
+			plog.Debugf("check hash kv failed %v", err)
+		}
+	}
+
+	alarms := s.alarmStore.Get(pb.AlarmType_CORRUPT)
+	corrupted := false
+	for _, a := range alarms {
+		if a.MemberID == uint64(s.ID()) {
+			plog.Warningf("alarm %s found in %s", a.Alarm, s.ID().String())
+			corrupted = true
+		}
+	}
+	if corrupted {
+		plog.Fatalf("corrupted %s", s.ID().String())
+	}
+	plog.Infof("no corruption found on %s", s.ID().String())
+
 	plog.Infof("enabled corruption checking with %s interval", t)
 	for {
 		select {
@@ -50,34 +72,8 @@ func (s *EtcdServer) checkHashKV() error {
 	if err != nil {
 		plog.Fatalf("failed to hash kv store (%v)", err)
 	}
-	resps := []*clientv3.HashKVResponse{}
-	for _, m := range s.cluster.Members() {
-		if m.ID == s.ID() {
-			continue
-		}
 
-		cli, cerr := clientv3.New(clientv3.Config{Endpoints: m.PeerURLs})
-		if cerr != nil {
-			continue
-		}
-
-		respsLen := len(resps)
-		for _, c := range cli.Endpoints() {
-			ctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
-			resp, herr := cli.HashKV(ctx, c, rev)
-			cancel()
-			if herr == nil {
-				cerr = herr
-				resps = append(resps, resp)
-				break
-			}
-		}
-		cli.Close()
-
-		if respsLen == len(resps) {
-			plog.Warningf("failed to hash kv for peer %s (%v)", types.ID(m.ID), cerr)
-		}
-	}
+	resps := s.getHashKVFromPeer(rev)
 
 	ctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
 	err = s.linearizableReadNotify(ctx)
@@ -108,42 +104,96 @@ func (s *EtcdServer) checkHashKV() error {
 		})
 	}
 
-	if h2 != h && rev2 == rev && crev == crev2 {
-		plog.Warningf("mismatched hashes %d and %d for revision %d", h, h2, rev)
-		mismatch(uint64(s.ID()))
+	errs := s.compareHashKV(h, h2, rev, rev2, crev, crev2, resps, mismatch)
+	for _, err := range errs {
+		plog.Warning(err)
+	}
+	return nil
+}
+
+func (s *EtcdServer) getHashKVFromPeer(rev int64) (resps []*clientv3.HashKVResponse) {
+	for _, m := range s.cluster.Members() {
+		if m.ID == s.ID() {
+			continue
+		}
+
+		cli, cerr := clientv3.New(clientv3.Config{Endpoints: m.PeerURLs})
+		if cerr != nil {
+			continue
+		}
+
+		respsLen := len(resps)
+		for _, c := range cli.Endpoints() {
+			ctx, cancel := context.WithTimeout(context.Background(), s.Cfg.ReqTimeout())
+			resp, herr := cli.HashKV(ctx, c, rev)
+			cancel()
+			if herr == nil {
+				cerr = herr
+				resps = append(resps, resp)
+				break
+			}
+		}
+		cli.Close()
+
+		if respsLen == len(resps) {
+			plog.Warningf("failed to hash kv for peer %s (%v)", types.ID(m.ID), cerr)
+		}
 	}
+	return resps
+}
 
-	for _, resp := range resps {
+func (s *EtcdServer) compareHashKV(
+	prevHash, curHash uint32, prevRev, curRev, prevCrev, curCrev int64,
+	respsPeer []*clientv3.HashKVResponse,
+	mismatchFunc func(id uint64)) (errs []error) {
+	if prevHash != curHash && prevRev == curRev && prevCrev == curCrev {
+		err := fmt.Errorf("mismatched hashes %d and %d for revision %d", prevHash, curHash, curRev)
+		errs = append(errs, err)
+		if mismatchFunc != nil {
+			mismatchFunc(uint64(s.ID()))
+		}
+	}
+	for _, resp := range respsPeer {
 		id := resp.Header.MemberId
-		if resp.Header.Revision > rev2 {
-			plog.Warningf(
+		if resp.Header.Revision > curRev {
+			err := fmt.Errorf(
 				"revision %d from member %v, expected at most %d",
 				resp.Header.Revision,
 				types.ID(id),
-				rev2)
-			mismatch(id)
+				curRev,
+			)
+			errs = append(errs, err)
+			if mismatchFunc != nil {
+				mismatchFunc(id)
+			}
 		}
-		if resp.CompactRevision > crev2 {
-			plog.Warningf(
+		if resp.CompactRevision > curCrev {
+			err := fmt.Errorf(
 				"compact revision %d from member %v, expected at most %d",
 				resp.CompactRevision,
 				types.ID(id),
-				crev2,
+				curCrev,
 			)
-			mismatch(id)
+			errs = append(errs, err)
+			if mismatchFunc != nil {
+				mismatchFunc(id)
+			}
 		}
-		if resp.CompactRevision == crev && resp.Hash != h {
-			plog.Warningf(
+		if resp.CompactRevision == prevCrev && resp.Hash != prevHash {
+			err := fmt.Errorf(
 				"hash %d at revision %d from member %v, expected hash %d",
 				resp.Hash,
-				rev,
+				prevRev,
 				types.ID(id),
-				h,
+				prevHash,
 			)
-			mismatch(id)
+			errs = append(errs, err)
+			if mismatchFunc != nil {
+				mismatchFunc(id)
+			}
 		}
 	}
-	return nil
+	return errs
 }
 
 type applierV3Corrupt struct {