fix(shardnode): fix panic problem when restart service with apply snapshot

supplement kvstore interface

with #22357426

Signed-off-by: xiejian <xiejian3@oppo.com>

blobstore/common/kvstorev2/kvstore.go

@@ -121,6 +121,7 @@ type (
 		Seek(key []byte)
 		SetFilterKey(key []byte)
 		Close()
+		CF() CF
 	}
 	KeyGetter interface {
 		Key() []byte
@@ -153,6 +154,14 @@ type (
 		Clear()
 		Close()
 		// Iterator()
+		Iterator() WriteBatchReader
+	}
+	WriteBatchReader interface {
+		Next() bool
+		Key() []byte
+		Value() []byte
+		CF() int
+		Type() byte
 	}
 
 	Stats struct {

blobstore/common/kvstorev2/rocksdb.go

@@ -78,6 +78,7 @@ type (
 		snap *rdb.Snapshot
 	}
 	listReader struct {
+		cf          CF
 		iterator    *rdb.Iterator
 		prefix      []byte
 		marker      []byte
@@ -102,6 +103,9 @@ type (
 		s     *rocksdb
 		batch *rdb.WriteBatch
 	}
+	writeBatchReader struct {
+		iterator *rdb.WriteBatchIterator
+	}
 )
 
 type opType int
@@ -561,6 +565,10 @@ func (lr *listReader) SetFilterKey(key []byte) {
 	lr.filterKeys = append(lr.filterKeys, key)
 }
 
+func (lr *listReader) CF() CF {
+	return lr.cf
+}
+
 func (lr *listReader) Close() {
 	lr.iterator.Close()
 }
@@ -607,10 +615,35 @@ func (w *writeBatch) Clear() {
 	w.batch.Clear()
 }
 
+func (w *writeBatch) Iterator() WriteBatchReader {
+	itr := w.batch.NewIterator()
+	return &writeBatchReader{iterator: itr}
+}
+
 func (w *writeBatch) Close() {
 	w.batch.Destroy()
 }
 
+func (br *writeBatchReader) Next() bool {
+	return br.iterator.Next()
+}
+
+func (br *writeBatchReader) Key() []byte {
+	return br.iterator.Record().Key
+}
+
+func (br *writeBatchReader) Value() []byte {
+	return br.iterator.Record().Value
+}
+
+func (br *writeBatchReader) CF() int {
+	return br.iterator.Record().CF
+}
+
+func (br *writeBatchReader) Type() byte {
+	return byte(br.iterator.Record().Type)
+}
+
 func (s *rocksdb) NewWriteBatch() WriteBatch {
 	return &writeBatch{
 		s:     s,
@@ -823,6 +856,7 @@ func (s *rocksdb) List(ctx context.Context, col CF, prefix []byte, marker []byte
 	}
 
 	lr := &listReader{
+		cf:          col,
 		iterator:    t,
 		marker:      marker,
 		prefix:      prefix,

blobstore/common/kvstorev2/rocksdb_test.go

@@ -547,3 +547,55 @@ func TestInstance_DeleteRange(t *testing.T) {
 		value.Close()
 	}
 }
+
+func TestWriteBatch(t *testing.T) {
+	ctx := context.TODO()
+	eg, err := newEngine(ctx, nil)
+	require.NoError(t, err)
+	defer eg.close()
+
+	col1 := CF("c1")
+	err = eg.engine.CreateColumn(col1)
+	require.Nil(t, err)
+
+	batch := eg.engine.NewWriteBatch()
+
+	n := 4
+	cfs := make([]CF, n)
+	keys := make([][]byte, n)
+	values := make([][]byte, n)
+
+	for i := 0; i < n; i++ {
+		keyStr := []byte(fmt.Sprintf("k%d", i))
+		valStr := []byte(fmt.Sprintf("v%d", i))
+		cfs[i] = col1
+		keys[i] = keyStr
+		values[i] = valStr
+		batch.Put(col1, keyStr, valStr)
+	}
+	raw := batch.Data()
+	t.Log(raw)
+
+	_batch := eg.engine.NewWriteBatch()
+	_batch.From(raw)
+	eg.engine.Write(context.TODO(), _batch)
+
+	reader := _batch.Iterator()
+	for {
+		ok := reader.Next()
+		if !ok {
+			break
+		}
+		t.Log(string(reader.Key()))
+		t.Log(string(reader.Value()))
+		t.Log(reader.CF())
+		t.Log(reader.Type())
+	}
+
+	for i := 0; i < n; i++ {
+		vg, err := eg.engine.Get(context.TODO(), col1, keys[i])
+		require.NoError(t, err)
+		require.Equal(t, values[i], vg.Value())
+		vg.Close()
+	}
+}

blobstore/common/raft/snapshot.go

@@ -105,13 +105,13 @@ func (i *incomingSnapshot) ReadBatch() (Batch, error) {
 }
 
 func (i *incomingSnapshot) Index() uint64 {
-	message := i.RaftMessageRequest.Message
-	return message.Index
+	snap := i.RaftMessageRequest.Message.Snapshot
+	return snap.Metadata.Index
 }
 
 func (i *incomingSnapshot) Term() uint64 {
-	message := i.RaftMessageRequest.Message
-	return message.Term
+	snap := i.RaftMessageRequest.Message.Snapshot
+	return snap.Metadata.Term
 }
 
 func (i *incomingSnapshot) Header() RaftSnapshotHeader {

blobstore/common/raft/storage.go

@@ -16,6 +16,7 @@ package raft
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
 	"fmt"
 	"math"
@@ -27,6 +28,8 @@ import (
 	"github.com/google/uuid"
 	"go.etcd.io/etcd/raft/v3"
 	"go.etcd.io/etcd/raft/v3/raftpb"
+
+	"github.com/cubefs/cubefs/blobstore/common/trace"
 )
 
 const (
@@ -159,6 +162,21 @@ func (s *storage) Term(i uint64) (uint64, error) {
 		return 0, nil
 	}
 
+	value, err := s.rawStg.Get(encodeIndexLogKey(s.id, i))
+	if err == nil {
+		entry := &raftpb.Entry{}
+		if err := entry.Unmarshal(value.Value()); err != nil {
+			return 0, err
+		}
+		return entry.Term, nil
+	}
+
+	// the first index log may not be found after apply snapshot,
+	// so return hard state commit first when i is equal to hard state's commit
+	if s.hardState.Commit == i {
+		return s.hardState.Term, nil
+	}
+
 	firstIndex, err := s.FirstIndex()
 	if err != nil {
 		return 0, err
@@ -167,42 +185,31 @@ func (s *storage) Term(i uint64) (uint64, error) {
 		return 0, raft.ErrCompacted
 	}
 
-	value, err := s.rawStg.Get(encodeIndexLogKey(s.id, i))
-	if err != nil {
-		// the first index log may not be found after apply snapshot,
-		// so return hard state commit first when i is equal to hard state's commit
-		if s.hardState.Commit == i {
-			return s.hardState.Term, nil
-		}
-		return 0, err
-	}
-
-	entry := &raftpb.Entry{}
-	if err := entry.Unmarshal(value.Value()); err != nil {
-		return 0, err
-	}
-
-	return entry.Term, nil
+	return 0, err
 }
 
 // LastIndex returns the index of the last entry in the log.
 func (s *storage) LastIndex() (uint64, error) {
+	span, _ := trace.StartSpanFromContext(context.Background(), "")
 	if lastIndex := atomic.LoadUint64(&s.lastIndex); lastIndex > 0 {
 		return lastIndex, nil
 	}
 
-	iterator := s.rawStg.Iter(nil)
+	iterator := s.rawStg.Iter(encodeIndexLogKeyPrefix(s.id))
 	defer iterator.Close()
 
 	if err := iterator.SeekForPrev(encodeIndexLogKey(s.id, math.MaxUint64)); err != nil {
+		span.Errorf("storage seek prev failed, err: %v", err)
 		return 0, err
 	}
 	keyGetter, valGetter, err := iterator.ReadPrev()
 	if err != nil {
+		span.Errorf("storage read prev failed, err: %v", err)
 		return 0, err
 	}
 	if valGetter == nil {
-		return 0, nil
+		return s.hardState.Commit, nil
+		// return 0, nil
 	}
 	defer func() {
 		keyGetter.Close()
@@ -210,7 +217,7 @@ func (s *storage) LastIndex() (uint64, error) {
 	}()
 
 	if !validForPrefix(keyGetter.Key(), encodeIndexLogKeyPrefix(s.id)) {
-		return 0, nil
+		return s.hardState.Commit, nil
 	}
 
 	entry := &raftpb.Entry{}
@@ -242,8 +249,9 @@ func (s *storage) FirstIndex() (uint64, error) {
 	if valGetter == nil {
 		// store the initialized value for first index when not found
 		// avoiding iterator call frequently
-		atomic.CompareAndSwapUint64(&s.firstIndex, uninitializedIndex, 1)
-		return 1, nil
+		// atomic.CompareAndSwapUint64(&s.firstIndex, uninitializedIndex, 1)
+		return s.hardState.Commit + 1, nil
+		// return 1, nil
 	}
 
 	defer valGetter.Close()

blobstore/common/raft/storage_test.go

@@ -155,7 +155,7 @@ func TestNewStorage_RaftStorage(t *testing.T) {
 		key := encodeIndexLogKey(cfg.id, math.MaxUint64)
 		mockIter.EXPECT().SeekForPrev(key).Times(1).Return(nil)
 
-		mockStorage.EXPECT().Iter(nil).Return(mockIter)
+		mockStorage.EXPECT().Iter(gomock.Any()).Return(mockIter)
 
 		index, err := s.LastIndex()
 		require.NoError(t, err)

blobstore/shardnode/storage/raft_impl.go

@@ -83,7 +83,7 @@ func (r *raftSnapshot) ReadBatch() (raft.Batch, error) {
 		}
 
 		if batch == nil {
-			batch = raftBatch{batch: r.kvStore.NewWriteBatch()}
+			batch = raftBatch{cf: r.lrs[r.iterIndex].CF(), batch: r.kvStore.NewWriteBatch()}
 		}
 		batch.Put(kg.Key(), vg.Value())
 		keyNum++

blobstore/shardnode/storage/shard_sm.go

@@ -79,7 +79,7 @@ func (s *shardSM) Apply(cxt context.Context, pd []raft.ProposalData, index uint6
 }
 
 func (s *shardSM) LeaderChange(peerID uint64) error {
-	log.Info("shard receive Leader change", peerID)
+	log.Info(fmt.Sprintf("shard receive Leader change, diskID: %d, suid: %d, peerID: %d", s.diskID, s.suid, peerID))
 	// todo: report Leader change to master
 	s.shardInfoMu.Lock()
 	s.shardInfoMu.leader = proto.DiskID(peerID)