diff --git a/etcdserver/corrupt.go b/etcdserver/corrupt.go index ea614954336e..a27def6d88de 100644 --- a/etcdserver/corrupt.go +++ b/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 {