WIP:standalone raftexample

go.mod

@@ -6,13 +6,34 @@ require (
 	github.com/cockroachdb/datadriven v1.0.2
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
+	github.com/prometheus/client_golang v1.11.1
 	github.com/stretchr/testify v1.8.2
+	go.etcd.io/etcd/client/pkg/v3 v3.5.7
+	go.etcd.io/etcd/pkg/v3 v3.5.7
+	go.etcd.io/etcd/raft/v3 v3.5.7
+	go.etcd.io/etcd/server/v3 v3.5.7
+	go.uber.org/zap v1.24.0
 )
 
 require (
+	github.com/benbjohnson/clock v1.1.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
+	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/google/go-cmp v0.5.8 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.26.0 // indirect
+	github.com/prometheus/procfs v0.6.0 // indirect
+	github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.7 // indirect
+	go.uber.org/atomic v1.7.0 // indirect
+	go.uber.org/multierr v1.6.0 // indirect
+	golang.org/x/sys v0.3.0 // indirect
+	golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba // indirect
 	google.golang.org/protobuf v1.27.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

raftexample/Procfile

@@ -0,0 +1,4 @@
+# Use goreman to run `go install github.com/mattn/goreman@latest`
+raftexample1: ./raftexample --id 1 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 12380
+raftexample2: ./raftexample --id 2 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 22380
+raftexample3: ./raftexample --id 3 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 32380

raftexample/README.md

@@ -0,0 +1,125 @@
+# raftexample
+
+raftexample is an example usage of etcd's [raft library](https://github.com/etcd-io/raft). It provides a simple REST API for a key-value store cluster backed by the [Raft][raft] consensus algorithm.
+
+[raft]: http://raftconsensus.github.io/
+
+## Getting Started
+
+### Building raftexample
+
+Clone `etcd` to `<directory>/src/go.etcd.io/etcd`
+
+```sh
+export GOPATH=<directory>
+cd <directory>/src/go.etcd.io/etcd/contrib/raftexample
+go build -o raftexample
+```
+
+### Running single node raftexample
+
+First start a single-member cluster of raftexample:
+
+```sh
+raftexample --id 1 --cluster http://127.0.0.1:12379 --port 12380
+```
+
+Each raftexample process maintains a single raft instance and a key-value server.
+The process's list of comma separated peers (--cluster), its raft ID index into the peer list (--id), and http key-value server port (--port) are passed through the command line.
+
+Next, store a value ("hello") to a key ("my-key"):
+
+```
+curl -L http://127.0.0.1:12380/my-key -XPUT -d hello
+```
+
+Finally, retrieve the stored key:
+
+```
+curl -L http://127.0.0.1:12380/my-key
+```
+
+### Running a local cluster
+
+First install [goreman](https://github.com/mattn/goreman), which manages Procfile-based applications.
+
+The [Procfile script](./Procfile) will set up a local example cluster. Start it with:
+
+```sh
+goreman start
+```
+
+This will bring up three raftexample instances.
+
+Now it's possible to write a key-value pair to any member of the cluster and likewise retrieve it from any member.
+
+### Fault Tolerance
+
+To test cluster recovery, first start a cluster and write a value "foo":
+```sh
+goreman start
+curl -L http://127.0.0.1:12380/my-key -XPUT -d foo
+```
+
+Next, remove a node and replace the value with "bar" to check cluster availability:
+
+```sh
+goreman run stop raftexample2
+curl -L http://127.0.0.1:12380/my-key -XPUT -d bar
+curl -L http://127.0.0.1:32380/my-key
+```
+
+Finally, bring the node back up and verify it recovers with the updated value "bar":
+```sh
+goreman run start raftexample2
+curl -L http://127.0.0.1:22380/my-key
+```
+
+### Dynamic cluster reconfiguration
+
+Nodes can be added to or removed from a running cluster using requests to the REST API.
+
+For example, suppose we have a 3-node cluster that was started with the commands:
+```sh
+raftexample --id 1 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 12380
+raftexample --id 2 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 22380
+raftexample --id 3 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379 --port 32380
+```
+
+A fourth node with ID 4 can be added by issuing a POST:
+```sh
+curl -L http://127.0.0.1:12380/4 -XPOST -d http://127.0.0.1:42379
+```
+
+Then the new node can be started as the others were, using the --join option:
+```sh
+raftexample --id 4 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379,http://127.0.0.1:42379 --port 42380 --join
+```
+
+The new node should join the cluster and be able to service key/value requests.
+
+We can remove a node using a DELETE request:
+```sh
+curl -L http://127.0.0.1:12380/3 -XDELETE
+```
+
+Node 3 should shut itself down once the cluster has processed this request.
+
+## Design
+
+The raftexample consists of three components: a raft-backed key-value store, a REST API server, and a raft consensus server based on etcd's raft implementation.
+
+The raft-backed key-value store is a key-value map that holds all committed key-values.
+The store bridges communication between the raft server and the REST server.
+Key-value updates are issued through the store to the raft server.
+The store updates its map once raft reports the updates are committed.
+
+The REST server exposes the current raft consensus by accessing the raft-backed key-value store.
+A GET command looks up a key in the store and returns the value, if any.
+A key-value PUT command issues an update proposal to the store.
+
+The raft server participates in consensus with its cluster peers.
+When the REST server submits a proposal, the raft server transmits the proposal to its peers.
+When raft reaches a consensus, the server publishes all committed updates over a commit channel.
+For raftexample, this commit channel is consumed by the key-value store.
+

raftexample/doc.go

@@ -0,0 +1,16 @@
+// 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.
+
+// raftexample is a simple KV store using the raft and rafthttp libraries.
+package main

raftexample/httpapi.go

@@ -0,0 +1,122 @@
+// Copyright 2015 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.
+
+package main
+
+import (
+	"io"
+	"log"
+	"net/http"
+	"strconv"
+
+	"go.etcd.io/raft/v3/raftpb"
+)
+
+// Handler for a http based key-value store backed by raft
+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 on PUT", http.StatusBadRequest)
+			return
+		}
+
+		h.store.Propose(key, string(v))
+
+		// Optimistic-- no waiting for ack from raft. Value is not yet
+		// committed so a subsequent GET on the key may return old value
+		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:
+		url, err := io.ReadAll(r.Body)
+		if err != nil {
+			log.Printf("Failed to read on POST (%v)\n", err)
+			http.Error(w, "Failed on POST", http.StatusBadRequest)
+			return
+		}
+
+		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
+		}
+
+		cc := raftpb.ConfChange{
+			Type:    raftpb.ConfChangeAddNode,
+			NodeID:  nodeID,
+			Context: url,
+		}
+		h.confChangeC <- cc
+		// As above, optimistic that raft will apply the conf change
+		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
+
+		// As above, optimistic that raft will apply the conf change
+		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)
+	}
+}
+
+// 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) {
+	srv := http.Server{
+		Addr: ":" + strconv.Itoa(port),
+		Handler: &httpKVAPI{
+			store:       kv,
+			confChangeC: confChangeC,
+		},
+	}
+	go func() {
+		if err := srv.ListenAndServe(); err != nil {
+			log.Fatal(err)
+		}
+	}()
+
+	// exit when raft goes down
+	if err, ok := <-errorC; ok {
+		log.Fatal(err)
+	}
+}