feat(master): master does snapshot optimization

close: #1916

Signed-off-by: NaturalSelect <2145973003@qq.com>

master/master_manager_test.go

@@ -6,9 +6,11 @@ import (
 	"io/ioutil"
 	"net/http"
 	"testing"
+	"time"
 
 	rproto "github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
 	"github.com/cubefs/cubefs/proto"
+	raftstore "github.com/cubefs/cubefs/raftstore/raftstore_db"
 )
 
 func TestHandleLeaderChange(t *testing.T) {
@@ -61,6 +63,74 @@ func TestRaft(t *testing.T) {
 	snapshotTest(t)
 }
 
+const volForSnapshot = "snapshtVol"
+
+const volForSnapshotCount = 300
+
+func BenchmarkSnapshot(b *testing.B) {
+	var err error
+	// perpare status
+	req := &createVolReq{
+		name:             "",
+		owner:            "cfs",
+		size:             1,
+		mpCount:          1,
+		dpReplicaNum:     3,
+		capacity:         300,
+		followerRead:     false,
+		authenticate:     false,
+		crossZone:        false,
+		normalZonesFirst: false,
+		zoneName:         testZone2,
+		description:      "",
+		qosLimitArgs:     &qosArgs{},
+	}
+	for i := 0; i != volForSnapshotCount; i++ {
+		req.name = fmt.Sprintf("%v_%v", volForSnapshot, i)
+		server.cluster.createVol(req)
+	}
+	time.Sleep(6 * time.Second)
+	mdSnapshot, err := server.cluster.fsm.Snapshot()
+	if err != nil {
+		b.Error(err)
+		return
+	}
+	b.Logf("snapshot apply index[%v]\n", mdSnapshot.ApplyIndex())
+	s := &Server{}
+
+	var dbStore *raftstore.RocksDBStore
+	dbStore, err = raftstore.NewRocksDBStore("/tmp/cubefs/raft3", LRUCacheSize, WriteBufferSize)
+	if err != nil {
+		b.Fatalf("init rocks db store fail cause: %v", err)
+	}
+	defer dbStore.Close()
+	fsm := &MetadataFsm{
+		rs:    server.fsm.rs,
+		store: dbStore,
+	}
+	fsm.registerApplySnapshotHandler(func() {
+		fsm.restore()
+	})
+	s.fsm = fsm
+	peers := make([]rproto.Peer, 0, len(server.config.peers))
+	for _, peer := range server.config.peers {
+		peers = append(peers, peer.Peer)
+	}
+	b.StopTimer()
+	b.ResetTimer()
+	b.StartTimer()
+	if err = fsm.ApplySnapshot(peers, mdSnapshot); err != nil {
+		b.Error(err)
+		return
+	}
+	b.StopTimer()
+	if fsm.applied != mdSnapshot.ApplyIndex() {
+		b.Errorf("applied not equal,applied[%v],snapshot applied[%v]\n", fsm.applied, mdSnapshot.ApplyIndex())
+		return
+	}
+	mdSnapshot.Close()
+}
+
 func snapshotTest(t *testing.T) {
 	var err error
 	mdSnapshot, err := server.cluster.fsm.Snapshot()
@@ -76,6 +146,7 @@ func snapshotTest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("init rocks db store fail cause: %v", err)
 	}
+	defer dbStore.Close()
 	fsm := &MetadataFsm{
 		rs:    server.fsm.rs,
 		store: dbStore,

master/metadata_fsm.go

@@ -18,11 +18,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"os"
 	"strconv"
 
 	"github.com/cubefs/cubefs/depends/tiglabs/raft"
 	"github.com/cubefs/cubefs/depends/tiglabs/raft/proto"
 	raftstore "github.com/cubefs/cubefs/raftstore/raftstore_db"
+	"github.com/cubefs/cubefs/util/fileutil"
 	"github.com/cubefs/cubefs/util/log"
 	"github.com/cubefs/cubefs/util/stat"
 )
@@ -179,25 +181,31 @@ func (mf *MetadataFsm) Snapshot() (proto.Snapshot, error) {
 
 // ApplySnapshot implements the interface of raft.StateMachine
 func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIterator) (err error) {
-	log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] reset rocksdb before applying snapshot"))
-	snap := mf.store.RocksDBSnapshot()
+	log.LogWarnf("action[ApplySnapshot] reset rocksdb before applying snapshot")
 	mf.onSnapshot = true
-	it := mf.store.Iterator(snap)
-
 	defer func() {
-		mf.store.ReleaseSnapshot(snap)
-		it.Close()
 		mf.onSnapshot = false
 	}()
-
-	for it.SeekToFirst(); it.Valid(); it.Next() {
-		key := string(it.Key().Data())
-		log.LogInfof("deleting Key: %v Value: %v", key, it.Value().Data())
-		mf.store.Del(key, false)
-	}
-
-	log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] begin,applied[%v]", mf.applied))
 	var data []byte
