You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "git-hulk (via GitHub)" <gi...@apache.org> on 2024/07/08 15:15:57 UTC
Re: [PR] Add support of the embedded storage [kvrocks-controller]
git-hulk commented on code in PR #183:
URL: https://github.com/apache/kvrocks-controller/pull/183#discussion_r1668778535
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
Review Comment:
Can we use a struct to group those channels to avoid overloading the function parameters? like
```Go
type RaftNotifier struct {
proposeC <-chan string
confChangeC <-chan raftpb.ConfChange
...
}
```
##########
store/engine/embedded/embedded.go:
##########
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "bytes"
+ "context"
+ "encoding/gob"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/apache/kvrocks-controller/consts"
+ "github.com/apache/kvrocks-controller/logger"
+ persistence "github.com/apache/kvrocks-controller/store/engine"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ "go.uber.org/zap"
+ )
+
+ type Peer struct {
+ Api string `yaml:"api"`
+ Raft string `yaml:"raft"`
+ }
+
+ type Config struct {
+ Peers []Peer `yaml:"peers"`
+ Join bool `yaml:"join"`
+ }
+
+ func parseConfig(id string, cfg *Config) (int, []string, []string, error) {
+ apiPeers := make([]string, len(cfg.Peers))
+ raftPeers := make([]string, len(cfg.Peers))
+ nodeId := -1
+ for i, peer := range cfg.Peers {
+ if peer.Api == id {
+ nodeId = i + 1
+ }
+ apiPeers[i] = peer.Api
+ if !strings.HasPrefix(peer.Raft, "http://") {
+ raftPeers[i] = fmt.Sprintf("http://%s", peer.Raft)
+ } else {
+ raftPeers[i] = peer.Raft
+ }
+ }
+ if nodeId == -1 {
+ return 0, apiPeers, raftPeers, fmt.Errorf("Address %s is not in embedded store peers configuration", id)
+ }
+ return nodeId, apiPeers, raftPeers, nil
+ }
+
+ type Embedded struct {
+ kv map[string][]byte
+ kvMu sync.RWMutex
+ snapshotter *snap.Snapshotter
+
+ node *raftNode
+
+ myID string
+ PeerIDs []string
+
+ quitCh chan struct{}
+ leaderChangeCh <-chan bool
+ proposeCh chan string
+ confChangeCh chan raftpb.ConfChange
+ }
+
+ func New(id string, cfg *Config) (*Embedded, error) {
+ nodeId, apiPeers, raftPeers, err := parseConfig(id, cfg)
+ if err != nil {
+ return nil, err
+ }
+
+ proposeCh := make(chan string)
+ confChangeCh := make(chan raftpb.ConfChange)
+ leaderChangeCh := make(chan bool)
+ commitCh := make(chan *commit)
+ errorCh := make(chan error)
+ snapshotterReady := make(chan *snap.Snapshotter, 1)
+
+ e := &Embedded{
+ kv: make(map[string][]byte),
+ myID: id,
+ PeerIDs: apiPeers,
+ quitCh: make(chan struct{}),
+ leaderChangeCh: leaderChangeCh,
+ proposeCh: proposeCh,
+ confChangeCh: confChangeCh,
+ }
+
+ getSnapshot := func() ([]byte, error) {
+ e.kvMu.RLock()
+ defer e.kvMu.RUnlock()
+ return json.Marshal(e.kv)
+ }
+ // start raft node synchronization loop
+ e.node = newRaftNode(nodeId, raftPeers, cfg.Join, getSnapshot, proposeCh, confChangeCh, leaderChangeCh, commitCh, errorCh, snapshotterReady)
+
+ // block until snapshotter initialized
+ e.snapshotter = <-snapshotterReady
+ snapshot := e.loadSnapshot()
+ if snapshot != nil {
+ logger.Get().Sugar().Infof("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
+ if err := e.recoverFromSnapshot(snapshot.Data); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to recover snapshot")
+ }
+ }
+
+ go e.readCommits(commitCh, errorCh)
+ return e, nil
+ }
+
+ func (e *Embedded) loadSnapshot() *raftpb.Snapshot {
+ snapshot, err := e.snapshotter.Load()
+ if err != nil {
+ if !errors.Is(err, snap.ErrNoSnapshot) && !errors.Is(err, snap.ErrEmptySnapshot) {
+ logger.Get().With(zap.Error(err)).Error("Failed to initialize snapshot")
+ }
+ return nil
+ }
+ return snapshot
+ }
+
+ func (e *Embedded) recoverFromSnapshot(snapshot []byte) error {
+ var store map[string][]byte
+ if err := json.Unmarshal(snapshot, &store); err != nil {
+ return err
+ }
+ e.kvMu.Lock()
+ defer e.kvMu.Unlock()
+ e.kv = store
+ return nil
+ }
+
+ func (e *Embedded) readCommits(commitCh <-chan *commit, errorCh <-chan error) {
+ for c := range commitCh {
+ if c == nil {
+ // signaled to load snapshot
+ snapshot := e.loadSnapshot()
+ if snapshot != nil {
+ logger.Get().Sugar().Infof("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
+ if err := e.recoverFromSnapshot(snapshot.Data); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to recover snapshot")
+ }
+ }
+ continue
+ }
+
+ for _, data := range c.data {
+ var entry persistence.Entry
+ dec := gob.NewDecoder(bytes.NewBufferString(data))
+ if err := dec.Decode(&entry); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to decode message")
Review Comment:
Should return here?
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
+ if len(ents) == 0 {
+ return ents
+ }
+ firstIdx := ents[0].Index
+ if firstIdx > rc.appliedIndex+1 {
+ logger.Get().Sugar().Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
+ }
+ if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) {
+ nents = ents[rc.appliedIndex-firstIdx+1:]
+ }
+ return nents
+ }
+
+ // publishEntries writes committed log entries to commit channel and returns
+ // whether all entries could be published.
+ func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {
+ if len(ents) == 0 {
+ return nil, true
+ }
+
+ data := make([]string, 0, len(ents))
+ for i := range ents {
+ switch ents[i].Type {
+ case raftpb.EntryNormal:
+ if len(ents[i].Data) == 0 {
+ // ignore empty messages
+ break
+ }
+ s := string(ents[i].Data)
+ data = append(data, s)
+ case raftpb.EntryConfChange:
+ var cc raftpb.ConfChange
+ _ = cc.Unmarshal(ents[i].Data)
+ rc.confState = *rc.node.ApplyConfChange(cc)
+ switch cc.Type {
+ case raftpb.ConfChangeAddNode:
+ if len(cc.Context) > 0 {
+ rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
+ }
+ case raftpb.ConfChangeRemoveNode:
+ if cc.NodeID == uint64(rc.id) {
+ logger.Get().Info("I've been removed from the cluster! Shutting down.")
+ return nil, false
+ }
+ rc.transport.RemovePeer(types.ID(cc.NodeID))
+ case raftpb.ConfChangeUpdateNode, raftpb.ConfChangeAddLearnerNode:
+ }
+ case raftpb.EntryConfChangeV2:
+ }
+ }
+
+ var applyDoneC chan struct{}
+
+ if len(data) > 0 {
+ applyDoneC = make(chan struct{}, 1)
+ select {
+ case rc.commitC <- &commit{data, applyDoneC}:
+ case <-rc.stopCh:
+ return nil, false
+ }
+ }
+
+ // after commit, update appliedIndex
+ rc.appliedIndex = ents[len(ents)-1].Index
+
+ return applyDoneC, true
+ }
+
+ func (rc *raftNode) loadSnapshot() *raftpb.Snapshot {
+ if wal.Exist(rc.walDir) {
+ walSnaps, err := wal.ValidSnapshotEntries(logger.Get(), rc.walDir)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error listing snapshots")
+ }
+ snapshot, err := rc.snapshotter.LoadNewestAvailable(walSnaps)
+ if err != nil && !errors.Is(err, snap.ErrNoSnapshot) {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading snapshot")
+ }
+ return snapshot
+ }
+ return &raftpb.Snapshot{}
+ }
+
+ // openWAL returns a WAL ready for reading.
+ func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL {
+ if !wal.Exist(rc.walDir) {
+ if err := os.Mkdir(rc.walDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for WAL")
+ }
+
+ w, err := wal.Create(zap.NewExample(), rc.walDir, nil)
Review Comment:
Why use zap.NewExample() here?
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
+ if len(ents) == 0 {
+ return ents
+ }
+ firstIdx := ents[0].Index
+ if firstIdx > rc.appliedIndex+1 {
+ logger.Get().Sugar().Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
+ }
+ if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) {
+ nents = ents[rc.appliedIndex-firstIdx+1:]
+ }
+ return nents
+ }
+
+ // publishEntries writes committed log entries to commit channel and returns
+ // whether all entries could be published.
+ func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {
+ if len(ents) == 0 {
+ return nil, true
+ }
+
+ data := make([]string, 0, len(ents))
+ for i := range ents {
+ switch ents[i].Type {
+ case raftpb.EntryNormal:
+ if len(ents[i].Data) == 0 {
+ // ignore empty messages
+ break
+ }
+ s := string(ents[i].Data)
+ data = append(data, s)
+ case raftpb.EntryConfChange:
+ var cc raftpb.ConfChange
+ _ = cc.Unmarshal(ents[i].Data)
+ rc.confState = *rc.node.ApplyConfChange(cc)
+ switch cc.Type {
+ case raftpb.ConfChangeAddNode:
+ if len(cc.Context) > 0 {
+ rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
+ }
+ case raftpb.ConfChangeRemoveNode:
+ if cc.NodeID == uint64(rc.id) {
+ logger.Get().Info("I've been removed from the cluster! Shutting down.")
+ return nil, false
+ }
+ rc.transport.RemovePeer(types.ID(cc.NodeID))
+ case raftpb.ConfChangeUpdateNode, raftpb.ConfChangeAddLearnerNode:
+ }
+ case raftpb.EntryConfChangeV2:
+ }
+ }
+
+ var applyDoneC chan struct{}
+
+ if len(data) > 0 {
+ applyDoneC = make(chan struct{}, 1)
+ select {
+ case rc.commitC <- &commit{data, applyDoneC}:
+ case <-rc.stopCh:
+ return nil, false
+ }
+ }
+
+ // after commit, update appliedIndex
+ rc.appliedIndex = ents[len(ents)-1].Index
+
+ return applyDoneC, true
+ }
+
+ func (rc *raftNode) loadSnapshot() *raftpb.Snapshot {
+ if wal.Exist(rc.walDir) {
+ walSnaps, err := wal.ValidSnapshotEntries(logger.Get(), rc.walDir)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error listing snapshots")
+ }
+ snapshot, err := rc.snapshotter.LoadNewestAvailable(walSnaps)
+ if err != nil && !errors.Is(err, snap.ErrNoSnapshot) {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading snapshot")
+ }
+ return snapshot
+ }
+ return &raftpb.Snapshot{}
+ }
+
+ // openWAL returns a WAL ready for reading.
+ func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL {
+ if !wal.Exist(rc.walDir) {
+ if err := os.Mkdir(rc.walDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for WAL")
+ }
+
+ w, err := wal.Create(zap.NewExample(), rc.walDir, nil)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Create WAL error")
+ }
+ _ = w.Close()
+ }
+
+ walSnap := walpb.Snapshot{}
+ if snapshot != nil {
+ walSnap.Index, walSnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
+ }
+ logger.Get().Sugar().Infof("Loading WAL at term %d and index %d", walSnap.Term, walSnap.Index)
+ w, err := wal.Open(logger.Get(), rc.walDir, walSnap)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading WAL")
+ }
+
+ return w
+ }
+
+ // replayWAL replays WAL entries into the raft instance.
+ func (rc *raftNode) replayWAL() *wal.WAL {
+ logger.Get().Sugar().Infof("replaying WAL of member %d", rc.id)
+ snapshot := rc.loadSnapshot()
+ w := rc.openWAL(snapshot)
+ _, st, ents, err := w.ReadAll()
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Failed to read WAL")
+ }
+ rc.raftStorage = raft.NewMemoryStorage()
+ if snapshot != nil {
+ _ = rc.raftStorage.ApplySnapshot(*snapshot)
+ }
+ _ = rc.raftStorage.SetHardState(st)
+
+ // append to storage so raft starts at the right place in log
+ _ = rc.raftStorage.Append(ents)
+
+ return w
+ }
+
+ func (rc *raftNode) writeError(err error) {
+ rc.stopHTTP()
+ close(rc.commitC)
+ rc.errorC <- err
+ close(rc.errorC)
+ rc.node.Stop()
+ }
+
+ func (rc *raftNode) startRaft() {
+ if !fileutil.Exist(rc.snapDir) {
+ if err := os.Mkdir(rc.snapDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for snapshot")
+ }
+ }
+ rc.snapshotter = snap.New(logger.Get(), rc.snapDir)
+
+ oldwal := wal.Exist(rc.walDir)
+ rc.wal = rc.replayWAL()
+
+ // signal replay has finished
+ rc.snapshotterReady <- rc.snapshotter
+
+ rpeers := make([]raft.Peer, len(rc.peers))
+ for i := range rpeers {
+ rpeers[i] = raft.Peer{ID: uint64(i + 1)}
+ }
+ c := &raft.Config{
+ ID: uint64(rc.id),
+ ElectionTick: 10,
+ HeartbeatTick: 1,
+ Storage: rc.raftStorage,
+ MaxSizePerMsg: 1024 * 1024,
+ MaxInflightMsgs: 256,
+ MaxUncommittedEntriesSize: 1 << 30,
+ }
+
+ if oldwal || rc.join {
+ rc.node = raft.RestartNode(c)
+ } else {
+ rc.node = raft.StartNode(c, rpeers)
+ }
+
+ rc.transport = &rafthttp.Transport{
+ Logger: logger.Get(),
+ ID: types.ID(rc.id),
+ ClusterID: 0x1000,
+ Raft: rc,
+ ServerStats: stats.NewServerStats("", ""),
+ LeaderStats: stats.NewLeaderStats(zap.NewExample(), strconv.Itoa(rc.id)),
+ ErrorC: make(chan error),
+
+ DialRetryFrequency: 1,
+ }
+
+ if err := rc.transport.Start(); err != nil {
+ logger.Get().With(zap.Error(err)).Panic("Failed to start raft http server")
+ }
+ for i := range rc.peers {
+ if i+1 != rc.id {
+ rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
+ }
+ }
+
+ go rc.serveRaft()
Review Comment:
Should wait for those two routines while stopping
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
Review Comment:
entriesToApply => applyEntries
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
+ if len(ents) == 0 {
+ return ents
+ }
+ firstIdx := ents[0].Index
+ if firstIdx > rc.appliedIndex+1 {
+ logger.Get().Sugar().Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
Review Comment:
Should return?
##########
store/engine/embedded/embedded.go:
##########
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "bytes"
+ "context"
+ "encoding/gob"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/apache/kvrocks-controller/consts"
+ "github.com/apache/kvrocks-controller/logger"
+ persistence "github.com/apache/kvrocks-controller/store/engine"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ "go.uber.org/zap"
+ )
+
+ type Peer struct {
+ Api string `yaml:"api"`
+ Raft string `yaml:"raft"`
+ }
+
+ type Config struct {
+ Peers []Peer `yaml:"peers"`
+ Join bool `yaml:"join"`
+ }
+
+ func parseConfig(id string, cfg *Config) (int, []string, []string, error) {
+ apiPeers := make([]string, len(cfg.Peers))
+ raftPeers := make([]string, len(cfg.Peers))
+ nodeId := -1
+ for i, peer := range cfg.Peers {
+ if peer.Api == id {
+ nodeId = i + 1
+ }
+ apiPeers[i] = peer.Api
+ if !strings.HasPrefix(peer.Raft, "http://") {
+ raftPeers[i] = fmt.Sprintf("http://%s", peer.Raft)
+ } else {
+ raftPeers[i] = peer.Raft
+ }
+ }
+ if nodeId == -1 {
+ return 0, apiPeers, raftPeers, fmt.Errorf("Address %s is not in embedded store peers configuration", id)
+ }
+ return nodeId, apiPeers, raftPeers, nil
+ }
+
+ type Embedded struct {
+ kv map[string][]byte
+ kvMu sync.RWMutex
+ snapshotter *snap.Snapshotter
+
+ node *raftNode
+
+ myID string
+ PeerIDs []string
+
+ quitCh chan struct{}
+ leaderChangeCh <-chan bool
+ proposeCh chan string
+ confChangeCh chan raftpb.ConfChange
+ }
+
+ func New(id string, cfg *Config) (*Embedded, error) {
+ nodeId, apiPeers, raftPeers, err := parseConfig(id, cfg)
+ if err != nil {
+ return nil, err
+ }
+
+ proposeCh := make(chan string)
+ confChangeCh := make(chan raftpb.ConfChange)
+ leaderChangeCh := make(chan bool)
+ commitCh := make(chan *commit)
+ errorCh := make(chan error)
+ snapshotterReady := make(chan *snap.Snapshotter, 1)
+
+ e := &Embedded{
+ kv: make(map[string][]byte),
+ myID: id,
+ PeerIDs: apiPeers,
+ quitCh: make(chan struct{}),
+ leaderChangeCh: leaderChangeCh,
+ proposeCh: proposeCh,
+ confChangeCh: confChangeCh,
+ }
+
+ getSnapshot := func() ([]byte, error) {
+ e.kvMu.RLock()
+ defer e.kvMu.RUnlock()
+ return json.Marshal(e.kv)
+ }
+ // start raft node synchronization loop
+ e.node = newRaftNode(nodeId, raftPeers, cfg.Join, getSnapshot, proposeCh, confChangeCh, leaderChangeCh, commitCh, errorCh, snapshotterReady)
+
+ // block until snapshotter initialized
+ e.snapshotter = <-snapshotterReady
+ snapshot := e.loadSnapshot()
+ if snapshot != nil {
+ logger.Get().Sugar().Infof("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
+ if err := e.recoverFromSnapshot(snapshot.Data); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to recover snapshot")
+ }
+ }
+
+ go e.readCommits(commitCh, errorCh)
+ return e, nil
+ }
+
+ func (e *Embedded) loadSnapshot() *raftpb.Snapshot {
+ snapshot, err := e.snapshotter.Load()
+ if err != nil {
+ if !errors.Is(err, snap.ErrNoSnapshot) && !errors.Is(err, snap.ErrEmptySnapshot) {
+ logger.Get().With(zap.Error(err)).Error("Failed to initialize snapshot")
+ }
+ return nil
+ }
+ return snapshot
+ }
+
+ func (e *Embedded) recoverFromSnapshot(snapshot []byte) error {
+ var store map[string][]byte
+ if err := json.Unmarshal(snapshot, &store); err != nil {
+ return err
+ }
+ e.kvMu.Lock()
+ defer e.kvMu.Unlock()
+ e.kv = store
+ return nil
+ }
+
+ func (e *Embedded) readCommits(commitCh <-chan *commit, errorCh <-chan error) {
+ for c := range commitCh {
+ if c == nil {
+ // signaled to load snapshot
+ snapshot := e.loadSnapshot()
+ if snapshot != nil {
+ logger.Get().Sugar().Infof("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
+ if err := e.recoverFromSnapshot(snapshot.Data); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to recover snapshot")
+ }
+ }
+ continue
+ }
+
+ for _, data := range c.data {
+ var entry persistence.Entry
+ dec := gob.NewDecoder(bytes.NewBufferString(data))
+ if err := dec.Decode(&entry); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to decode message")
+ }
+ e.kvMu.Lock()
+ if entry.Value == nil {
+ delete(e.kv, entry.Key)
+ } else {
+ e.kv[entry.Key] = entry.Value
+ }
+ e.kvMu.Unlock()
+ }
+ close(c.applyDoneC)
+ }
+ if err, ok := <-errorCh; ok {
+ logger.Get().With(zap.Error(err)).Error("Error occurred during reading commits")
+ }
+ }
+
+ func (e *Embedded) ID() string {
+ return e.myID
+ }
+
+ func (e *Embedded) Leader() string {
+ if e.node.leader.Load() == 0 {
+ return e.myID
+ }
+ return e.PeerIDs[e.node.leader.Load()-1]
+ }
+
+ func (e *Embedded) LeaderChange() <-chan bool {
+ return e.leaderChangeCh
+ }
+
+ func (e *Embedded) IsReady(ctx context.Context) bool {
+ for {
+ select {
+ case <-e.quitCh:
+ return false
+ case <-time.After(100 * time.Millisecond):
+ if e.node.leader.Load() != 0 {
+ return true
+ }
+ case <-ctx.Done():
+ return e.node.leader.Load() != 0
+ }
+ }
+ }
+
+ func (e *Embedded) Get(_ context.Context, key string) ([]byte, error) {
+ e.kvMu.RLock()
+ defer e.kvMu.RUnlock()
+ value, ok := e.kv[key]
+ if !ok {
+ return nil, consts.ErrNotFound
+ }
+ return value, nil
+ }
+
+ func (e *Embedded) Exists(ctx context.Context, key string) (bool, error) {
+ _, err := e.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, consts.ErrNotFound) {
+ return false, nil
+ }
+ return false, err
+ }
+ return true, nil
+ }
+
+ func (e *Embedded) Propose(k string, v []byte) {
+ var buf strings.Builder
+ if err := gob.NewEncoder(&buf).Encode(persistence.Entry{Key: k, Value: v}); err != nil {
+ logger.Get().With(zap.Error(err)).Error("Failed to propose changes")
Review Comment:
Should we continue sending the buffer string while it's failed to encode?
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
+ if len(ents) == 0 {
+ return ents
+ }
+ firstIdx := ents[0].Index
+ if firstIdx > rc.appliedIndex+1 {
+ logger.Get().Sugar().Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
+ }
+ if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) {
+ nents = ents[rc.appliedIndex-firstIdx+1:]
+ }
+ return nents
+ }
+
+ // publishEntries writes committed log entries to commit channel and returns
+ // whether all entries could be published.
+ func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {
+ if len(ents) == 0 {
+ return nil, true
+ }
+
+ data := make([]string, 0, len(ents))
+ for i := range ents {
+ switch ents[i].Type {
+ case raftpb.EntryNormal:
+ if len(ents[i].Data) == 0 {
+ // ignore empty messages
+ break
+ }
+ s := string(ents[i].Data)
+ data = append(data, s)
+ case raftpb.EntryConfChange:
+ var cc raftpb.ConfChange
+ _ = cc.Unmarshal(ents[i].Data)
+ rc.confState = *rc.node.ApplyConfChange(cc)
+ switch cc.Type {
+ case raftpb.ConfChangeAddNode:
+ if len(cc.Context) > 0 {
+ rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
+ }
+ case raftpb.ConfChangeRemoveNode:
+ if cc.NodeID == uint64(rc.id) {
+ logger.Get().Info("I've been removed from the cluster! Shutting down.")
+ return nil, false
+ }
+ rc.transport.RemovePeer(types.ID(cc.NodeID))
+ case raftpb.ConfChangeUpdateNode, raftpb.ConfChangeAddLearnerNode:
+ }
+ case raftpb.EntryConfChangeV2:
+ }
+ }
+
+ var applyDoneC chan struct{}
+
+ if len(data) > 0 {
+ applyDoneC = make(chan struct{}, 1)
+ select {
+ case rc.commitC <- &commit{data, applyDoneC}:
+ case <-rc.stopCh:
+ return nil, false
+ }
+ }
+
+ // after commit, update appliedIndex
+ rc.appliedIndex = ents[len(ents)-1].Index
+
+ return applyDoneC, true
+ }
+
+ func (rc *raftNode) loadSnapshot() *raftpb.Snapshot {
+ if wal.Exist(rc.walDir) {
+ walSnaps, err := wal.ValidSnapshotEntries(logger.Get(), rc.walDir)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error listing snapshots")
+ }
+ snapshot, err := rc.snapshotter.LoadNewestAvailable(walSnaps)
+ if err != nil && !errors.Is(err, snap.ErrNoSnapshot) {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading snapshot")
+ }
+ return snapshot
+ }
+ return &raftpb.Snapshot{}
+ }
+
+ // openWAL returns a WAL ready for reading.
+ func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL {
+ if !wal.Exist(rc.walDir) {
+ if err := os.Mkdir(rc.walDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for WAL")
+ }
+
+ w, err := wal.Create(zap.NewExample(), rc.walDir, nil)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Create WAL error")
+ }
+ _ = w.Close()
+ }
+
+ walSnap := walpb.Snapshot{}
+ if snapshot != nil {
+ walSnap.Index, walSnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
+ }
+ logger.Get().Sugar().Infof("Loading WAL at term %d and index %d", walSnap.Term, walSnap.Index)
+ w, err := wal.Open(logger.Get(), rc.walDir, walSnap)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading WAL")
+ }
+
+ return w
+ }
+
+ // replayWAL replays WAL entries into the raft instance.
+ func (rc *raftNode) replayWAL() *wal.WAL {
+ logger.Get().Sugar().Infof("replaying WAL of member %d", rc.id)
+ snapshot := rc.loadSnapshot()
+ w := rc.openWAL(snapshot)
+ _, st, ents, err := w.ReadAll()
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Failed to read WAL")
+ }
+ rc.raftStorage = raft.NewMemoryStorage()
+ if snapshot != nil {
+ _ = rc.raftStorage.ApplySnapshot(*snapshot)
+ }
+ _ = rc.raftStorage.SetHardState(st)
+
+ // append to storage so raft starts at the right place in log
+ _ = rc.raftStorage.Append(ents)
+
+ return w
+ }
+
+ func (rc *raftNode) writeError(err error) {
+ rc.stopHTTP()
+ close(rc.commitC)
+ rc.errorC <- err
+ close(rc.errorC)
+ rc.node.Stop()
+ }
+
+ func (rc *raftNode) startRaft() {
+ if !fileutil.Exist(rc.snapDir) {
+ if err := os.Mkdir(rc.snapDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for snapshot")
+ }
+ }
+ rc.snapshotter = snap.New(logger.Get(), rc.snapDir)
+
+ oldwal := wal.Exist(rc.walDir)
+ rc.wal = rc.replayWAL()
+
+ // signal replay has finished
+ rc.snapshotterReady <- rc.snapshotter
+
+ rpeers := make([]raft.Peer, len(rc.peers))
+ for i := range rpeers {
+ rpeers[i] = raft.Peer{ID: uint64(i + 1)}
+ }
+ c := &raft.Config{
+ ID: uint64(rc.id),
+ ElectionTick: 10,
+ HeartbeatTick: 1,
+ Storage: rc.raftStorage,
+ MaxSizePerMsg: 1024 * 1024,
+ MaxInflightMsgs: 256,
+ MaxUncommittedEntriesSize: 1 << 30,
+ }
+
+ if oldwal || rc.join {
+ rc.node = raft.RestartNode(c)
+ } else {
+ rc.node = raft.StartNode(c, rpeers)
+ }
+
+ rc.transport = &rafthttp.Transport{
+ Logger: logger.Get(),
+ ID: types.ID(rc.id),
+ ClusterID: 0x1000,
+ Raft: rc,
+ ServerStats: stats.NewServerStats("", ""),
+ LeaderStats: stats.NewLeaderStats(zap.NewExample(), strconv.Itoa(rc.id)),
+ ErrorC: make(chan error),
+
+ DialRetryFrequency: 1,
+ }
+
+ if err := rc.transport.Start(); err != nil {
+ logger.Get().With(zap.Error(err)).Panic("Failed to start raft http server")
+ }
+ for i := range rc.peers {
+ if i+1 != rc.id {
+ rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
+ }
+ }
+
+ go rc.serveRaft()
+ go rc.serveChannels()
+ }
+
+ // stop closes http, closes all channels, and stops raft.
+ func (rc *raftNode) stop() {
+ rc.stopHTTP()
+ close(rc.commitC)
+ close(rc.errorC)
+ rc.node.Stop()
+ }
+
+ func (rc *raftNode) stopHTTP() {
+ rc.transport.Stop()
+ close(rc.httpStopCh)
+ <-rc.httpDoneCh
+ }
+
+ func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) {
+ if raft.IsEmptySnap(snapshotToSave) {
+ return
+ }
+
+ logger.Get().Sugar().Infof("Publishing snapshot at index %d", rc.snapshotIndex)
+ defer logger.Get().Sugar().Infof("Finished publishing snapshot at index %d", rc.snapshotIndex)
+
+ if snapshotToSave.Metadata.Index <= rc.appliedIndex {
+ logger.Get().Sugar().Fatalf("snapshot index [%d] should > progress.appliedIndex [%d]", snapshotToSave.Metadata.Index, rc.appliedIndex)
+ }
+ rc.commitC <- nil // trigger kvstore to load snapshot
+
+ rc.confState = snapshotToSave.Metadata.ConfState
+ rc.snapshotIndex = snapshotToSave.Metadata.Index
+ rc.appliedIndex = snapshotToSave.Metadata.Index
+ }
+
+ var snapshotCatchUpEntriesN uint64 = 10000
+
+ func (rc *raftNode) maybeTriggerSnapshot(applyDoneC <-chan struct{}) {
+ if rc.appliedIndex-rc.snapshotIndex <= rc.snapCount {
+ return
+ }
+
+ // wait until all committed entries are applied (or server is closed)
+ if applyDoneC != nil {
+ select {
+ case <-applyDoneC:
+ case <-rc.stopCh:
+ return
+ }
+ }
+
+ logger.Get().Sugar().Infof("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex)
+ data, err := rc.getSnapshot()
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Panic("Failed to snapshot")
+ }
+ snapshot, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data)
+ if err != nil {
+ panic(err)
Review Comment:
Why do we use panic here and followings?
##########
store/engine/embedded/raft.go:
##########
@@ -0,0 +1,550 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 embedded
+
+ import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/apache/kvrocks-controller/logger"
+ "go.etcd.io/etcd/raft/v3"
+ "go.etcd.io/etcd/raft/v3/raftpb"
+ "go.uber.org/atomic"
+
+ "go.etcd.io/etcd/client/pkg/v3/fileutil"
+ "go.etcd.io/etcd/client/pkg/v3/types"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/rafthttp"
+ "go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+ stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"
+ "go.etcd.io/etcd/server/v3/wal"
+ "go.etcd.io/etcd/server/v3/wal/walpb"
+ "go.uber.org/zap"
+ )
+
+ type commit struct {
+ data []string
+ applyDoneC chan<- struct{}
+ }
+
+ // A key-value stream backed by raft
+ type raftNode struct {
+ proposeC <-chan string // proposed messages (k,v)
+ confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
+ leaderChangeCh chan<- bool // leader changes
+ commitC chan<- *commit // entries committed to log (k,v)
+ errorC chan<- error // errors from raft session
+
+ id int // client ID for raft session
+ peers []string // raft peer URLs
+ join bool // node is joining an existing cluster
+ walDir string // path to WAL directory
+ snapDir string // path to snapshot directory
+ getSnapshot func() ([]byte, error)
+
+ isLeader atomic.Bool
+ leader atomic.Uint64
+
+ confState raftpb.ConfState
+ snapshotIndex uint64
+ appliedIndex uint64
+
+ // raft backing for the commit/error channel
+ node raft.Node
+ raftStorage *raft.MemoryStorage
+ wal *wal.WAL
+
+ snapshotter *snap.Snapshotter
+ snapshotterReady chan<- *snap.Snapshotter // signals when snapshotter is ready
+
+ snapCount uint64
+ transport *rafthttp.Transport
+ stopCh chan struct{} // signals proposal channel closed
+ httpStopCh chan struct{} // signals http server to shutdown
+ httpDoneCh chan struct{} // signals http server shutdown complete
+ }
+
+ var defaultSnapshotCount uint64 = 10000
+
+ // newRaftNode initiates a raft instance and returns a committed log entry
+ // channel and error channel. Proposals for log updates are sent over the
+ // provided the proposal channel. All log entries are replayed over the
+ // commit channel, followed by a nil message (to indicate the channel is
+ // current), then new log entries. To shutdown, close proposeC and read errorC.
+ func newRaftNode(
+ id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string,
+ confChangeC <-chan raftpb.ConfChange, leaderChangeCh chan<- bool, commitC chan<- *commit, errorC chan<- error,
+ snapshotterReady chan<- *snap.Snapshotter,
+ ) *raftNode {
+
+ rc := &raftNode{
+ proposeC: proposeC,
+ confChangeC: confChangeC,
+ leaderChangeCh: leaderChangeCh,
+ commitC: commitC,
+ errorC: errorC,
+ id: id,
+ peers: peers,
+ join: join,
+ walDir: fmt.Sprintf("raftexample-%d", id),
+ snapDir: fmt.Sprintf("raftexample-%d-snap", id),
+ getSnapshot: getSnapshot,
+ snapCount: defaultSnapshotCount,
+ stopCh: make(chan struct{}),
+ httpStopCh: make(chan struct{}),
+ httpDoneCh: make(chan struct{}),
+
+ snapshotterReady: snapshotterReady,
+ // rest of structure populated after WAL replay
+ }
+ rc.isLeader.Store(false)
+ go rc.startRaft()
+ return rc
+ }
+
+ func (rc *raftNode) saveSnap(snap raftpb.Snapshot) error {
+ walSnap := walpb.Snapshot{
+ Index: snap.Metadata.Index,
+ Term: snap.Metadata.Term,
+ ConfState: &snap.Metadata.ConfState,
+ }
+ // save the snapshot file before writing the snapshot to the wal.
+ // This makes it possible for the snapshot file to become orphaned, but prevents
+ // a WAL snapshot entry from having no corresponding snapshot file.
+ if err := rc.snapshotter.SaveSnap(snap); err != nil {
+ return err
+ }
+ if err := rc.wal.SaveSnapshot(walSnap); err != nil {
+ return err
+ }
+ return rc.wal.ReleaseLockTo(snap.Metadata.Index)
+ }
+
+ func (rc *raftNode) entriesToApply(ents []raftpb.Entry) (nents []raftpb.Entry) {
+ if len(ents) == 0 {
+ return ents
+ }
+ firstIdx := ents[0].Index
+ if firstIdx > rc.appliedIndex+1 {
+ logger.Get().Sugar().Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex)
+ }
+ if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) {
+ nents = ents[rc.appliedIndex-firstIdx+1:]
+ }
+ return nents
+ }
+
+ // publishEntries writes committed log entries to commit channel and returns
+ // whether all entries could be published.
+ func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {
+ if len(ents) == 0 {
+ return nil, true
+ }
+
+ data := make([]string, 0, len(ents))
+ for i := range ents {
+ switch ents[i].Type {
+ case raftpb.EntryNormal:
+ if len(ents[i].Data) == 0 {
+ // ignore empty messages
+ break
+ }
+ s := string(ents[i].Data)
+ data = append(data, s)
+ case raftpb.EntryConfChange:
+ var cc raftpb.ConfChange
+ _ = cc.Unmarshal(ents[i].Data)
+ rc.confState = *rc.node.ApplyConfChange(cc)
+ switch cc.Type {
+ case raftpb.ConfChangeAddNode:
+ if len(cc.Context) > 0 {
+ rc.transport.AddPeer(types.ID(cc.NodeID), []string{string(cc.Context)})
+ }
+ case raftpb.ConfChangeRemoveNode:
+ if cc.NodeID == uint64(rc.id) {
+ logger.Get().Info("I've been removed from the cluster! Shutting down.")
+ return nil, false
+ }
+ rc.transport.RemovePeer(types.ID(cc.NodeID))
+ case raftpb.ConfChangeUpdateNode, raftpb.ConfChangeAddLearnerNode:
+ }
+ case raftpb.EntryConfChangeV2:
+ }
+ }
+
+ var applyDoneC chan struct{}
+
+ if len(data) > 0 {
+ applyDoneC = make(chan struct{}, 1)
+ select {
+ case rc.commitC <- &commit{data, applyDoneC}:
+ case <-rc.stopCh:
+ return nil, false
+ }
+ }
+
+ // after commit, update appliedIndex
+ rc.appliedIndex = ents[len(ents)-1].Index
+
+ return applyDoneC, true
+ }
+
+ func (rc *raftNode) loadSnapshot() *raftpb.Snapshot {
+ if wal.Exist(rc.walDir) {
+ walSnaps, err := wal.ValidSnapshotEntries(logger.Get(), rc.walDir)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error listing snapshots")
+ }
+ snapshot, err := rc.snapshotter.LoadNewestAvailable(walSnaps)
+ if err != nil && !errors.Is(err, snap.ErrNoSnapshot) {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading snapshot")
+ }
+ return snapshot
+ }
+ return &raftpb.Snapshot{}
+ }
+
+ // openWAL returns a WAL ready for reading.
+ func (rc *raftNode) openWAL(snapshot *raftpb.Snapshot) *wal.WAL {
+ if !wal.Exist(rc.walDir) {
+ if err := os.Mkdir(rc.walDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for WAL")
+ }
+
+ w, err := wal.Create(zap.NewExample(), rc.walDir, nil)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Create WAL error")
+ }
+ _ = w.Close()
+ }
+
+ walSnap := walpb.Snapshot{}
+ if snapshot != nil {
+ walSnap.Index, walSnap.Term = snapshot.Metadata.Index, snapshot.Metadata.Term
+ }
+ logger.Get().Sugar().Infof("Loading WAL at term %d and index %d", walSnap.Term, walSnap.Index)
+ w, err := wal.Open(logger.Get(), rc.walDir, walSnap)
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Error loading WAL")
+ }
+
+ return w
+ }
+
+ // replayWAL replays WAL entries into the raft instance.
+ func (rc *raftNode) replayWAL() *wal.WAL {
+ logger.Get().Sugar().Infof("replaying WAL of member %d", rc.id)
+ snapshot := rc.loadSnapshot()
+ w := rc.openWAL(snapshot)
+ _, st, ents, err := w.ReadAll()
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Failed to read WAL")
+ }
+ rc.raftStorage = raft.NewMemoryStorage()
+ if snapshot != nil {
+ _ = rc.raftStorage.ApplySnapshot(*snapshot)
+ }
+ _ = rc.raftStorage.SetHardState(st)
+
+ // append to storage so raft starts at the right place in log
+ _ = rc.raftStorage.Append(ents)
+
+ return w
+ }
+
+ func (rc *raftNode) writeError(err error) {
+ rc.stopHTTP()
+ close(rc.commitC)
+ rc.errorC <- err
+ close(rc.errorC)
+ rc.node.Stop()
+ }
+
+ func (rc *raftNode) startRaft() {
+ if !fileutil.Exist(rc.snapDir) {
+ if err := os.Mkdir(rc.snapDir, 0750); err != nil {
+ logger.Get().With(zap.Error(err)).Fatal("Cannot create dir for snapshot")
+ }
+ }
+ rc.snapshotter = snap.New(logger.Get(), rc.snapDir)
+
+ oldwal := wal.Exist(rc.walDir)
+ rc.wal = rc.replayWAL()
+
+ // signal replay has finished
+ rc.snapshotterReady <- rc.snapshotter
+
+ rpeers := make([]raft.Peer, len(rc.peers))
+ for i := range rpeers {
+ rpeers[i] = raft.Peer{ID: uint64(i + 1)}
+ }
+ c := &raft.Config{
+ ID: uint64(rc.id),
+ ElectionTick: 10,
+ HeartbeatTick: 1,
+ Storage: rc.raftStorage,
+ MaxSizePerMsg: 1024 * 1024,
+ MaxInflightMsgs: 256,
+ MaxUncommittedEntriesSize: 1 << 30,
+ }
+
+ if oldwal || rc.join {
+ rc.node = raft.RestartNode(c)
+ } else {
+ rc.node = raft.StartNode(c, rpeers)
+ }
+
+ rc.transport = &rafthttp.Transport{
+ Logger: logger.Get(),
+ ID: types.ID(rc.id),
+ ClusterID: 0x1000,
+ Raft: rc,
+ ServerStats: stats.NewServerStats("", ""),
+ LeaderStats: stats.NewLeaderStats(zap.NewExample(), strconv.Itoa(rc.id)),
+ ErrorC: make(chan error),
+
+ DialRetryFrequency: 1,
+ }
+
+ if err := rc.transport.Start(); err != nil {
+ logger.Get().With(zap.Error(err)).Panic("Failed to start raft http server")
+ }
+ for i := range rc.peers {
+ if i+1 != rc.id {
+ rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
+ }
+ }
+
+ go rc.serveRaft()
+ go rc.serveChannels()
+ }
+
+ // stop closes http, closes all channels, and stops raft.
+ func (rc *raftNode) stop() {
+ rc.stopHTTP()
+ close(rc.commitC)
+ close(rc.errorC)
+ rc.node.Stop()
+ }
+
+ func (rc *raftNode) stopHTTP() {
+ rc.transport.Stop()
+ close(rc.httpStopCh)
+ <-rc.httpDoneCh
+ }
+
+ func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) {
+ if raft.IsEmptySnap(snapshotToSave) {
+ return
+ }
+
+ logger.Get().Sugar().Infof("Publishing snapshot at index %d", rc.snapshotIndex)
+ defer logger.Get().Sugar().Infof("Finished publishing snapshot at index %d", rc.snapshotIndex)
+
+ if snapshotToSave.Metadata.Index <= rc.appliedIndex {
+ logger.Get().Sugar().Fatalf("snapshot index [%d] should > progress.appliedIndex [%d]", snapshotToSave.Metadata.Index, rc.appliedIndex)
+ }
+ rc.commitC <- nil // trigger kvstore to load snapshot
+
+ rc.confState = snapshotToSave.Metadata.ConfState
+ rc.snapshotIndex = snapshotToSave.Metadata.Index
+ rc.appliedIndex = snapshotToSave.Metadata.Index
+ }
+
+ var snapshotCatchUpEntriesN uint64 = 10000
+
+ func (rc *raftNode) maybeTriggerSnapshot(applyDoneC <-chan struct{}) {
+ if rc.appliedIndex-rc.snapshotIndex <= rc.snapCount {
+ return
+ }
+
+ // wait until all committed entries are applied (or server is closed)
+ if applyDoneC != nil {
+ select {
+ case <-applyDoneC:
+ case <-rc.stopCh:
+ return
+ }
+ }
+
+ logger.Get().Sugar().Infof("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex)
+ data, err := rc.getSnapshot()
+ if err != nil {
+ logger.Get().With(zap.Error(err)).Panic("Failed to snapshot")
+ }
+ snapshot, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data)
+ if err != nil {
+ panic(err)
+ }
+ if err := rc.saveSnap(snapshot); err != nil {
+ panic(err)
+ }
+
+ compactIndex := uint64(1)
+ if rc.appliedIndex > snapshotCatchUpEntriesN {
+ compactIndex = rc.appliedIndex - snapshotCatchUpEntriesN
+ }
+ if err := rc.raftStorage.Compact(compactIndex); err != nil {
+ if !errors.Is(err, raft.ErrCompacted) {
+ panic(err)
+ }
+ } else {
+ logger.Get().Sugar().Infof("compacted log at index %d", compactIndex)
+ }
+
+ rc.snapshotIndex = rc.appliedIndex
+ }
+
+ func (rc *raftNode) serveChannels() {
Review Comment:
What's the meaning of Channels here? would you mind refactoring this function to make it shorter by moving the routine into a single function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org