cleanup WAL incase of KVValue decode error before panic

server/etcdserver/corrupt.go

@@ -133,6 +133,8 @@ func (s *EtcdServer) monitorKVDecoding() {
 			return
 		case <-s.kv.ErrorC():
 			s.sendCorruptionAlarmRequest(uint64(s.ID()))
+			s.r.storage.Cleanup()
+			panic("watchable kv decoding error")
 		}
 	}
 }

server/storage/storage.go

@@ -37,6 +37,8 @@ type Storage interface {
 	Release(snap raftpb.Snapshot) error
 	// Sync WAL
 	Sync() error
+	// CLean up WAL
+	Cleanup() error
 	// MinimalEtcdVersion returns minimal etcd storage able to interpret WAL log.
 	MinimalEtcdVersion() *semver.Version
 }
@@ -87,6 +89,13 @@ func (st *storage) Release(snap raftpb.Snapshot) error {
 	return st.s.ReleaseSnapDBs(snap)
 }
 
+func (st *storage) Cleanup() error {
+	st.mux.RLock()
+	defer st.mux.RUnlock()
+	st.w.CleanupWAL(st.lg)
+	return nil
+}
+
 func (st *storage) Save(s raftpb.HardState, ents []raftpb.Entry) error {
 	st.mux.RLock()
 	defer st.mux.RUnlock()

server/storage/wal/wal.go

@@ -188,7 +188,7 @@ func Create(lg *zap.Logger, dirpath string, metadata []byte) (*WAL, error) {
 	var perr error
 	defer func() {
 		if perr != nil {
-			w.cleanupWAL(lg)
+			w.CleanupWAL(lg)
 		}
 	}()
 
@@ -246,7 +246,7 @@ func (w *WAL) SetUnsafeNoFsync() {
 	w.unsafeNoSync = true
 }
 
-func (w *WAL) cleanupWAL(lg *zap.Logger) {
+func (w *WAL) CleanupWAL(lg *zap.Logger) {
 	var err error
 	if err = w.Close(); err != nil {
 		lg.Panic("failed to close WAL during cleanup", zap.Error(err))

server/storage/wal/wal_test.go

@@ -127,7 +127,7 @@ func TestWalCleanup(t *testing.T) {
 	if err != nil {
 		t.Fatalf("err = %v, want nil", err)
 	}
-	w.cleanupWAL(logger)
+	w.CleanupWAL(logger)
 	fnames, err := fileutil.ReadDir(testRoot)
 	if err != nil {
 		t.Fatalf("err = %v, want nil", err)