+	// clear recovery dir
+	recoveryDir := raftstore.GetRocksDBStoreRecoveryDir(mf.store.GetDir())
+	if fileutil.ExistDir(recoveryDir) {
+		if err = os.RemoveAll(recoveryDir); err != nil {
+			log.LogErrorf("failed to remove temp dir %v, error %v", recoveryDir, err.Error())
+			return
+		}
+	}
+	rocksdbOpened := true
+	// open temp rocksdb
+	tempDb, err := raftstore.NewRocksDBStore(recoveryDir, mf.store.GetLruCacheSize(), mf.store.GetWriteBufferSize())
+	if err != nil {
+		log.LogErrorf("failed to open temp rocksdb %v", err.Error())
+		goto errHandler
+	}
+	// close rocksdb
+	mf.store.Close()
+	rocksdbOpened = false
+	log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] begin,applied[%v]", mf.applied))
 	for err == nil {
 		bgTime := stat.BeginStat()
 		if data, err = iterator.Next(); err != nil {
@@ -206,27 +214,47 @@ func (mf *MetadataFsm) ApplySnapshot(peers []proto.Peer, iterator proto.SnapIter
 		stat.EndStat("ApplySnapshot-Next", err, bgTime, 1)
 		cmd := &RaftCmd{}
 		if err = json.Unmarshal(data, cmd); err != nil {
+			tempDb.Close()
 			goto errHandler
 		}
 		bgTime = stat.BeginStat()
-		if _, err = mf.store.Put(cmd.K, cmd.V, false); err != nil {
+		if _, err = tempDb.Put(cmd.K, cmd.V, false); err != nil {
+			tempDb.Close()
 			goto errHandler
 		}
 		stat.EndStat("ApplySnapshot-Put", err, bgTime, 1)
 	}
 	if err != nil && err != io.EOF {
+		tempDb.Close()
 		goto errHandler
 	}
 
-	if err = mf.store.Flush(); err != nil {
+	if err = tempDb.Flush(); err != nil {
 		log.LogError(fmt.Sprintf("action[ApplySnapshot] Flush failed,err:%v", err.Error()))
+		tempDb.Close()
 		goto errHandler
 	}
-
+	tempDb.Close()
+	// commit point
+	if err = os.RemoveAll(mf.store.GetDir()); err != nil {
+		goto errHandler
+	}
+	if err = os.Rename(tempDb.GetDir(), mf.store.GetDir()); err != nil {
+		goto errHandler
+	}
+	// finish snapshot
+	err = mf.store.Open()
+	if err != nil {
+		log.LogErrorf("failed to open rocksdb %v", err.Error())
+		return err
+	}
 	mf.snapshotHandler()
 	log.LogWarnf(fmt.Sprintf("action[ApplySnapshot] success,applied[%v]", mf.applied))
 	return nil
 errHandler:
+	if !rocksdbOpened {
+		mf.store.Open()
+	}
 	log.LogError(fmt.Sprintf("action[ApplySnapshot] failed,err:%v", err.Error()))
 	return err
 }

raftstore/raftstore_db/store_rocksdb.go

@@ -16,38 +16,91 @@ package raftstore_db
 
 import (
 	"fmt"
+	"strings"
 
 	"os"
 
+	"github.com/cubefs/cubefs/util/fileutil"
+	"github.com/cubefs/cubefs/util/log"
 	"github.com/tecbot/gorocksdb"
 )
 
 // RocksDBStore is a wrapper of the gorocksdb.DB
 type RocksDBStore struct {
-	dir string
-	db  *gorocksdb.DB
+	dir             string
+	lruCacheSize    int
+	writeBufferSize int
+	db              *gorocksdb.DB
+}
+
+func (rs *RocksDBStore) GetLruCacheSize() int {
+	return rs.lruCacheSize
+}
+
+func (rs *RocksDBStore) GetWriteBufferSize() int {
+	return rs.writeBufferSize
+}
+
+func (rs *RocksDBStore) GetDir() string {
+	return rs.dir
 }
 
 // NewRocksDBStore returns a new RocksDB instance.
 func NewRocksDBStore(dir string, lruCacheSize, writeBufferSize int) (store *RocksDBStore, err error) {
+
 	if err = os.MkdirAll(dir, os.ModePerm); err != nil {
 		return
 	}
-	store = &RocksDBStore{dir: dir}
-	if err = store.Open(lruCacheSize, writeBufferSize); err != nil {
+	store = &RocksDBStore{
+		dir:             dir,
+		lruCacheSize:    lruCacheSize,
+		writeBufferSize: writeBufferSize,
+	}
+	if err = store.Open(); err != nil {
+		return
+	}
+	return
+}
+
+func GetRocksDBStoreRecoveryDir(dir string) string {
+	dir = strings.TrimSuffix(dir, "/")
+	return fmt.Sprintf("%v_temp", dir)
+}
+
+// NewRocksDBStoreAndRecovery returns a new RocksDB instance after execute recovery.
+func NewRocksDBStoreAndRecovery(dir string, lruCacheSize, writeBufferSize int) (store *RocksDBStore, err error) {
+	// start recovery
+	recoverDir := GetRocksDBStoreRecoveryDir(dir)
+	// if rocksdb dir is not exists but temp dir is exist
+	if !fileutil.ExistDir(dir) && fileutil.ExistDir(recoverDir) {
+		// we move temp dir to rocksdb dir for commiting transaction
+		if err = os.Rename(recoverDir, dir); err != nil {
+			log.LogErrorf("failed to rename rocksdb recovery dir %v", err.Error())
+			return
+		}
+		log.LogDebug("recovery rocksdb success")
+	} else if err = os.MkdirAll(dir, os.ModePerm); err != nil {
+		return
+	}
+	store = &RocksDBStore{
+		dir:             dir,
+		lruCacheSize:    lruCacheSize,
+		writeBufferSize: writeBufferSize,
+	}
+	if err = store.Open(); err != nil {
 		return
 	}
 	return
 }
 
 // Open opens the RocksDB instance.
-func (rs *RocksDBStore) Open(lruCacheSize, writeBufferSize int) error {
+func (rs *RocksDBStore) Open() error {
 	basedTableOptions := gorocksdb.NewDefaultBlockBasedTableOptions()
-	basedTableOptions.SetBlockCache(gorocksdb.NewLRUCache(uint64(lruCacheSize)))
+	basedTableOptions.SetBlockCache(gorocksdb.NewLRUCache(uint64(rs.lruCacheSize)))
 	opts := gorocksdb.NewDefaultOptions()
 	opts.SetBlockBasedTableFactory(basedTableOptions)
 	opts.SetCreateIfMissing(true)
-	opts.SetWriteBufferSize(writeBufferSize)
+	opts.SetWriteBufferSize(rs.writeBufferSize)
 	opts.SetMaxWriteBufferNumber(2)
 	opts.SetCompression(gorocksdb.NoCompression)
 	db, err := gorocksdb.OpenDb(opts, rs.dir)
@@ -57,7 +110,10 @@ func (rs *RocksDBStore) Open(lruCacheSize, writeBufferSize int) error {
 	}
 	rs.db = db
 	return nil
+}
 
+func (rs *RocksDBStore) Close() {
+	rs.db.Close()
 }
 
 // Del deletes a key-value pair.

util/fileutil/exists.go

@@ -0,0 +1,30 @@
+// Copyright 2023 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package fileutil
+
+import "os"
+
+func Exist(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil || !os.IsNotExist(err)
+}
+
+func ExistDir(path string) bool {
+	state, err := os.Stat(path)
+	if err == nil || !os.IsNotExist(err) {
+		return state.IsDir()
+	}
+	return false
+}

util/fileutil/exists_test.go

@@ -0,0 +1,47 @@
+// Copyright 2023 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package fileutil_test
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/cubefs/cubefs/util/fileutil"
+)
+
+func TestExits(t *testing.T) {
+	if fileutil.Exist("/not_exist") {
+		t.Fail()
+		return
+	}
+	tempDir := os.TempDir()
+	tempFile := fmt.Sprintf("%v/exist", tempDir)
+	file, err := os.Create(tempFile)
+	if err != nil {
+		t.Errorf("failed to create file %v error: %v\n", tempFile, err.Error())
+		t.Fail()
+		return
+	}
+	file.Close()
+	if !fileutil.Exist(tempFile) {
+		t.Fail()
+		return
+	}
+	if !fileutil.ExistDir(tempDir) {
+		t.Fail()
+		return
+	}
+}