[tests] Watch events during linearizability test and compare history

tests/linearizability/linearizability_test.go

@@ -16,7 +16,6 @@ package linearizability
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
@@ -25,8 +24,10 @@ import (
 	"time"
 
 	"github.com/anishathalye/porcupine"
-	"go.etcd.io/etcd/tests/v3/framework/e2e"
+	"github.com/google/go-cmp/cmp"
 	"golang.org/x/time/rate"
+
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
 )
 
 const (
@@ -96,23 +97,31 @@ func testLinearizability(ctx context.Context, t *testing.T, config e2e.EtcdProce
 		t.Fatal(err)
 	}
 	defer clus.Close()
-	ctx, cancel := context.WithCancel(ctx)
+	trafficCtx, trafficCancel := context.WithCancel(ctx)
 	go func() {
-		defer cancel()
-		err := triggerFailpoints(ctx, t, clus, failpoint)
-		if err != nil {
-			t.Error(err)
-		}
+		defer trafficCancel()
+		triggerFailpoints(ctx, t, clus, failpoint)
+		// Wait second to collect traffic after triggering last failpoint.
+		time.Sleep(time.Second)
 	}()
-	operations := simulateTraffic(ctx, t, clus, traffic)
+	watchCtx, watchCancel := context.WithCancel(ctx)
+	var operations []porcupine.Operation
+	go func() {
+		defer watchCancel()
+		operations = simulateTraffic(trafficCtx, t, clus, traffic)
+		// Wait second to collect watch events after all traffic was sent.
+		time.Sleep(time.Second)
+	}()
+	events := watchClusterChanges(watchCtx, t, clus)
 	err = clus.Stop()
 	if err != nil {
 		t.Error(err)
 	}
+	validateEventsMatch(t, events)
 	checkOperationsAndPersistResults(t, operations, clus)
 }
 
-func triggerFailpoints(ctx context.Context, t *testing.T, clus *e2e.EtcdProcessCluster, config FailpointConfig) error {
+func triggerFailpoints(ctx context.Context, t *testing.T, clus *e2e.EtcdProcessCluster, config FailpointConfig) {
 	var err error
 	successes := 0
 	failures := 0
@@ -127,10 +136,8 @@ func triggerFailpoints(ctx context.Context, t *testing.T, clus *e2e.EtcdProcessC
 		successes++
 	}
 	if successes < config.count || failures >= config.retries {
-		return fmt.Errorf("failed to trigger failpoints enough times, err: %v", err)
+		t.Errorf("failed to trigger failpoints enough times, err: %v", err)
 	}
-	time.Sleep(config.waitBetweenTriggers)
-	return nil
 }
 
 type FailpointConfig struct {
@@ -187,6 +194,18 @@ type trafficConfig struct {
 	traffic     Traffic
 }
 
+func validateEventsMatch(t *testing.T, ops [][]watchEvent) {
+	for i := 1; i < len(ops); i++ {
+		shorterLength := len(ops[0])
+		if len(ops[i]) < shorterLength {
+			shorterLength = len(ops[i])
+		}
+		if diff := cmp.Diff(ops[0][:shorterLength], ops[i][:shorterLength]); diff != "" {
+			t.Errorf("Events in watches do not match, %s", diff)
+		}
+	}
+}
+
 func checkOperationsAndPersistResults(t *testing.T, operations []porcupine.Operation, clus *e2e.EtcdProcessCluster) {
 	path, err := testResultsDirectory(t)
 	if err != nil {

tests/linearizability/watch.go

@@ -0,0 +1,96 @@
+// 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"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/zap"
+
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/tests/v3/framework/e2e"
+)
+
+func watchClusterChanges(ctx context.Context, t *testing.T, clus *e2e.EtcdProcessCluster) [][]watchEvent {
+	mux := sync.Mutex{}
+	var wg sync.WaitGroup
+	memberEvents := make([][]watchEvent, len(clus.Procs))
+	for i, member := range clus.Procs {
+		c, err := clientv3.New(clientv3.Config{
+			Endpoints:            member.EndpointsV3(),
+			Logger:               zap.NewNop(),
+			DialKeepAliveTime:    1 * time.Millisecond,
+			DialKeepAliveTimeout: 5 * time.Millisecond,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		wg.Add(1)
+		go func(i int, c *clientv3.Client) {
+			defer wg.Done()
+			defer c.Close()
+			events := watchMemberEvents(ctx, c)
+			mux.Lock()
+			memberEvents[i] = events
+			mux.Unlock()
+		}(i, c)
+	}
+	wg.Wait()
+	return memberEvents
+}
+
+func watchMemberEvents(ctx context.Context, c *clientv3.Client) []watchEvent {
+	events := []watchEvent{}
+	var lastRevision int64 = 1
+	for {
+		select {
+		case <-ctx.Done():
+			return events
+		default:
+		}
+		for resp := range c.Watch(ctx, "", clientv3.WithPrefix(), clientv3.WithRev(lastRevision)) {
+			lastRevision = resp.Header.Revision
+			for _, event := range resp.Events {
+				switch event.Type {
+				case mvccpb.PUT:
+					events = append(events, watchEvent{
+						Op:       Put,
+						Key:      string(event.Kv.Key),
+						Value:    string(event.Kv.Value),
+						Revision: event.Kv.ModRevision,
+					})
+				case mvccpb.DELETE:
+					events = append(events, watchEvent{
+						Op:       Delete,
+						Key:      string(event.Kv.Key),
+						Revision: event.Kv.ModRevision,
+					})
+				}
+			}
+		}
+	}
+}
+
+type watchEvent struct {
+	Op       Operation
+	Key      string
+	Value    string
+	Revision int64
+}