etcd-test-proxy: initial commit

Signed-off-by: Gyuho Lee <gyuhox@gmail.com>

tools/etcd-test-proxy/Procfile

@@ -0,0 +1,14 @@
+s1: bin/etcd --name s1 --data-dir /tmp/etcd-test-proxy-data.s1 --listen-client-urls http://127.0.0.1:1379 --advertise-client-urls http://127.0.0.1:13790 --listen-peer-urls http://127.0.0.1:1380 --initial-advertise-peer-urls http://127.0.0.1:13800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
+
+s1-client-proxy: bin/etcd-test-proxy --from localhost:13790 --to localhost:1379 --http-port 1378
+s1-peer-proxy: bin/etcd-test-proxy --from localhost:13800 --to localhost:1380 --http-port 1381
+
+s2: bin/etcd --name s2 --data-dir /tmp/etcd-test-proxy-data.s2 --listen-client-urls http://127.0.0.1:2379 --advertise-client-urls http://127.0.0.1:23790 --listen-peer-urls http://127.0.0.1:2380 --initial-advertise-peer-urls http://127.0.0.1:23800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
+
+s2-client-proxy: bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378
+s2-peer-proxy: bin/etcd-test-proxy --from localhost:23800 --to localhost:2380 --http-port 2381
+
+s3: bin/etcd --name s3 --data-dir /tmp/etcd-test-proxy-data.s3 --listen-client-urls http://127.0.0.1:3379 --advertise-client-urls http://127.0.0.1:33790 --listen-peer-urls http://127.0.0.1:3380 --initial-advertise-peer-urls http://127.0.0.1:33800 --initial-cluster-token tkn --initial-cluster 's1=http://127.0.0.1:13800,s2=http://127.0.0.1:23800,s3=http://127.0.0.1:33800' --initial-cluster-state new
+
+s3-client-proxy: bin/etcd-test-proxy --from localhost:33790 --to localhost:3379 --http-port 3378
+s3-client-proxy: bin/etcd-test-proxy --from localhost:33800 --to localhost:3380 --http-port 3381

tools/etcd-test-proxy/README.md

@@ -0,0 +1,184 @@
+#### etcd-test-proxy
+
+Proxy layer that simulates various network conditions.
+
+Test locally
+
+```bash
+$ ./build
+$ ./bin/etcd
+
+$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
+
+$ ./bin/etcd-test-proxy --help
+$ ./bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378 --verbose
+
+$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:2379 put foo bar
+$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
+```
+
+Proxy overhead per request is under 500μs
+
+```bash
+$ go build -v -o ./bin/benchmark ./cmd/tools/benchmark
+
+$ ./bin/benchmark \
+  --endpoints localhost:2379 \
+  --conns 5 \
+  --clients 15 \
+  put \
+  --key-size 48 \
+  --val-size 50000 \
+  --total 10000
+
+<<COMMENT
+Summary:
+  Total:	8.4611 secs.
+  Slowest:	0.1324 secs.
+  Fastest:	0.0011 secs.
+  Average:	0.0121 secs.
+  Stddev:	0.0125 secs.
+  Requests/sec:	1181.8758
+
+Response time histogram:
+  0.0011 [1]	|
+  0.0142 [7899]	|∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
+  0.0273 [1339]	|∎∎∎∎∎∎
+  0.0405 [543]	|∎∎
+  0.0536 [67]	|
+  0.0667 [49]	|
+  0.0798 [9]	|
+  0.0930 [15]	|
+  0.1061 [42]	|
+  0.1192 [21]	|
+  0.1324 [15]	|
+
+Latency distribution:
+  10% in 0.0049 secs.
+  25% in 0.0064 secs.
+  50% in 0.0085 secs.
+  75% in 0.0126 secs.
+  90% in 0.0243 secs.
+  95% in 0.0307 secs.
+  99% in 0.0686 secs.
+  99.9% in 0.1294 secs.
+COMMENT
+
+$ ./bin/benchmark \
+  --endpoints localhost:23790 \
+  --conns 5 \
+  --clients 15 \
+  put \
+  --key-size 48 \
+  --val-size 50000 \
+  --total 10000
+
+<<COMMENT
+Summary:
+  Total:	9.1128 secs.
+  Slowest:	0.1363 secs.
+  Fastest:	0.0015 secs.
+  Average:	0.0131 secs.
+  Stddev:	0.0113 secs.
+  Requests/sec:	1097.3613
+
+Response time histogram:
+  0.0015 [1]	|
+  0.0150 [7407]	|∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎∎
+  0.0285 [2017]	|∎∎∎∎∎∎∎∎∎∎
+  0.0419 [440]	|∎∎
+  0.0554 [30]	|
+  0.0689 [13]	|
+  0.0824 [12]	|
+  0.0959 [48]	|
+  0.1093 [2]	|
+  0.1228 [16]	|
+  0.1363 [14]	|
+
+Latency distribution:
+  10% in 0.0054 secs.
+  25% in 0.0071 secs.
+  50% in 0.0100 secs.
+  75% in 0.0153 secs.
+  90% in 0.0241 secs.
+  95% in 0.0297 secs.
+  99% in 0.0584 secs.
+  99.9% in 0.1312 secs.
+COMMENT
+```
+
+Delay client transmit
+
+```bash
+$ curl -L http://localhost:2378/delay-tx -X PUT \
+  -d "latency=5s&random-variable=100ms"
+# added send latency 5s±100ms (current latency 4.92143955s)
+
+$ curl -L http://localhost:2378/delay-tx
+# current send latency 4.92143955s
+
+$ ETCDCTL_API=3 ./bin/etcdctl \
+  --endpoints localhost:23790 \
+  --command-timeout=3s \
+  put foo bar
+# Error: context deadline exceeded
+
+$ curl -L http://localhost:2378/delay-tx -X DELETE
+# removed latency 4.92143955s
+
+$ curl -L http://localhost:2378/delay-tx
+# current send latency 0s
+
+$ ETCDCTL_API=3 ./bin/etcdctl \
+  --endpoints localhost:23790 \
+  --command-timeout=3s \
+  put foo bar
+# OK
+```
+
+Pause client transmit
+
+```bash
+$ curl -L http://localhost:2378/pause-tx -X PUT
+# paused forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
+
+$ ETCDCTL_API=3 ./bin/etcdctl \
+  --endpoints localhost:23790 \
+  put foo bar
+# Error: context deadline exceeded
+
+$ curl -L http://localhost:2378/pause-tx -X DELETE
+# unpaused forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
+```
+
+Drop client packets
+
+```bash
+$ curl -L http://localhost:2378/blackhole-tx -X PUT
+# blackholed; dropping packets [tcp://localhost:23790 -> tcp://localhost:2379]
+
+$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
+# Error: context deadline exceeded
+
+$ curl -L http://localhost:2378/blackhole-tx -X DELETE
+# unblackholed; restart forwarding [tcp://localhost:23790 -> tcp://localhost:2379]
+```
+
+Trigger leader election
+
+```bash
+$ ./build
+$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
+
+$ rm -rf /tmp/etcd-test-proxy-data.s*
+$ goreman -f ./tools/etcd-test-proxy/Procfile start
+
+$ ETCDCTL_API=3 ./bin/etcdctl \
+  --endpoints localhost:13790,localhost:23790,localhost:33790 \
+  member list
+
+# isolate s1 when s1 is the current leader
+$ curl -L http://localhost:1381/blackhole-tx -X PUT
+$ curl -L http://localhost:1381/blackhole-rx -X PUT
+# s1 becomes follower after election timeout
+```

