tests/robustness: Extract validation to separate package

Signed-off-by: Marek Siarkowicz <serathius@users.noreply.github.com>

tests/robustness/linearizability_test.go

@@ -26,8 +26,8 @@ import (
 	"go.etcd.io/etcd/api/v3/version"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
-	"go.etcd.io/etcd/tests/v3/robustness/model"
 	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+	"go.etcd.io/etcd/tests/v3/robustness/validate"
 )
 
 func TestRobustness(t *testing.T) {
@@ -65,10 +65,7 @@ func TestRobustness(t *testing.T) {
 			name:      traffic.Name + "ClusterOfSize3",
 			failpoint: RandomFailpoint,
 			traffic:   traffic,
-			watch: watchConfig{
-				expectUniqueRevision: traffic.Traffic.ExpectUniqueRevision(),
-			},
-			cluster: *e2e.NewConfig(clusterOfSize3Options...),
+			cluster:   *e2e.NewConfig(clusterOfSize3Options...),
 		})
 	}
 	scenarios = append(scenarios, testScenario{
@@ -160,7 +157,8 @@ func testRobustness(ctx context.Context, t *testing.T, lg *zap.Logger, s testSce
 
 	watchProgressNotifyEnabled := r.clus.Cfg.WatchProcessNotifyInterval != 0
 	validateGotAtLeastOneProgressNotify(t, r.clientReports, s.watch.requestProgress || watchProgressNotifyEnabled)
-	r.visualizeHistory = validateCorrectness(t, lg, s.watch, r.clientReports)
+	validateConfig := validate.Config{ExpectRevisionUnique: s.traffic.Traffic.ExpectUniqueRevision()}
+	r.visualizeHistory = validate.ValidateAndReturnVisualize(t, lg, validateConfig, r.clientReports)
 
 	panicked = false
 }
@@ -213,9 +211,3 @@ func forcestopCluster(clus *e2e.EtcdProcessCluster) error {
 	}
 	return clus.ConcurrentStop()
 }
-
-func validateCorrectness(t *testing.T, lg *zap.Logger, cfg watchConfig, reports []traffic.ClientReport) (visualize func(basepath string)) {
-	validateWatchCorrectness(t, cfg, reports)
-	operations := operationsFromClientReports(reports)
-	return model.ValidateOperationHistoryAndReturnVisualize(t, lg, operations)
-}

tests/robustness/model/history.go

@@ -16,36 +16,15 @@ package model
 
 import (
 	"fmt"
-	"testing"
 	"time"
 
 	"github.com/anishathalye/porcupine"
-	"go.uber.org/zap"
-
 	"go.etcd.io/etcd/api/v3/etcdserverpb"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
 )
 
-// ValidateOperationHistoryAndReturnVisualize return visualize as porcupine.linearizationInfo used to generate visualization is private.
-func ValidateOperationHistoryAndReturnVisualize(t *testing.T, lg *zap.Logger, operations []porcupine.Operation) (visualize func(basepath string)) {
-	linearizable, info := porcupine.CheckOperationsVerbose(NonDeterministicModel, operations, 5*time.Minute)
-	if linearizable == porcupine.Illegal {
-		t.Error("Model is not linearizable")
-	}
-	if linearizable == porcupine.Unknown {
-		t.Error("Linearization timed out")
-	}
-	return func(path string) {
-		lg.Info("Saving visualization", zap.String("path", path))
-		err := porcupine.VisualizePath(NonDeterministicModel, info, path)
-		if err != nil {
-			t.Errorf("Failed to visualize, err: %v", err)
-		}
-	}
-}
-
 // AppendableHistory allows to collect history of sequential operations.
 //
 // Ensures that operation history is compatible with porcupine library, by preventing concurrent requests sharing the

tests/robustness/validate/operations.go

@@ -0,0 +1,26 @@
+package validate
+
+import (
+	"github.com/anishathalye/porcupine"
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.uber.org/zap"
+	"testing"
+	"time"
+)
+
+func validateOperationHistoryAndReturnVisualize(t *testing.T, lg *zap.Logger, operations []porcupine.Operation) (visualize func(basepath string)) {
+	linearizable, info := porcupine.CheckOperationsVerbose(model.NonDeterministicModel, operations, 5*time.Minute)
+	if linearizable == porcupine.Illegal {
+		t.Error("Model is not linearizable")
+	}
+	if linearizable == porcupine.Unknown {
+		t.Error("Linearization timed out")
+	}
+	return func(path string) {
+		lg.Info("Saving visualization", zap.String("path", path))
+		err := porcupine.VisualizePath(model.NonDeterministicModel, info, path)
+		if err != nil {
+			t.Errorf("Failed to visualize, err: %v", err)
+		}
+	}
+}

tests/robustness/validate/patch_history.go

@@ -0,0 +1,92 @@
+package validate
+
+import (
+	"github.com/anishathalye/porcupine"
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+)
+
+func patchOperationsWithWatchEvents(operations []porcupine.Operation, persisted map[model.EtcdOperation]traffic.TimedWatchEvent) []porcupine.Operation {
+	newOperations := make([]porcupine.Operation, 0, len(operations))
+	lastObservedOperation := lastOperationObservedInWatch(operations, persisted)
+
+	for _, op := range operations {
+		request := op.Input.(model.EtcdRequest)
+		resp := op.Output.(model.EtcdNonDeterministicResponse)
+		if resp.Err == nil || op.Call > lastObservedOperation.Call || request.Type != model.Txn {
+			// Cannot patch those requests.
+			newOperations = append(newOperations, op)
+			continue
+		}
+		event := matchWatchEvent(request.Txn, persisted)
+		if event != nil {
+			// Set revision and time based on watchEvent.
+			op.Return = event.Time.Nanoseconds()
+			op.Output = model.EtcdNonDeterministicResponse{
+				EtcdResponse:  model.EtcdResponse{Revision: event.Revision},
+				ResultUnknown: true,
+			}
+			newOperations = append(newOperations, op)
+			continue
+		}
+		if hasNonUniqueWriteOperation(request.Txn) && !hasUniqueWriteOperation(request.Txn) {
+			// Leave operation as it is as we cannot match non-unique operations to watch events.
+			newOperations = append(newOperations, op)
+			continue
+		}
+		// Remove non persisted operations
+	}
+	return newOperations
+}
+
+func lastOperationObservedInWatch(operations []porcupine.Operation, watchEvents map[model.EtcdOperation]traffic.TimedWatchEvent) porcupine.Operation {
+	var maxCallTime int64
+	var lastOperation porcupine.Operation
+	for _, op := range operations {
+		request := op.Input.(model.EtcdRequest)
+		if request.Type != model.Txn {
+			continue
+		}
+		event := matchWatchEvent(request.Txn, watchEvents)
+		if event != nil && op.Call > maxCallTime {
+			maxCallTime = op.Call
+			lastOperation = op
+		}
+	}
+	return lastOperation
+}
+
+func matchWatchEvent(request *model.TxnRequest, watchEvents map[model.EtcdOperation]traffic.TimedWatchEvent) *traffic.TimedWatchEvent {
+	for _, etcdOp := range append(request.OperationsOnSuccess, request.OperationsOnFailure...) {
+		if etcdOp.Type == model.Put {
+			// Remove LeaseID which is not exposed in watch.
+			event, ok := watchEvents[model.EtcdOperation{
+				Type:  etcdOp.Type,
+				Key:   etcdOp.Key,
+				Value: etcdOp.Value,
+			}]
+			if ok {
+				return &event
+			}
+		}
+	}
+	return nil
+}
+
+func hasNonUniqueWriteOperation(request *model.TxnRequest) bool {
+	for _, etcdOp := range request.OperationsOnSuccess {
+		if etcdOp.Type == model.Put || etcdOp.Type == model.Delete {
+			return true
+		}
+	}
+	return false
+}
+
+func hasUniqueWriteOperation(request *model.TxnRequest) bool {
+	for _, etcdOp := range request.OperationsOnSuccess {
+		if etcdOp.Type == model.Put {
+			return true
+		}
+	}
+	return false
+}

tests/robustness/validate/validate.go

@@ -0,0 +1,42 @@
+package validate
+
+import (
+	"github.com/anishathalye/porcupine"
+	"go.etcd.io/etcd/tests/v3/robustness/model"
+	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+	"go.uber.org/zap"
+	"testing"
+)
+
+// ValidateAndReturnVisualize return visualize as porcupine.linearizationInfo used to generate visualization is private.
+func ValidateAndReturnVisualize(t *testing.T, lg *zap.Logger, cfg Config, reports []traffic.ClientReport) (visualize func(basepath string)) {
+	validateWatch(t, cfg, reports)
+	allOperations := operations(reports)
+	persisted := persistedOperations(reports)
+	newOperations := patchOperationsWithWatchEvents(allOperations, persisted)
+	return validateOperationHistoryAndReturnVisualize(t, lg, newOperations)
+}
+
+func operations(reports []traffic.ClientReport) []porcupine.Operation {
+	var ops []porcupine.Operation
+	for _, r := range reports {
+		ops = append(ops, r.OperationHistory.Operations()...)
+	}
+	return ops
+}
+
+func persistedOperations(reports []traffic.ClientReport) map[model.EtcdOperation]traffic.TimedWatchEvent {
+	persisted := map[model.EtcdOperation]traffic.TimedWatchEvent{}
+	for _, r := range reports {
+		for _, resp := range r.Watch {
+			for _, event := range resp.Events {
+				persisted[event.Op] = traffic.TimedWatchEvent{Time: resp.Time, WatchEvent: event}
+			}
+		}
+	}
+	return persisted
+}
+
+type Config struct {
+	ExpectRevisionUnique bool
+}

tests/robustness/validate/watch.go

@@ -0,0 +1,144 @@
+package validate
+
+import (
+	"github.com/google/go-cmp/cmp"
+	"go.etcd.io/etcd/tests/v3/robustness/traffic"
+	"testing"
+)
+
+func validateWatch(t *testing.T, cfg Config, reports []traffic.ClientReport) {
+	// Validate etcd watch properties defined in https://etcd.io/docs/v3.6/learning/api_guarantees/#watch-apis
+	for _, r := range reports {
+		validateOrdered(t, r)
+		validateUnique(t, cfg.ExpectRevisionUnique, r)
+		validateAtomic(t, r)
+		// TODO: Validate Resumable
+		validateBookmarkable(t, r)
+	}
+	validateEventsMatch(t, reports)
+	// Expects that longest history encompasses all events.
+	// TODO: Use combined events from all histories instead of the longest history.
+	// TODO: Validate that each watch report is reliable, not only the longest one.
+	validateReliable(t, longestEventHistory(reports))
+}
+
+func validateBookmarkable(t *testing.T, report traffic.ClientReport) {
+	var lastProgressNotifyRevision int64 = 0
+	for _, resp := range report.Watch {
+		for _, event := range resp.Events {
+			if event.Revision <= lastProgressNotifyRevision {
+				t.Errorf("Broke watch guarantee: Renewable - watch can renewed using revision in last progress notification; Progress notification guarantees that previous events have been already delivered, eventRevision: %d, progressNotifyRevision: %d", event.Revision, lastProgressNotifyRevision)
+			}
+		}
+		if resp.IsProgressNotify {
+			lastProgressNotifyRevision = resp.Revision
+		}
+	}
+}
+
+func validateOrdered(t *testing.T, report traffic.ClientReport) {
+	var lastEventRevision int64 = 1
+	for _, resp := range report.Watch {
+		for _, event := range resp.Events {
+			if event.Revision < lastEventRevision {
+				t.Errorf("Broke watch guarantee: Ordered - events are ordered by revision; an event will never appear on a watch if it precedes an event in time that has already been posted, lastRevision: %d, currentRevision: %d, client: %d", lastEventRevision, event.Revision, report.ClientId)
+			}
+			lastEventRevision = event.Revision
+		}
+	}
+}
+
+func validateUnique(t *testing.T, expectUniqueRevision bool, report traffic.ClientReport) {
+	uniqueOperations := map[interface{}]struct{}{}
+
+	for _, resp := range report.Watch {
+		for _, event := range resp.Events {
+			var key interface{}
+			if expectUniqueRevision {
+				key = event.Revision
+			} else {
+				key = struct {
+					revision int64
+					key      string
+				}{event.Revision, event.Op.Key}
+			}
+
+			if _, found := uniqueOperations[key]; found {
+				t.Errorf("Broke watch guarantee: Unique - an event will never appear on a watch twice, key: %q, revision: %d, client: %d", event.Op.Key, event.Revision, report.ClientId)
+			}
+			uniqueOperations[key] = struct{}{}
+		}
+	}
+}
+
+func validateAtomic(t *testing.T, report traffic.ClientReport) {
+	var lastEventRevision int64 = 1
+	for _, resp := range report.Watch {
+		if len(resp.Events) > 0 {
+			if resp.Events[0].Revision == lastEventRevision {
+				t.Errorf("Broke watch guarantee: Atomic - a list of events is guaranteed to encompass complete revisions; updates in the same revision over multiple keys will not be split over several lists of events, previousListEventRevision: %d, currentListEventRevision: %d, client: %d", lastEventRevision, resp.Events[0].Revision, report.ClientId)
+			}
+			lastEventRevision = resp.Events[len(resp.Events)-1].Revision
+		}
+	}
+}
+
+func validateReliable(t *testing.T, events []traffic.TimedWatchEvent) {
+	var lastEventRevision int64 = 1
+	for _, event := range events {
+		if event.Revision > lastEventRevision && event.Revision != lastEventRevision+1 {
+			t.Errorf("Broke watch guarantee: Reliable - a sequence of events will never drop any subsequence of events; if there are events ordered in time as a < b < c, then if the watch receives events a and c, it is guaranteed to receive b, missing revisions from range: %d-%d", lastEventRevision, event.Revision)
+		}
+		lastEventRevision = event.Revision
+	}
+}
+
+func toWatchEvents(responses []traffic.WatchResponse) (events []traffic.TimedWatchEvent) {
+	for _, resp := range responses {
+		for _, event := range resp.Events {
+			events = append(events, traffic.TimedWatchEvent{
+				Time:       resp.Time,
+				WatchEvent: event,
+			})
+		}
+	}
+	return events
+}
+
+func validateEventsMatch(t *testing.T, reports []traffic.ClientReport) {
+	type revisionKey struct {
+		revision int64
+		key      string
+	}
+	type eventClientId struct {
+		traffic.WatchEvent
+		ClientId int
+	}
+	revisionKeyToEvent := map[revisionKey]eventClientId{}
+	for _, r := range reports {
+		for _, resp := range r.Watch {
+			for _, event := range resp.Events {
+				rk := revisionKey{key: event.Op.Key, revision: event.Revision}
+				if prev, found := revisionKeyToEvent[rk]; found {
+					if prev.WatchEvent != event {
+						t.Errorf("Events between clients %d and %d don't match, key: %q, revision: %d, diff: %s", prev.ClientId, r.ClientId, rk.key, rk.revision, cmp.Diff(prev, event))
+					}
+				}
+				revisionKeyToEvent[rk] = eventClientId{ClientId: r.ClientId, WatchEvent: event}
+			}
+		}
+	}
+}
+
+func longestEventHistory(report []traffic.ClientReport) []traffic.TimedWatchEvent {
+	longestIndex := 0
+	longestEventCount := 0
+	for i, r := range report {
+		rEventCount := r.WatchEventCount()
+		if rEventCount > longestEventCount {
+			longestIndex = i
+			longestEventCount = rEventCount
+		}
+	}
+	return toWatchEvents(report[longestIndex].Watch)
+}