Validate etcd linearizability

.github/workflows/linearizability.yaml

@@ -0,0 +1,17 @@
+name: Linearizability
+on: [push, pull_request]
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - uses: actions/setup-go@v2
+      with:
+        go-version: "1.19.1"
+    - run: |
+        mkdir -p /tmp/linearizability
+        EXPECT_DEBUG=true GO_TEST_FLAGS=-v RESULTS_DIR=/tmp/linearizability make test-linearizability
+    - uses: actions/upload-artifact@v2
+      if: always()
+      with:
+        path: /tmp/linearizability/*

Makefile

@@ -7,25 +7,31 @@ build:
 
 # Tests
 
+GO_TEST_FLAGS?=
+
 .PHONY: test
 test:
-	PASSES="unit integration release e2e" ./scripts/test.sh
+	PASSES="unit integration release e2e" ./scripts/test.sh $(GO_TEST_FLAGS)
 
 .PHONY: test-unit
 test-unit:
-	PASSES="unit" ./scripts/test.sh
+	PASSES="unit" ./scripts/test.sh $(GO_TEST_FLAGS)
 
 .PHONY: test-integration
 test-integration:
-	PASSES="integration" ./scripts/test.sh
+	PASSES="integration" ./scripts/test.sh $(GO_TEST_FLAGS)
 
 .PHONY: test-e2e
 test-e2e: build
-	PASSES="e2e" ./scripts/test.sh
+	PASSES="e2e" ./scripts/test.sh $(GO_TEST_FLAGS)
 
 .PHONY: test-e2e-release
 test-e2e-release: build
-	PASSES="release e2e" ./scripts/test.sh
+	PASSES="release e2e" ./scripts/test.sh $(GO_TEST_FLAGS)
+
+.PHONY: test-linearizability
+test-linearizability: build
+	PASSES="linearizability" ./scripts/test.sh $(GO_TEST_FLAGS)
 
 # Static analysis
 

bill-of-materials.json

@@ -8,6 +8,15 @@
 			}
 		]
 	},
+	{
+		"project": "github.com/anishathalye/porcupine",
+		"licenses": [
+			{
+				"type": "MIT License",
+				"confidence": 1
+			}
+		]
+	},
 	{
 		"project": "github.com/benbjohnson/clock",
 		"licenses": [

pkg/expect/expect.go

@@ -182,6 +182,11 @@ func (ep *ExpectProcess) Signal(sig os.Signal) error {
 	return ep.cmd.Process.Signal(sig)
 }
 
+func (ep *ExpectProcess) Wait() error {
+	_, err := ep.cmd.Process.Wait()
+	return err
+}
+
 // Close waits for the expect process to exit.
 // Close currently does not return error if process exited with !=0 status.
 // TODO: Close should expose underlying process failure by default.

scripts/test.sh

@@ -124,6 +124,11 @@ function e2e_pass {
   run_for_module "tests" go_test "./common/..." "keep_going" : --tags=e2e -timeout="${TIMEOUT:-30m}" "${RUN_ARG[@]}" "$@"
 }
 
+function linearizability_pass {
+  # e2e tests are running pre-build binary. Settings like --race,-cover,-cpu does not have any impact.
+  run_for_module "tests" go_test "./linearizability/..." "keep_going" : -timeout="${TIMEOUT:-30m}" "${RUN_ARG[@]}" "$@"
+}
+
 function integration_e2e_pass {
   run_pass "integration" "${@}"
   run_pass "e2e" "${@}"

tests/framework/e2e.go

@@ -109,6 +109,10 @@ func (c *e2eCluster) Client(cfg clientv3.AuthConfig) (Client, error) {
 	return e2eClient{etcdctl}, nil
 }
 
+func (c *e2eCluster) Endpoints() []string {
+	return c.EndpointsV3()
+}
+
 func (c *e2eCluster) Members() (ms []Member) {
 	for _, proc := range c.EtcdProcessCluster.Procs {
 		ms = append(ms, e2eMember{EtcdProcess: proc, Cfg: c.Cfg})

tests/framework/e2e/cluster_proxy.go

@@ -103,6 +103,14 @@ func (p *proxyEtcdProcess) Logs() LogsExpect {
 	return p.etcdProc.Logs()
 }
 
+func (p *proxyEtcdProcess) Kill() error {
+	return p.etcdProc.Kill()
+}
+
+func (p *proxyEtcdProcess) Wait() error {
+	return p.etcdProc.Wait()
+}
+
 type proxyProc struct {
 	lg       *zap.Logger
 	name     string

tests/framework/e2e/etcd_process.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"syscall"
 	"testing"
 	"time"
 
@@ -38,12 +39,14 @@ type EtcdProcess interface {
 	EndpointsV3() []string
 	EndpointsMetrics() []string
 
+	Wait() error
 	Start(ctx context.Context) error
 	Restart(ctx context.Context) error
 	Stop() error
 	Close() error
 	Config() *EtcdServerProcessConfig
 	Logs() LogsExpect
+	Kill() error
 }
 
 type LogsExpect interface {
@@ -173,6 +176,22 @@ func (ep *EtcdServerProcess) Logs() LogsExpect {
 	return ep.proc
 }
 
+func (ep *EtcdServerProcess) Kill() error {
+	ep.cfg.lg.Info("killing server...", zap.String("name", ep.cfg.Name))
+	return ep.proc.Signal(syscall.SIGKILL)
+}
+
+func (ep *EtcdServerProcess) Wait() error {
+	err := ep.proc.Wait()
+	if err != nil {
+		ep.cfg.lg.Error("failed to wait for server exit", zap.String("name", ep.cfg.Name))
+		return err
+	}
+	ep.cfg.lg.Info("server exited", zap.String("name", ep.cfg.Name))
+	ep.proc = nil
+	return nil
+}
+
 func AssertProcessLogs(t *testing.T, ep EtcdProcess, expectLog string) {
 	t.Helper()
 	var err error

tests/framework/interface.go

@@ -33,6 +33,7 @@ type Cluster interface {
 	Client(cfg clientv3.AuthConfig) (Client, error)
 	WaitLeader(t testing.TB) int
 	Close() error
+	Endpoints() []string
 }
 
 type Member interface {

tests/go.mod

@@ -15,6 +15,7 @@ replace (
 )
 
 require (
+	github.com/anishathalye/porcupine v0.1.2
 	github.com/coreos/go-semver v0.3.0
 	github.com/dustin/go-humanize v1.0.0
 	github.com/gogo/protobuf v1.3.2

tests/go.sum

@@ -43,6 +43,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
+github.com/anishathalye/porcupine v0.1.2 h1:eqWNeLcnTzXt6usipDJ4RFn6XOWqY5wEqBYVG3yFLSE=
+github.com/anishathalye/porcupine v0.1.2/go.mod h1:/X9OQYnVb7DzfKCQVO4tI1Aq+o56UJW+RvN/5U4EuZA=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=

tests/linearizability/client.go

@@ -0,0 +1,87 @@
+// Copyright 2022 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 linearizability
+
+import (
+	"context"
+	"time"
+
+	"github.com/anishathalye/porcupine"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.uber.org/zap"
+)
+
+type recordingClient struct {
+	client clientv3.Client
+	id     int
+
+	operations []porcupine.Operation
+}
+
+func NewClient(endpoints []string, id int) (*recordingClient, error) {
+	cc, err := clientv3.New(clientv3.Config{
+		Endpoints:            endpoints,
+		Logger:               zap.NewNop(),
+		DialKeepAliveTime:    1 * time.Millisecond,
+		DialKeepAliveTimeout: 5 * time.Millisecond,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &recordingClient{
+		client:     *cc,
+		id:         id,
+		operations: []porcupine.Operation{},
+	}, nil
+}
+
+func (c *recordingClient) Close() error {
+	return c.client.Close()
+}
+
+func (c *recordingClient) Get(ctx context.Context, key string) error {
+	callTime := time.Now()
+	resp, err := c.client.Get(ctx, key)
+	returnTime := time.Now()
+	if err != nil {
+		return err
+	}
+	var readData string
+	if len(resp.Kvs) == 1 {
+		readData = string(resp.Kvs[0].Value)
+	}
+	c.operations = append(c.operations, porcupine.Operation{
+		ClientId: c.id,
+		Input:    etcdRequest{op: Get, key: key},
+		Call:     callTime.UnixNano(),
+		Output:   etcdResponse{getData: readData},
+		Return:   returnTime.UnixNano(),
+	})
+	return nil
+}
+
+func (c *recordingClient) Put(ctx context.Context, key, value string) error {
+	callTime := time.Now()
+	_, err := c.client.Put(ctx, key, value)
+	returnTime := time.Now()
+	c.operations = append(c.operations, porcupine.Operation{
+		ClientId: c.id,
+		Input:    etcdRequest{op: Put, key: key, putData: value},
+		Call:     callTime.UnixNano(),
+		Output:   etcdResponse{err: err},
+		Return:   returnTime.UnixNano(),
+	})
+	return nil
+}

tests/linearizability/failpoints.go

@@ -0,0 +1,49 @@
+// Copyright 2022 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 linearizability
+
+import (
+	"context"
+	"math/rand"
+
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+)
+
+var (
+	KillFailpoint Failpoint = killFailpoint{}
+)
+
+type Failpoint interface {
+	Trigger(ctx context.Context, clus *e2e.EtcdProcessCluster) error
+}
+
+type killFailpoint struct{}
+
+func (f killFailpoint) Trigger(ctx context.Context, clus *e2e.EtcdProcessCluster) error {
+	member := clus.Procs[rand.Int()%len(clus.Procs)]
+	err := member.Kill()
+	if err != nil {
+		return err
+	}
+	err = member.Wait()
+	if err != nil {
+		return err
+	}
+	err = member.Start(ctx)
+	if err != nil {
+		return err
+	}
+	return nil
+}