tools/etcd-test-proxy/main.go

@@ -0,0 +1,216 @@
+// Copyright 2018 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 (
+	"context"
+	"flag"
+	"fmt"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/coreos/etcd/pkg/transport"
+
+	"google.golang.org/grpc/grpclog"
+)
+
+var from string
+var to string
+var httpPort int
+var verbose bool
+
+func main() {
+	// TODO: support TLS
+	flag.StringVar(&from, "from", "localhost:23790", "Address URL to proxy from.")
+	flag.StringVar(&to, "to", "localhost:2379", "Address URL to forward.")
+	flag.IntVar(&httpPort, "http-port", 2378, "Port to serve etcd-test-proxy API.")
+	flag.BoolVar(&verbose, "verbose", false, "'true' to run proxy in verbose mode.")
+
+	flag.Usage = func() {
+		fmt.Fprintf(os.Stderr, "Usage of %q:\n", os.Args[0])
+		fmt.Fprintln(os.Stderr, `
+etcd-test-proxy simulates various network conditions for etcd testing purposes.
+See README.md for more examples.
+
+Example:
+
+# build etcd
+$ ./build
+$ ./bin/etcd
+
+# build etcd-test-proxy
+$ make build-etcd-test-proxy -f ./hack/scripts-dev/Makefile
+
+# to test etcd with proxy layer
+$ ./bin/etcd-test-proxy --help
+$ ./bin/etcd-test-proxy --from localhost:23790 --to localhost:2379 --http-port 2378 --verbose
+
+$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:2379 put foo bar
+$ ETCDCTL_API=3 ./bin/etcdctl --endpoints localhost:23790 put foo bar
+`)
+		flag.PrintDefaults()
+	}
+
+	flag.Parse()
+
+	cfg := transport.ProxyConfig{
+		From: url.URL{Scheme: "tcp", Host: from},
+		To:   url.URL{Scheme: "tcp", Host: to},
+	}
+	if verbose {
+		cfg.Logger = grpclog.NewLoggerV2WithVerbosity(os.Stderr, os.Stderr, os.Stderr, 5)
+	}
+	p := transport.NewProxy(cfg)
+	<-p.Ready()
+	defer p.Close()
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
+		w.Write([]byte(fmt.Sprintf("proxying [%s -> %s]\n", p.From(), p.To())))
+	})
+	mux.HandleFunc("/delay-tx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodGet:
+			w.Write([]byte(fmt.Sprintf("current send latency %v\n", p.LatencyTx())))
+		case http.MethodPut, http.MethodPost:
+			if err := req.ParseForm(); err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong form %q\n", err.Error())))
+				return
+			}
+			lat, err := time.ParseDuration(req.PostForm.Get("latency"))
+			if err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong latency form %q\n", err.Error())))
+				return
+			}
+			rv, err := time.ParseDuration(req.PostForm.Get("random-variable"))
+			if err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong random-variable form %q\n", err.Error())))
+				return
+			}
+			p.DelayTx(lat, rv)
+			w.Write([]byte(fmt.Sprintf("added send latency %v±%v (current latency %v)\n", lat, rv, p.LatencyTx())))
+		case http.MethodDelete:
+			lat := p.LatencyTx()
+			p.UndelayTx()
+			w.Write([]byte(fmt.Sprintf("removed latency %v\n", lat)))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	mux.HandleFunc("/delay-rx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodGet:
+			w.Write([]byte(fmt.Sprintf("current receive latency %v\n", p.LatencyRx())))
+		case http.MethodPut, http.MethodPost:
+			if err := req.ParseForm(); err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong form %q\n", err.Error())))
+				return
+			}
+			lat, err := time.ParseDuration(req.PostForm.Get("latency"))
+			if err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong latency form %q\n", err.Error())))
+				return
+			}
+			rv, err := time.ParseDuration(req.PostForm.Get("random-variable"))
+			if err != nil {
+				w.Write([]byte(fmt.Sprintf("wrong random-variable form %q\n", err.Error())))
+				return
+			}
+			p.DelayRx(lat, rv)
+			w.Write([]byte(fmt.Sprintf("added receive latency %v±%v (current latency %v)\n", lat, rv, p.LatencyRx())))
+		case http.MethodDelete:
+			lat := p.LatencyRx()
+			p.UndelayRx()
+			w.Write([]byte(fmt.Sprintf("removed latency %v\n", lat)))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	mux.HandleFunc("/pause-tx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodPut, http.MethodPost:
+			p.PauseTx()
+			w.Write([]byte(fmt.Sprintf("paused forwarding [%s -> %s]\n", p.From(), p.To())))
+		case http.MethodDelete:
+			p.UnpauseTx()
+			w.Write([]byte(fmt.Sprintf("unpaused forwarding [%s -> %s]\n", p.From(), p.To())))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	mux.HandleFunc("/pause-rx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodPut, http.MethodPost:
+			p.PauseRx()
+			w.Write([]byte(fmt.Sprintf("paused forwarding [%s <- %s]\n", p.From(), p.To())))
+		case http.MethodDelete:
+			p.UnpauseRx()
+			w.Write([]byte(fmt.Sprintf("unpaused forwarding [%s <- %s]\n", p.From(), p.To())))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	mux.HandleFunc("/blackhole-tx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodPut, http.MethodPost:
+			p.BlackholeTx()
+			w.Write([]byte(fmt.Sprintf("blackholed; dropping packets [%s -> %s]\n", p.From(), p.To())))
+		case http.MethodDelete:
+			p.UnblackholeTx()
+			w.Write([]byte(fmt.Sprintf("unblackholed; restart forwarding [%s -> %s]\n", p.From(), p.To())))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	mux.HandleFunc("/blackhole-rx", func(w http.ResponseWriter, req *http.Request) {
+		switch req.Method {
+		case http.MethodPut, http.MethodPost:
+			p.BlackholeRx()
+			w.Write([]byte(fmt.Sprintf("blackholed; dropping packets [%s <- %s]\n", p.From(), p.To())))
+		case http.MethodDelete:
+			p.UnblackholeRx()
+			w.Write([]byte(fmt.Sprintf("unblackholed; restart forwarding [%s <- %s]\n", p.From(), p.To())))
+		default:
+			w.Write([]byte(fmt.Sprintf("unsupported method %q\n", req.Method)))
+		}
+	})
+	srv := &http.Server{
+		Addr:    fmt.Sprintf(":%d", httpPort),
+		Handler: mux,
+	}
+	defer srv.Close()
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+	defer signal.Stop(sig)
+
+	go func() {
+		s := <-sig
+		fmt.Printf("\n\nreceived signal %q, shutting down HTTP server\n\n", s)
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		err := srv.Shutdown(ctx)
+		cancel()
+		fmt.Printf("gracefully stopped HTTP server with %v\n\n", err)
+		os.Exit(0)
+	}()
+
+	fmt.Printf("\nserving HTTP server http://localhost:%d\n\n", httpPort)
+	err := srv.ListenAndServe()
+	fmt.Printf("HTTP server exit with error %v\n", err)
+}