Refactor raftexample using interfaces to better separate concerns

example_test.go

@@ -18,10 +18,10 @@ import (
 	pb "go.etcd.io/raft/v3/raftpb"
 )
 
-func applyToStore(ents []pb.Entry)    {}
-func sendMessages(msgs []pb.Message)  {}
-func saveStateToDisk(st pb.HardState) {}
-func saveToDisk(ents []pb.Entry)      {}
+func applyToStore(_ []pb.Entry)      {}
+func sendMessages(_ []pb.Message)    {}
+func saveStateToDisk(_ pb.HardState) {}
+func saveToDisk(_ []pb.Entry)        {}
 
 func ExampleNode() {
 	c := &Config{}

node_test.go

@@ -1090,7 +1090,7 @@ type ignoreSizeHintMemStorage struct {
 	*MemoryStorage
 }
 
-func (s *ignoreSizeHintMemStorage) Entries(lo, hi uint64, maxSize uint64) ([]raftpb.Entry, error) {
+func (s *ignoreSizeHintMemStorage) Entries(lo, hi uint64, _ uint64) ([]raftpb.Entry, error) {
 	return s.MemoryStorage.Entries(lo, hi, math.MaxUint64)
 }
 

raftexample/etcdserverpb/raft_internal_stringer.go

@@ -78,6 +78,7 @@ type txnRequestStringer struct {
 	Request *TxnRequest
 }
 
+//nolint:revive
 func NewLoggableTxnRequest(request *TxnRequest) *txnRequestStringer {
 	return &txnRequestStringer{request}
 }
@@ -167,6 +168,7 @@ type loggablePutRequest struct {
 	IgnoreLease bool   `protobuf:"varint,6,opt,name=ignore_lease,proto3"`
 }
 
+//nolint:revive
 func NewLoggablePutRequest(request *PutRequest) *loggablePutRequest {
 	return &loggablePutRequest{
 		request.Key,

raftexample/fileutil/lock_linux.go

@@ -0,0 +1,92 @@
+// Copyright 2016 The etcd 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.
+
+//go:build linux
+
+package fileutil
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"syscall"
+
+	"golang.org/x/sys/unix"
+)
+
+// This used to call syscall.Flock() but that call fails with EBADF on NFS.
+// An alternative is lockf() which works on NFS but that call lets a process lock
+// the same file twice. Instead, use Linux's non-standard open file descriptor
+// locks which will block if the process already holds the file lock.
+
+var (
+	wrlck = syscall.Flock_t{
+		Type:   syscall.F_WRLCK,
+		Whence: int16(io.SeekStart),
+		Start:  0,
+		Len:    0,
+	}
+
+	linuxTryLockFile = flockTryLockFile
+	linuxLockFile    = flockLockFile
+)
+
+func init() {
+	// use open file descriptor locks if the system supports it
+	getlk := syscall.Flock_t{Type: syscall.F_RDLCK}
+	if err := syscall.FcntlFlock(0, unix.F_OFD_GETLK, &getlk); err == nil {
+		linuxTryLockFile = ofdTryLockFile
+		linuxLockFile = ofdLockFile
+	}
+}
+
+func TryLockFile(path string, flag int, perm os.FileMode) (*LockedFile, error) {
+	return linuxTryLockFile(path, flag, perm)
+}
+
+func ofdTryLockFile(path string, flag int, perm os.FileMode) (*LockedFile, error) {
+	f, err := os.OpenFile(path, flag, perm)
+	if err != nil {
+		return nil, fmt.Errorf("ofdTryLockFile failed to open %q (%v)", path, err)
+	}
+
+	flock := wrlck
+	if err = syscall.FcntlFlock(f.Fd(), unix.F_OFD_SETLK, &flock); err != nil {
+		f.Close()
+		if err == syscall.EWOULDBLOCK {
+			err = ErrLocked
+		}
+		return nil, err
+	}
+	return &LockedFile{f}, nil
+}
+
+func LockFile(path string, flag int, perm os.FileMode) (*LockedFile, error) {
+	return linuxLockFile(path, flag, perm)
+}
+
+func ofdLockFile(path string, flag int, perm os.FileMode) (*LockedFile, error) {
+	f, err := os.OpenFile(path, flag, perm)
+	if err != nil {
+		return nil, fmt.Errorf("ofdLockFile failed to open %q (%v)", path, err)
+	}
+
+	flock := wrlck
+	err = syscall.FcntlFlock(f.Fd(), unix.F_OFD_SETLKW, &flock)
+	if err != nil {
+		f.Close()
+		return nil, err
+	}
+	return &LockedFile{f}, nil
+}

raftexample/httpapi.go

@@ -23,9 +23,22 @@ import (
 	"go.etcd.io/raft/v3/raftpb"
 )
 
+// KVStore is the interface to the key-value store that is required by
+// `httpKVAPI`.
+type KVStore interface {
+	// Propose proposes to set key `k` to value `v` in the key-value
+	// store. It returns immediately, before the change has
+	// necessarily taken effect.
+	Propose(k, v string)
+
+	// Lookup looks up the value for key `k` in the current state of
+	// the store.
+	Lookup(k string) (string, bool)
+}
+
 // Handler for a http based key-value store backed by raft
 type httpKVAPI struct {
-	store       *kvstore
+	store       KVStore
 	confChangeC chan<- raftpb.ConfChange
 }
 
@@ -101,7 +114,9 @@ func (h *httpKVAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 }
 
 // serveHTTPKVAPI starts a key-value server with a GET/PUT API and listens.
-func serveHTTPKVAPI(kv *kvstore, port int, confChangeC chan<- raftpb.ConfChange, errorC <-chan error) {
+func serveHTTPKVAPI(
+	kv KVStore, port int, confChangeC chan<- raftpb.ConfChange, done <-chan struct{},
+) {
 	srv := http.Server{
 		Addr: ":" + strconv.Itoa(port),
 		Handler: &httpKVAPI{
@@ -116,7 +131,6 @@ func serveHTTPKVAPI(kv *kvstore, port int, confChangeC chan<- raftpb.ConfChange,
 	}()
 
 	// exit when raft goes down
-	if err, ok := <-errorC; ok {
-		log.Fatal(err)
-	}
+	<-done
+	_ = srv.Close()
 }

raftexample/kvstore.go

@@ -18,42 +18,36 @@ import (
 	"bytes"
 	"encoding/gob"
 	"encoding/json"
+	"fmt"
 	"log"
 	"strings"
 	"sync"
-
-	"go.etcd.io/raft/v3/raftexample/snap"
-	"go.etcd.io/raft/v3/raftpb"
 )
 
 // a key-value store backed by raft
 type kvstore struct {
-	proposeC    chan<- string // channel for proposing updates
-	mu          sync.RWMutex
-	kvStore     map[string]string // current committed key-value pairs
-	snapshotter *snap.Snapshotter
+	proposeC chan<- string // channel for proposing updates
+	mu       sync.RWMutex
+	kvStore  map[string]string // current committed key-value pairs
 }
 
 type kv struct {
 	Key string
 	Val string
 }
 
-func newKVStore(snapshotter *snap.Snapshotter, proposeC chan<- string, commitC <-chan *commit, errorC <-chan error) *kvstore {
-	s := &kvstore{proposeC: proposeC, kvStore: make(map[string]string), snapshotter: snapshotter}
-	snapshot, err := s.loadSnapshot()
-	if err != nil {
-		log.Panic(err)
+// newKVStore creates and returns a new `kvstore`. The second return
+// value can be used as the finite state machine that is driven by a
+// `raftNode`.
+func newKVStore(proposeC chan<- string) (*kvstore, kvfsm) {
+	s := &kvstore{
+		proposeC: proposeC,
+		kvStore:  make(map[string]string),
 	}
-	if snapshot != nil {
-		log.Printf("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
-		if err := s.recoverFromSnapshot(snapshot.Data); err != nil {
-			log.Panic(err)
-		}
+	fsm := kvfsm{
+		kvs: s,
 	}
-	// read commits from raft into kvStore map until error
-	go s.readCommits(commitC, errorC)
-	return s
+	return s, fsm
 }
 
 func (s *kvstore) Lookup(key string) (string, bool) {
@@ -71,64 +65,53 @@ func (s *kvstore) Propose(k string, v string) {
 	s.proposeC <- buf.String()
 }
 
-func (s *kvstore) readCommits(commitC <-chan *commit, errorC <-chan error) {
-	for commit := range commitC {
-		if commit == nil {
-			// signaled to load snapshot
-			snapshot, err := s.loadSnapshot()
-			if err != nil {
-				log.Panic(err)
-			}
-			if snapshot != nil {
-				log.Printf("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
-				if err := s.recoverFromSnapshot(snapshot.Data); err != nil {
-					log.Panic(err)
-				}
-			}
-			continue
-		}
-
-		for _, data := range commit.data {
-			var dataKv kv
-			dec := gob.NewDecoder(bytes.NewBufferString(data))
-			if err := dec.Decode(&dataKv); err != nil {
-				log.Fatalf("raftexample: could not decode message (%v)", err)
-			}
-			s.mu.Lock()
-			s.kvStore[dataKv.Key] = dataKv.Val
-			s.mu.Unlock()
-		}
-		close(commit.applyDoneC)
-	}
-	if err, ok := <-errorC; ok {
-		log.Fatal(err)
-	}
+// Set sets a single value. It should only be called by `kvfsm`.
+func (s *kvstore) set(k, v string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kvStore[k] = v
 }
 
-func (s *kvstore) getSnapshot() ([]byte, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	return json.Marshal(s.kvStore)
+func (s *kvstore) restoreFromSnapshot(store map[string]string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.kvStore = store
 }
 
-func (s *kvstore) loadSnapshot() (*raftpb.Snapshot, error) {
-	snapshot, err := s.snapshotter.Load()
-	if err == snap.ErrNoSnapshot {
-		return nil, nil
-	}
-	if err != nil {
-		return nil, err
-	}
-	return snapshot, nil
+// kvfsm implements the `FSM` interface for the underlying `*kvstore`.
+type kvfsm struct {
+	kvs *kvstore
 }
 
-func (s *kvstore) recoverFromSnapshot(snapshot []byte) error {
+// RestoreSnapshot restores the current state of the KV store to the
+// value encoded in `snapshot`.
+func (fsm kvfsm) RestoreSnapshot(snapshot []byte) error {
 	var store map[string]string
 	if err := json.Unmarshal(snapshot, &store); err != nil {
 		return err
 	}
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.kvStore = store
+	fsm.kvs.restoreFromSnapshot(store)
+	return nil
+}
+
+func (fsm kvfsm) TakeSnapshot() ([]byte, error) {
+	fsm.kvs.mu.RLock()
+	defer fsm.kvs.mu.RUnlock()
+	return json.Marshal(fsm.kvs.kvStore)
+}
+
+// ApplyCommits decodes and applies each of the commits in `commit` to
+// the current state, then signals that it is done by closing
+// `commit.applyDoneC`.
+func (fsm kvfsm) ApplyCommits(commit *commit) error {
+	for _, data := range commit.data {
+		var dataKv kv
+		dec := gob.NewDecoder(bytes.NewBufferString(data))
+		if err := dec.Decode(&dataKv); err != nil {
+			return fmt.Errorf("could not decode message: %w", err)
+		}
+		fsm.kvs.set(dataKv.Key, dataKv.Val)
+	}
+	close(commit.applyDoneC)
 	return nil
 }

raftexample/kvstore_test.go

@@ -22,19 +22,22 @@ import (
 func Test_kvstore_snapshot(t *testing.T) {
 	tm := map[string]string{"foo": "bar"}
 	s := &kvstore{kvStore: tm}
+	fsm := kvfsm{
+		kvs: s,
+	}
 
 	v, _ := s.Lookup("foo")
 	if v != "bar" {
 		t.Fatalf("foo has unexpected value, got %s", v)
 	}
 
-	data, err := s.getSnapshot()
+	data, err := fsm.TakeSnapshot()
 	if err != nil {
 		t.Fatal(err)
 	}
 	s.kvStore = nil
 
-	if err := s.recoverFromSnapshot(data); err != nil {
+	if err := fsm.RestoreSnapshot(data); err != nil {
 		t.Fatal(err)
 	}
 	v, _ = s.Lookup("foo")