diff --git a/master/master_manager_test.go b/master/master_manager_test.go index 597c0d5bc..0c4823c1a 100644 --- a/master/master_manager_test.go +++ b/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, diff --git a/master/metadata_fsm.go b/master/metadata_fsm.go index 4001ac915..eb3b049ef 100644 --- a/master/metadata_fsm.go +++ b/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 } diff --git a/raftstore/raftstore_db/store_rocksdb.go b/raftstore/raftstore_db/store_rocksdb.go index b7428c61d..a1c3febe4 100644 --- a/raftstore/raftstore_db/store_rocksdb.go +++ b/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. diff --git a/util/fileutil/exists.go b/util/fileutil/exists.go new file mode 100644 index 000000000..b52fc33b4 --- /dev/null +++ b/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 +} diff --git a/util/fileutil/exists_test.go b/util/fileutil/exists_test.go new file mode 100644 index 000000000..c5c8e1f87 --- /dev/null +++ b/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 + } +}