diff --git a/contrib/raftsimple/Procfile b/contrib/raftsimple/Procfile new file mode 100644 index 000000000..6d1f17fd6 --- /dev/null +++ b/contrib/raftsimple/Procfile @@ -0,0 +1,3 @@ +raft1: ./raftsimple --id 1 --cluster http://127.0.0.1:9021,http://127.0.0.1:9022,http://127.0.0.1:9023 --port 9121 +raft2: ./raftsimple --id 2 --cluster http://127.0.0.1:9021,http://127.0.0.1:9022,http://127.0.0.1:9023 --port 9122 +raft3: ./raftsimple --id 3 --cluster http://127.0.0.1:9021,http://127.0.0.1:9022,http://127.0.0.1:9023 --port 9123 diff --git a/contrib/raftsimple/README.md b/contrib/raftsimple/README.md new file mode 100644 index 000000000..a13331707 --- /dev/null +++ b/contrib/raftsimple/README.md @@ -0,0 +1,109 @@ +# raftsimple + + + +This project is designed for developers who want to understand the core mechanics of the Raft algorithm and the `Ready()` event loop without the complexity of the full etcd server ecosystem. + +## Getting started + +### Building raftsimple + +Build the binary: + +```bash +go build -o raftsimple +``` + +### Running single-node raftsimple + +To start a single-node Raft cluster, run: + +```bash +./raftsimple --id 1 --cluster http://127.0.0.1:9021 --port 9121 +``` + +This starts a node with ID 1, listening for Raft messages on `127.0.0.1:9021`, and exposing a key-value API on port `9121`. + +You can now interact with the key-value store: + +```bash +# Propose a value +curl -L http://127.0.0.1:9121/foo -XPUT -d bar + +# Retrieve the value +curl -L http://127.0.0.1:9121/foo +``` + +### Running a local cluster + +For a more realistic distributed setup, use [goreman](https://github.com/mattn/goreman) to run a 3-node cluster: + +```bash +goreman start +``` + +This uses the included `Procfile` to start three independent `raftsimple` processes on different ports. Each node can be communicated with via its respective key-value API at ports `9121`, `9122`, and `9123`. + +### Fault Tolerance + +Raft's key feature is its ability to withstand node failures. You can simulate a node crash with `goreman` (it uses Procfile file): + +```bash +# Stop Node 2 +goreman run stop raft2 +``` + +The cluster will still operate with Node 1 and Node 3. Propose a new value on Node 1: + +```bash +curl -L http://127.0.0.1:9121/foo -XPUT -d baz +``` + +Now restart Node 2: + +```bash +goreman run restart raft2 +``` + +Node 2 will automatically recover its state from its local WAL and snapshot, then catch up on any missed data from the leader. You can verify it has recovered: + +```bash +curl -L http://127.0.0.1:9122/foo +``` + +### Dynamic Reconfiguration + +`raftsimple` also supports membership changes via the KV API. To add a new node (Node 4) to the cluster: + +```bash +# Register Node 4 with the cluster, providing its URL in the request body +curl -L http://127.0.0.1:9121/4 -XPOST -d http://127.0.0.1:9024 + +# Start Node 4 in a separate terminal with the --join flag +./raftsimple --id 4 --cluster http://127.0.0.1:9021,http://127.0.0.1:9022,http://127.0.0.1:9023,http://127.0.0.1:9024 --port 9124 --join +``` + +Similarly, to remove a node from the cluster: + +```bash +curl -L http://127.0.0.1:9121/4 -XDELETE +``` + +## Design + +The `raftsimple` architecture is divided into three main components: + +1. **KV Store (`kvstore.go`):** The application-level state machine. It manages the actual key-value data and provides methods for proposing changes to the Raft log and applying committed entries. +2. **REST API (`httpapi.go`):** The user-facing interface. It handles incoming HTTP requests and converts them into proposals for the Raft node. +3. **Raft Node (`raft.go`):** The core consensus engine. It manages the `etcd/raft` state machine, handles the `Ready()` loop, and coordinates with the transport and storage layers. + +## Project Philosophy: Simplicity First + +`raftsimple` is designed as an educational tool, not a production-grade database. To keep the code as readable and linear as possible, we have made deliberate design choices that prioritize **simplicity over robustness**: + +* **Synchronous Storage:** Unlike production systems that use complex asynchronous I/O and batching, `raftsimple` writes to disk synchronously within the main Raft loop. +* **Minimalist WAL:** The Write-Ahead Log is a simple append-only file. It does not include checksums, CRCs, or automatic truncation after snapshots. +* **Simple Transport:** The inter-node communication uses standard Go `net/http` for simplicity, rather than the more complex but efficient `rafthttp` package used in `etcd`. +* **Direct File I/O:** Snapshots and HardState are overwritten directly on disk. + +These choices make the codebase significantly easier to audit and learn from, though they mean the project should not be used for storing critical data. diff --git a/contrib/raftsimple/go.mod b/contrib/raftsimple/go.mod new file mode 100644 index 000000000..9b977ed68 --- /dev/null +++ b/contrib/raftsimple/go.mod @@ -0,0 +1,17 @@ +module raftsimple + +go 1.25.7 + +require ( + github.com/stretchr/testify v1.10.0 + go.etcd.io/raft/v3 v3.6.0 +) + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/protobuf v1.5.4 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + google.golang.org/protobuf v1.33.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/contrib/raftsimple/go.sum b/contrib/raftsimple/go.sum new file mode 100644 index 000000000..b8105a37b --- /dev/null +++ b/contrib/raftsimple/go.sum @@ -0,0 +1,51 @@ +github.com/cockroachdb/datadriven v1.0.2 h1:H9MtNqVoVhvd9nCBwOyDjUEdZCREqbIdCJD93PBm/jA= +github.com/cockroachdb/datadriven v1.0.2/go.mod h1:a9RdTaap04u637JoCzcUoIcDmvwSUtcUFtT/C3kJlTU= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg= +github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.etcd.io/raft/v3 v3.6.0 h1:5NtvbDVYpnfZWcIHgGRk9DyzkBIXOi8j+DDp1IcnUWQ= +go.etcd.io/raft/v3 v3.6.0/go.mod h1:nLvLevg6+xrVtHUmVaTcTz603gQPHfh7kUAwV6YpfGo= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= +google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/contrib/raftsimple/httpapi.go b/contrib/raftsimple/httpapi.go new file mode 100644 index 000000000..7cf0039b2 --- /dev/null +++ b/contrib/raftsimple/httpapi.go @@ -0,0 +1,107 @@ +package main + +import ( + "io" + "log" + "net/http" + "strconv" + + "go.etcd.io/raft/v3/raftpb" +) + +type httpKVAPI struct { + store *kvstore + confChangeC chan<- raftpb.ConfChange +} + +func (h *httpKVAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) { + key := r.RequestURI + defer r.Body.Close() + + switch r.Method { + case http.MethodPut: + v, err := io.ReadAll(r.Body) + if err != nil { + log.Printf("Failed to read on PUT (%v)\n", err) + http.Error(w, "Failed to PUT", http.StatusBadRequest) + return + } + h.store.Propose(key, string(v)) + w.WriteHeader(http.StatusNoContent) + + case http.MethodGet: + if v, ok := h.store.Lookup(key); ok { + w.Write([]byte(v)) + } else { + http.Error(w, "Failed to GET", http.StatusNotFound) + } + + case http.MethodPost: + nodeID, err := strconv.ParseUint(key[1:], 0, 64) + if err != nil { + log.Printf("Failed to convert ID for conf change (%v)\n", err) + http.Error(w, "Failed on POST", http.StatusBadRequest) + return + } + + url, err := io.ReadAll(r.Body) + if err != nil { + log.Printf("Failed to read body for conf change (%v)\n", err) + http.Error(w, "Failed on POST", http.StatusBadRequest) + return + } + + cc := raftpb.ConfChange{ + Type: raftpb.ConfChangeAddNode, + NodeID: nodeID, + Context: url, + } + h.confChangeC <- cc + + w.WriteHeader(http.StatusNoContent) + + case http.MethodDelete: + nodeID, err := strconv.ParseUint(key[1:], 0, 64) + if err != nil { + log.Printf("Failed to convert ID for conf change (%v)\n", err) + http.Error(w, "Failed on DELETE", http.StatusBadRequest) + return + } + + cc := raftpb.ConfChange{ + Type: raftpb.ConfChangeRemoveNode, + NodeID: nodeID, + } + h.confChangeC <- cc + + w.WriteHeader(http.StatusNoContent) + + default: + w.Header().Set("Allow", http.MethodPut) + w.Header().Add("Allow", http.MethodGet) + w.Header().Add("Allow", http.MethodPost) + w.Header().Add("Allow", http.MethodDelete) + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + } +} + +func serveHTTPKVAPI(kv *kvstore, port uint64, confChangeC chan<- raftpb.ConfChange, done <-chan struct{}) { + srv := http.Server{ + Addr: ":" + strconv.FormatUint(port, 10), + Handler: &httpKVAPI{ + store: kv, + confChangeC: confChangeC, + }, + } + go func() { + if err := srv.ListenAndServe(); err != nil { + if err != http.ErrServerClosed { + log.Fatal(err) + } + } + }() + + // exit when raft goes down + <-done + _ = srv.Close() +} diff --git a/contrib/raftsimple/integration_test.go b/contrib/raftsimple/integration_test.go new file mode 100644 index 000000000..9785d4191 --- /dev/null +++ b/contrib/raftsimple/integration_test.go @@ -0,0 +1,89 @@ +package main + +import ( + "fmt" + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.etcd.io/raft/v3/raftpb" +) + +func findLeader(nodes map[uint64]*raftNode) *raftNode { + for range 50 { + for _, rn := range nodes { + if rn.node != nil && rn.node.Status().Lead == rn.id { + return rn + } + } + time.Sleep(100 * time.Millisecond) + } + return nil +} + +func TestIntegration_Lifecycle(t *testing.T) { + tmpDir := t.TempDir() + defer os.RemoveAll(tmpDir) + + peers := []uint64{1, 2, 3} + peerURLs := make(map[uint64]string) + for _, id := range peers { + peerURLs[id] = fmt.Sprintf("http://127.0.0.1:%d", 10000+id) + } + + nodes := make(map[uint64]*raftNode) + kvstores := make(map[uint64]*kvstore) + + createNode := func(id uint64) { + snapdir := fmt.Sprintf("%s/node-%d", tmpDir, id) + ss, _ := newSnapshotStorage(snapdir) + proposeC := make(chan string) + confChangeC := make(chan raftpb.ConfChange) + kvs, fsm := newKVStore(proposeC) + rn := newRaftNode(id, peerURLs, false, fsm, ss, proposeC, confChangeC) + nodes[id] = rn + kvstores[id] = kvs + go rn.processCommits() + } + + for _, id := range peers { + createNode(id) + } + + defer func() { + for _, rn := range nodes { + rn.stop() + } + }() + + leader := findLeader(nodes) + require.NotNil(t, leader, "A leader should be elected") + kvstores[leader.id].Propose("key1", "val1") + + require.Eventually(t, func() bool { + v, ok := kvstores[3].Lookup("key1") + return ok && v == "val1" + }, 5*time.Second, 100*time.Millisecond, "Data should replicate to Node 3") + + // Simulate node 2 crash + nodes[2].stop() + + leader = findLeader(nodes) + require.NotNil(t, leader, "A leader should be elected after crash") + kvstores[leader.id].Propose("key2", "val2") + + require.Eventually(t, func() bool { + v, ok := kvstores[3].Lookup("key2") + return ok && v == "val2" + }, 5*time.Second, 100*time.Millisecond, "Data should replicate to Node 3 while Node 2 is offline") + + // Restart node 2 + createNode(2) + + require.Eventually(t, func() bool { + v1, ok1 := kvstores[2].Lookup("key1") + v2, ok2 := kvstores[2].Lookup("key2") + return ok1 && ok2 && v1 == "val1" && v2 == "val2" + }, 10*time.Second, 100*time.Millisecond, "Node 2 should recover and catch up on all data") +} diff --git a/contrib/raftsimple/kvstore.go b/contrib/raftsimple/kvstore.go new file mode 100644 index 000000000..76bfaad80 --- /dev/null +++ b/contrib/raftsimple/kvstore.go @@ -0,0 +1,91 @@ +package main + +import ( + "bytes" + "encoding/gob" + "encoding/json" + "fmt" + "log" + "strings" + "sync" +) + +type kvstore struct { + proposeC chan<- string + mu sync.RWMutex + kvStore map[string]string +} + +type kv struct { + Key string + Val string +} + +func newKVStore(proposeC chan<- string) (*kvstore, kvfsm) { + s := &kvstore{proposeC: proposeC, kvStore: make(map[string]string)} + fsm := kvfsm{kvs: s} + return s, fsm +} + +func (s *kvstore) Lookup(key string) (string, bool) { + s.mu.RLock() + defer s.mu.RUnlock() + v, ok := s.kvStore[key] + return v, ok +} + +func (s *kvstore) Propose(key string, value string) { + var buf strings.Builder + if err := gob.NewEncoder(&buf).Encode(kv{key, value}); err != nil { + log.Fatal(err) + } + s.proposeC <- buf.String() +} + +// 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) restoreFromSnapshot(store map[string]string) { + s.mu.Lock() + defer s.mu.Unlock() + s.kvStore = store +} + +// kvfsm implements the `FSM` interface for the underlying `*kvstore`. +type kvfsm struct { + kvs *kvstore +} + +// 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 + } + 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) +} + +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 +} diff --git a/contrib/raftsimple/kvstore_test.go b/contrib/raftsimple/kvstore_test.go new file mode 100644 index 000000000..7fb2dc1b2 --- /dev/null +++ b/contrib/raftsimple/kvstore_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "bytes" + "encoding/gob" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestKVStore(t *testing.T) { + mockedProposeC := make(chan string, 1) + s, _ := newKVStore(mockedProposeC) + s.kvStore["foo"] = "bar" + + v, _ := s.Lookup("foo") + require.Equalf(t, "bar", v, "foo has unexpected value, got %s", v) + + s.Propose("x", "y") + got := <-mockedProposeC + dec := gob.NewDecoder(bytes.NewBufferString(got)) + + var dataKv kv + err := dec.Decode(&dataKv) + require.NoError(t, err) + + require.Equal(t, "x", dataKv.Key) + require.Equal(t, "y", dataKv.Val) +} + +func TestFSM(t *testing.T) { + mockedProposeC := make(chan string, 1) + s, fsm := newKVStore(mockedProposeC) + s.kvStore["foo"] = "bar" + + data, err := fsm.TakeSnapshot() + require.NoError(t, err) + s.kvStore = make(map[string]string) + + err = fsm.RestoreSnapshot(data) + require.NoError(t, err) + v, _ := s.Lookup("foo") + require.Equalf(t, "bar", v, "foo has unexpected value, got %s", v) + + var buf strings.Builder + _ = gob.NewEncoder(&buf).Encode(kv{"x", "y"}) + c := &commit{data: []string{buf.String()}, applyDoneC: make(chan struct{})} + fsm.ApplyCommits(c) + + v, _ = s.Lookup("x") + require.Equalf(t, "y", v, "x has unexpected value, got %s", v) +} diff --git a/contrib/raftsimple/main.go b/contrib/raftsimple/main.go new file mode 100644 index 000000000..2a7b0f7e0 --- /dev/null +++ b/contrib/raftsimple/main.go @@ -0,0 +1,46 @@ +package main + +import ( + "flag" + "fmt" + "log" + "strings" + + "go.etcd.io/raft/v3/raftpb" +) + +func main() { + cluster := flag.String("cluster", "http://127.0.0.1:9021", "comma separated cluster peers") + id := flag.Uint64("id", 1, "node ID") + kvport := flag.Int("port", 9121, "key-value server port") + join := flag.Bool("join", false, "set to true to join an existing cluster") + flag.Parse() + + proposeC := make(chan string) + confChangeC := make(chan raftpb.ConfChange) + + // For this simple example, we assume node IDs are 1, 2, 3... corresponding to the list order. + peerURLs := make(map[uint64]string) + for i, url := range strings.Split(*cluster, ",") { + peerURLs[uint64(i+1)] = url + } + + snapdir := fmt.Sprintf("raftsimple-%d-snap", *id) + ss, err := newSnapshotStorage(snapdir) + if err != nil { + log.Fatalf("raftsimple: error creating storage: %v", err) + } + + kvs, fsm := newKVStore(proposeC) + + rn := newRaftNode(*id, peerURLs, *join, fsm, ss, proposeC, confChangeC) + + go func() { + if err := rn.processCommits(); err != nil { + log.Fatalf("raftsimple: processCommits exited: %v", err) + } + }() + + log.Printf("Node %d starting KV API on port %d", *id, *kvport) + serveHTTPKVAPI(kvs, uint64(*kvport), confChangeC, rn.donec) +} diff --git a/contrib/raftsimple/network.go b/contrib/raftsimple/network.go new file mode 100644 index 000000000..cb3f4f3d3 --- /dev/null +++ b/contrib/raftsimple/network.go @@ -0,0 +1,131 @@ +package main + +import ( + "bytes" + "context" + "io" + "log" + "net/http" + "sync" + + "go.etcd.io/raft/v3/raftpb" +) + +type transport struct { + mu sync.RWMutex + id uint64 + peers map[uint64]string + server *http.Server + node interface { + Step(ctx context.Context, m raftpb.Message) error + } +} + +func newTransport(id uint64, peers map[uint64]string) *transport { + return &transport{ + id: id, + peers: peers, + } +} + +func (t *transport) Handler() http.Handler { + return &raftHandler{transport: t} +} + +func (t *transport) Close() error { + t.mu.Lock() + defer t.mu.Unlock() + if t.server != nil { + return t.server.Close() + } + return nil +} + +type raftHandler struct { + transport *transport +} + +func (h *raftHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + data, err := io.ReadAll(r.Body) + if err != nil { + log.Printf("transport: failed to read request body: %v", err) + http.Error(w, "Failed to read body", http.StatusBadRequest) + return + } + + var m raftpb.Message + if err := m.Unmarshal(data); err != nil { + log.Printf("transport: failed to unmarshal message: %v", err) + http.Error(w, "Failed to unmarshal", http.StatusBadRequest) + return + } + + if h.transport.node == nil { + // Node not yet ready + w.WriteHeader(http.StatusServiceUnavailable) + return + } + + if err := h.transport.node.Step(context.TODO(), m); err != nil { + log.Printf("transport: failed to step message: %v", err) + http.Error(w, "Failed to process message", http.StatusInternalServerError) + return + } + + w.WriteHeader(http.StatusNoContent) +} + +func (t *transport) send(msgs []raftpb.Message) { + for _, m := range msgs { + if m.To == 0 { + continue + } + t.mu.RLock() + url, ok := t.peers[m.To] + t.mu.RUnlock() + + if !ok { + log.Printf("transport: no URL for node %d", m.To) + continue + } + + go t.sendHTTP(url, m) + } +} + +func (t *transport) sendHTTP(url string, m raftpb.Message) { + data, err := m.Marshal() + if err != nil { + log.Printf("transport: failed to marshal message: %v", err) + return + } + + resp, err := http.Post(url, "application/octet-stream", bytes.NewReader(data)) + if err != nil { + // Suppress logging of connection errors to keep the output clean during churn. + // A real implementation would have more sophisticated retry/backoff logic. + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusNoContent { + log.Printf("transport: node %d returned status %s", m.To, resp.Status) + } +} + +func (t *transport) addPeer(id uint64, url string) { + t.mu.Lock() + defer t.mu.Unlock() + t.peers[id] = url +} + +func (t *transport) removePeer(id uint64) { + t.mu.Lock() + defer t.mu.Unlock() + delete(t.peers, id) +} diff --git a/contrib/raftsimple/raft.go b/contrib/raftsimple/raft.go new file mode 100644 index 000000000..dbbc7d72c --- /dev/null +++ b/contrib/raftsimple/raft.go @@ -0,0 +1,402 @@ +package main + +import ( + "context" + "errors" + "log" + "net" + "net/http" + "net/url" + "os" + "time" + + "go.etcd.io/raft/v3" + "go.etcd.io/raft/v3/raftpb" +) + +type commit struct { + data []string + applyDoneC chan<- struct{} +} + +// FSM is the interface that must be implemented by a finite state +// machine for it to be driven by raft. +type FSM interface { + TakeSnapshot() ([]byte, error) + RestoreSnapshot(snapshot []byte) error + ApplyCommits(commit *commit) error +} + +type raftNode struct { + proposeC <-chan string // proposed messages (k,v) + confChangeC <-chan raftpb.ConfChange // proposed cluster config changes + commitC chan *commit // entries committed to log (k,v) + errorC chan<- error // errors from raft session + + id uint64 + join bool + peers []raft.Peer + peerURLs map[uint64]string + fsm FSM + ss snapshotStorage + transport *transport + + confState raftpb.ConfState + snapshotIndex uint64 + appliedIndex uint64 + snapCount uint64 + + node raft.Node + raftStorage *raft.MemoryStorage + stopc chan struct{} // signals proposal channel closed + donec chan struct{} // signals serveChannels has exited + err error +} + +var DefaultSnapshotCount uint64 = 10000 + +func newRaftNode(id uint64, peerURLs map[uint64]string, join bool, fsm FSM, ss snapshotStorage, proposeC <-chan string, confChangeC <-chan raftpb.ConfChange) *raftNode { + commitC := make(chan *commit) + errorC := make(chan error) + + rpeers := make([]raft.Peer, 0, len(peerURLs)) + for pID := range peerURLs { + rpeers = append(rpeers, raft.Peer{ID: pID}) + } + + rc := &raftNode{ + proposeC: proposeC, + confChangeC: confChangeC, + commitC: commitC, + errorC: errorC, + id: id, + join: join, + peers: rpeers, + peerURLs: peerURLs, + fsm: fsm, + ss: ss, + snapCount: DefaultSnapshotCount, + stopc: make(chan struct{}), + donec: make(chan struct{}), + transport: newTransport(id, peerURLs), + } + + go rc.startRaft() + return rc +} + +func (rc *raftNode) startRaft() { + rc.raftStorage = raft.NewMemoryStorage() + + hasState := rc.replayWAL() + + c := &raft.Config{ + ID: rc.id, + ElectionTick: 10, + HeartbeatTick: 1, + Storage: rc.raftStorage, + MaxSizePerMsg: 4096, + MaxInflightMsgs: 256, + } + + if hasState || rc.join { + rc.node = raft.RestartNode(c) + } else { + rc.node = raft.StartNode(c, rc.peers) + } + + rc.transport.node = rc.node + + go rc.serveRaft() + go rc.serveChannels() +} + +func (rc *raftNode) serveRaft() { + u, err := url.Parse(rc.peerURLs[rc.id]) + if err != nil { + log.Fatalf("raftsimple: failed to parse peer URL: %v", err) + } + + ln, err := net.Listen("tcp", u.Host) + if err != nil { + log.Fatalf("raftsimple: failed to listen on %s: %v", u.Host, err) + } + + rc.transport.mu.Lock() + rc.transport.server = &http.Server{Handler: rc.transport.Handler()} + rc.transport.mu.Unlock() + + err = rc.transport.server.Serve(ln) + if err != nil && err != http.ErrServerClosed { + log.Fatalf("raftsimple: failed to serve raft HTTP: %v", err) + } +} + +func (rc *raftNode) replayWAL() bool { + hasState := false + + snapshot, err := rc.ss.loadSnap() + if err != nil && !errors.Is(err, os.ErrNotExist) { + log.Panic(err) + } + if snapshot != nil && !raft.IsEmptySnap(*snapshot) { + rc.raftStorage.ApplySnapshot(*snapshot) + hasState = true + } + + hs, err := rc.ss.loadHardState() + if err != nil && !errors.Is(err, os.ErrNotExist) { + log.Panic(err) + } + if hs != nil && !raft.IsEmptyHardState(*hs) { + rc.raftStorage.SetHardState(*hs) + hasState = true + } + + ents, err := rc.ss.loadEntries() + if err != nil && !errors.Is(err, os.ErrNotExist) { + log.Panic(err) + } + if len(ents) > 0 { + rc.raftStorage.Append(ents) + hasState = true + } + + return hasState +} + +func (rc *raftNode) publishSnapshot(snapshotToSave raftpb.Snapshot) { + if raft.IsEmptySnap(snapshotToSave) { + return + } + + log.Printf("publishing snapshot at index %d", rc.snapshotIndex) + if snapshotToSave.Metadata.Index <= rc.appliedIndex { + log.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 + } + + if applyDoneC != nil { + select { + case <-applyDoneC: + case <-rc.stopc: + return + } + } + + log.Printf("start snapshot [applied index: %d | last snapshot index: %d]", rc.appliedIndex, rc.snapshotIndex) + data, err := rc.fsm.TakeSnapshot() + if err != nil { + log.Panic(err) + } + snap, err := rc.raftStorage.CreateSnapshot(rc.appliedIndex, &rc.confState, data) + if err != nil { + panic(err) + } + if err := rc.ss.saveSnap(snap); 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 { + log.Printf("compacted log at index %d", compactIndex) + } + + rc.snapshotIndex = rc.appliedIndex +} + +func (rc *raftNode) entriesToApply(ents []raftpb.Entry) []raftpb.Entry { + if len(ents) == 0 { + return ents + } + firstIdx := ents[0].Index + if firstIdx > rc.appliedIndex+1 { + log.Fatalf("first index of committed entry[%d] should <= progress.appliedIndex[%d]+1", firstIdx, rc.appliedIndex) + } + if rc.appliedIndex-firstIdx+1 < uint64(len(ents)) { + return ents[rc.appliedIndex-firstIdx+1:] + } + return nil +} + +func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) { + if len(ents) == 0 { + return nil, true + } + + data := make([]string, 0, len(ents)) + for _, entry := range ents { + switch entry.Type { + case raftpb.EntryNormal: + if len(entry.Data) == 0 { + break + } + s := string(entry.Data) + data = append(data, s) + case raftpb.EntryConfChange: + var cc raftpb.ConfChange + cc.Unmarshal(entry.Data) + rc.confState = *rc.node.ApplyConfChange(cc) + + switch cc.Type { + case raftpb.ConfChangeAddNode: + if len(cc.Context) > 0 { + rc.transport.addPeer(cc.NodeID, string(cc.Context)) + } + case raftpb.ConfChangeRemoveNode: + if cc.NodeID == rc.id { + log.Printf("Node %d: I've been removed from the cluster! Shutting down.", rc.id) + return nil, false + } + rc.transport.removePeer(cc.NodeID) + } + } + } + var applyDoneC chan struct{} + + if len(data) > 0 { + applyDoneC = make(chan struct{}, 1) + select { + case rc.commitC <- &commit{data, applyDoneC}: + case <-rc.stopc: + return nil, false + } + } + rc.appliedIndex = ents[len(ents)-1].Index + + return applyDoneC, true +} + +func (rc *raftNode) serveChannels() { + snap, err := rc.raftStorage.Snapshot() + if err != nil { + panic(err) + } + rc.confState = snap.Metadata.ConfState + rc.snapshotIndex = snap.Metadata.Index + rc.appliedIndex = snap.Metadata.Index + + ticker := time.NewTicker(100 * time.Millisecond) + defer ticker.Stop() + + go func() { + confChangeCount := uint64(0) + for rc.proposeC != nil && rc.confChangeC != nil { + select { + case prop, ok := <-rc.proposeC: + if !ok { + rc.proposeC = nil + } else { + rc.node.Propose(context.TODO(), []byte(prop)) + } + case cc, ok := <-rc.confChangeC: + if !ok { + rc.confChangeC = nil + } else { + confChangeCount++ + cc.ID = confChangeCount + rc.node.ProposeConfChange(context.TODO(), cc) + } + } + } + close(rc.stopc) + }() + + for { + select { + case <-ticker.C: + rc.node.Tick() + case rd := <-rc.node.Ready(): + if !raft.IsEmptySnap(rd.Snapshot) { + rc.ss.saveSnap(rd.Snapshot) + } + if !raft.IsEmptyHardState(rd.HardState) { + rc.ss.saveHardState(rd.HardState) + } + if len(rd.Entries) > 0 { + rc.ss.saveEntries(rd.Entries) + } + + rc.transport.send(rd.Messages) + + if !raft.IsEmptySnap(rd.Snapshot) { + rc.raftStorage.ApplySnapshot(rd.Snapshot) + rc.publishSnapshot(rd.Snapshot) + } + rc.raftStorage.Append(rd.Entries) + + applyDoneC, ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)) + if !ok { + rc.stop() + return + } + + rc.maybeTriggerSnapshot(applyDoneC) + rc.node.Advance() + case <-rc.stopc: + rc.stop() + return + } + } +} + +func (rc *raftNode) loadAndApplySnapshot() error { + snapshot, err := rc.ss.loadSnap() + if err != nil { + return err + } + if snapshot != nil && !raft.IsEmptySnap(*snapshot) { + log.Printf("loading snapshot at index %d", snapshot.Metadata.Index) + if err := rc.fsm.RestoreSnapshot(snapshot.Data); err != nil { + return err + } + } + return nil +} + +func (rc *raftNode) processCommits() error { + if err := rc.loadAndApplySnapshot(); err != nil { + return err + } + for commit := range rc.commitC { + if commit == nil { + if err := rc.loadAndApplySnapshot(); err != nil { + return err + } + continue + } + if err := rc.fsm.ApplyCommits(commit); err != nil { + return err + } + } + <-rc.donec + return rc.err +} + +func (rc *raftNode) stop() { + log.Printf("node %d: Executing stop()\n", rc.id) + close(rc.commitC) + close(rc.errorC) + close(rc.donec) + rc.transport.Close() + rc.node.Stop() +} diff --git a/contrib/raftsimple/snapshot_test.go b/contrib/raftsimple/snapshot_test.go new file mode 100644 index 000000000..b82cdaf3f --- /dev/null +++ b/contrib/raftsimple/snapshot_test.go @@ -0,0 +1,95 @@ +package main + +import ( + "fmt" + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.etcd.io/raft/v3/raftpb" +) + +func findFollower(nodes map[uint64]*raftNode) *raftNode { + for range 50 { + for _, rn := range nodes { + if rn.node != nil && rn.node.Status().Lead != rn.id && rn.node.Status().Lead != 0 { + return rn + } + } + time.Sleep(100 * time.Millisecond) + } + return nil +} + +func TestSnapshot_Recovery(t *testing.T) { + oldSnapCount := DefaultSnapshotCount + DefaultSnapshotCount = 5 + defer func() { DefaultSnapshotCount = oldSnapCount }() + + tmpDir := t.TempDir() + defer os.RemoveAll(tmpDir) + + peers := []uint64{1, 2, 3} + peerURLs := make(map[uint64]string) + for _, id := range peers { + peerURLs[id] = fmt.Sprintf("http://127.0.0.1:%d", 20000+id) + } + + nodes := make(map[uint64]*raftNode) + kvstores := make(map[uint64]*kvstore) + + createNode := func(id uint64) { + snapdir := fmt.Sprintf("%s/node-%d", tmpDir, id) + ss, _ := newSnapshotStorage(snapdir) + proposeC := make(chan string) + confChangeC := make(chan raftpb.ConfChange) + kvs, fsm := newKVStore(proposeC) + rn := newRaftNode(id, peerURLs, false, fsm, ss, proposeC, confChangeC) + nodes[id] = rn + kvstores[id] = kvs + go rn.processCommits() + } + + for _, id := range peers { + createNode(id) + } + + defer func() { + for _, rn := range nodes { + rn.stop() + } + }() + + leader := findLeader(nodes) + require.NotNil(t, leader, "A leader should be elected") + + follower := findFollower(nodes) + require.NotNil(t, follower, "A follower should be found") + followerID := follower.id + + // Stop follower + nodes[followerID].stop() + + for i := 0; i < int(DefaultSnapshotCount)+2; i++ { + kvstores[leader.id].Propose(fmt.Sprintf("key%d", i), fmt.Sprintf("val%d", i)) + } + + require.Eventually(t, func() bool { + return nodes[leader.id].snapshotIndex > 0 + }, 10*time.Second, 100*time.Millisecond, "Leader should have taken a snapshot") + + // Restart follower + createNode(followerID) + + require.Eventually(t, func() bool { + v, ok := kvstores[followerID].Lookup("key0") + return ok && v == "val0" + }, 10*time.Second, 100*time.Millisecond, "Follower should catch up via snapshot") + + require.Eventually(t, func() bool { + lastIdx := int(DefaultSnapshotCount) + 1 + v, ok := kvstores[followerID].Lookup(fmt.Sprintf("key%d", lastIdx)) + return ok && v == fmt.Sprintf("val%d", lastIdx) + }, 5*time.Second, 100*time.Millisecond, "Follower should catch up to the latest entry") +} diff --git a/contrib/raftsimple/storage.go b/contrib/raftsimple/storage.go new file mode 100644 index 000000000..ef5b1e134 --- /dev/null +++ b/contrib/raftsimple/storage.go @@ -0,0 +1,164 @@ +package main + +import ( + "encoding/binary" + "fmt" + "io" + "os" + + "go.etcd.io/raft/v3/raftpb" +) + +type snapshotStorage interface { + saveSnap(snapshot raftpb.Snapshot) error + loadSnap() (*raftpb.Snapshot, error) + + saveHardState(st raftpb.HardState) error + loadHardState() (*raftpb.HardState, error) + + saveEntries(entries []raftpb.Entry) error + loadEntries() ([]raftpb.Entry, error) +} + +type snapStore struct { + dir string +} + +func newSnapshotStorage(dir string) (snapshotStorage, error) { + err := os.MkdirAll(dir, 0o750) + if err != nil { + return nil, err + } + ss := snapStore{dir: dir} + return &ss, nil +} + +func (ss *snapStore) saveSnap(snapshot raftpb.Snapshot) error { + snap, err := snapshot.Marshal() + if err != nil { + return err + } + return ss.saveToFile("snap", snap) +} + +func (ss *snapStore) loadSnap() (*raftpb.Snapshot, error) { + data, err := os.ReadFile(fmt.Sprintf("%s/snap", ss.dir)) + if err != nil { + if os.IsNotExist(err) { + return &raftpb.Snapshot{}, nil + } + return nil, err + } + + var newSnap raftpb.Snapshot + if err := newSnap.Unmarshal(data); err != nil { + return nil, err + } + return &newSnap, nil +} + +func (ss *snapStore) saveHardState(st raftpb.HardState) error { + data, err := st.Marshal() + if err != nil { + return err + } + return ss.saveToFile("hardstate", data) +} + +func (ss *snapStore) loadHardState() (*raftpb.HardState, error) { + data, err := os.ReadFile(fmt.Sprintf("%s/hardstate", ss.dir)) + if err != nil { + if os.IsNotExist(err) { + return &raftpb.HardState{}, nil + } + return nil, err + } + + var st raftpb.HardState + if err := st.Unmarshal(data); err != nil { + return nil, err + } + return &st, nil +} + +func (ss *snapStore) saveEntries(entries []raftpb.Entry) error { + if len(entries) == 0 { + return nil + } + + f, err := os.OpenFile(fmt.Sprintf("%s/wal", ss.dir), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if err != nil { + return err + } + defer f.Close() + + for _, ent := range entries { + data, err := ent.Marshal() + if err != nil { + return err + } + + lenBuf := make([]byte, 4) + binary.LittleEndian.PutUint32(lenBuf, uint32(len(data))) + if _, err := f.Write(lenBuf); err != nil { + return err + } + + if _, err := f.Write(data); err != nil { + return err + } + } + return f.Sync() +} + +func (ss *snapStore) loadEntries() ([]raftpb.Entry, error) { + walPath := fmt.Sprintf("%s/wal", ss.dir) + f, err := os.Open(walPath) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + defer f.Close() + + var entries []raftpb.Entry + lenBuf := make([]byte, 4) + + for { + _, err := io.ReadFull(f, lenBuf) + if err != nil { + if err == io.EOF { + break + } + if err == io.ErrUnexpectedEOF { + return entries, fmt.Errorf("corrupted wal: incomplete length prefix") + } + return entries, err + } + + entryLen := binary.LittleEndian.Uint32(lenBuf) + dataBuf := make([]byte, entryLen) + + _, err = io.ReadFull(f, dataBuf) + if err != nil { + return entries, fmt.Errorf("corrupted wal: unable to read full entry payload: %v", err) + } + + var ent raftpb.Entry + if err := ent.Unmarshal(dataBuf); err != nil { + return entries, fmt.Errorf("failed to unmarshal entry: %v", err) + } + + entries = append(entries, ent) + } + + return entries, nil +} + +func (ss *snapStore) saveToFile(name string, data []byte) error { + if err := os.WriteFile(fmt.Sprintf("%s/%s", ss.dir, name), data, 0o644); err != nil { + return err + } + return nil +}