tests: Add linearizability tests scenario for etcd-io#14370

scripts/build_lib.sh

@@ -19,6 +19,22 @@ toggle_failpoints() {
   mode="$1"
   if command -v gofail >/dev/null 2>&1; then
     run gofail "$mode" server/etcdserver/ server/storage/backend/
+    (
+      cd ./server
+      run go get go.etcd.io/gofail/runtime
+    )
+    (
+      cd ./etcdutl
+      run go get go.etcd.io/gofail/runtime
+    )
+    (
+      cd ./etcdctl
+      run go get go.etcd.io/gofail/runtime
+    )
+    (
+      cd ./tests
+      run go get go.etcd.io/gofail/runtime
+    )
   elif [[ "$mode" != "disable" ]]; then
     log_error "FAILPOINTS set but gofail not found"
     exit 1

tests/framework/e2e/cluster.go

@@ -182,6 +182,7 @@ type EtcdProcessClusterConfig struct {
 	CorruptCheckTime        time.Duration
 	CompactHashCheckEnabled bool
 	CompactHashCheckTime    time.Duration
+	GoFailEnabled           bool
 }
 
 // NewEtcdProcessCluster launches a new cluster from etcd processes, returning
@@ -388,12 +389,20 @@ func (cfg *EtcdProcessClusterConfig) EtcdServerProcessConfig(tb testing.TB, i in
 	if cfg.CompactHashCheckTime != 0 {
 		args = append(args, "--experimental-compact-hash-check-time", cfg.CompactHashCheckTime.String())
 	}
+	envVars := map[string]string{}
+	for key, value := range cfg.EnvVars {
+		envVars[key] = value
+	}
+	if cfg.GoFailEnabled {
+		port = (i+1)*10000 + 2381
+		envVars["GOFAIL_HTTP"] = fmt.Sprintf("127.0.0.1:%d", port)
+	}
 
 	return &EtcdServerProcessConfig{
 		lg:           cfg.Logger,
 		ExecPath:     cfg.ExecPath,
 		Args:         args,
-		EnvVars:      cfg.EnvVars,
+		EnvVars:      envVars,
 		TlsArgs:      cfg.TlsArgs(),
 		DataDirPath:  dataDirPath,
 		KeepDataDir:  cfg.KeepDataDir,

tests/linearizability/failpoints.go

@@ -15,14 +15,19 @@
 package linearizability
 
 import (
+	"bytes"
 	"context"
+	"fmt"
 	"math/rand"
+	"net/http"
+	"time"
 
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 )
 
 var (
-	KillFailpoint Failpoint = killFailpoint{}
+	KillFailpoint       Failpoint = killFailpoint{}
+	RaftBeforeSavePanic Failpoint = goFailpoint{"go.etcd.io/etcd/server/etcdserver/raftBeforeSave", "panic"}
 )
 
 type Failpoint interface {
@@ -43,3 +48,45 @@ func (f killFailpoint) Trigger(ctx context.Context, clus *e2e.EtcdProcessCluster
 	}
 	return nil
 }
+
+type goFailpoint struct {
+	failpoint string
+	payload   string
+}
+
+func (f goFailpoint) Trigger(ctx context.Context, clus *e2e.EtcdProcessCluster) error {
+	err := triggerGoFailpoint(f.failpoint, f.payload)
+	if err != nil {
+		return fmt.Errorf("failed to trigger failpoint %q, err: %v", f.failpoint, err)
+	}
+	err = clus.Procs[0].Wait()
+	if err != nil {
+		return err
+	}
+	err = clus.Procs[0].Start(ctx)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func triggerGoFailpoint(failpoint, payload string) error {
+	// TODO: Send failpoints to different members
+	r, err := http.NewRequest("PUT", "http://127.0.0.1:12381/"+failpoint, bytes.NewBuffer([]byte(payload)))
+	if err != nil {
+		return err
+	}
+	resp, err := httpClient.Do(r)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNoContent {
+		return fmt.Errorf("bad status code: %d", resp.StatusCode)
+	}
+	return nil
+}
+
+var httpClient = http.Client{
+	Timeout: 10 * time.Millisecond,
+}

tests/linearizability/linearizability_test.go

@@ -33,7 +33,7 @@ var (
 	// maximalQPS limits number of requests send to etcd to avoid linearizability analysis taking too long.
 	maximalQPS = 200.0
 	// failpointTriggersCount
-	failpointTriggersCount = 60
+	failpointTriggersCount = 10
 	// waitBetweenFailpointTriggers
 	waitBetweenFailpointTriggers = time.Second
 )
@@ -59,6 +59,14 @@ func TestLinearizability(t *testing.T) {
 				ClusterSize: 3,
 			},
 		},
+		{
+			name:      "Issue14370",
+			failpoint: RaftBeforeSavePanic,
+			config: e2e.EtcdProcessClusterConfig{
+				ClusterSize:   1,
+				GoFailEnabled: true,
+			},
+		},
 	}
 	for _, tc := range tcs {
 		t.Run(tc.name, func(t *testing.T) {