tests/robustness: check for compaction before prevKV validation

We can check for the condition that Kubernetes checks for, i.e.
prevKV can be nil iff the event is not a create a event, only if
we know whether compaction has occured or not. If compaction has
occured, prevKV can be nil and that is completely valid.

This commit checks if compaction took place during the test run
by querying the /metrics endpoint. Based on if compaction occured,
we now check the Kubernetes condition in the prevKV robustness test.

Signed-off-by: Madhav Jivrajani <madhav.jiv@gmail.com>

tests/robustness/main_test.go

@@ -16,6 +16,7 @@ package robustness
 
 import (
 	"context"
+	"strings"
 	"testing"
 	"time"
 
@@ -23,6 +24,7 @@ import (
 	"go.uber.org/zap/zaptest"
 	"golang.org/x/sync/errgroup"
 
+	"go.etcd.io/etcd/pkg/v3/expect"
 	"go.etcd.io/etcd/tests/v3/framework"
 	"go.etcd.io/etcd/tests/v3/framework/e2e"
 	"go.etcd.io/etcd/tests/v3/robustness/failpoint"
@@ -89,11 +91,18 @@ func testRobustness(ctx context.Context, t *testing.T, lg *zap.Logger, s testSce
 		report.Report(t, panicked)
 	}()
 	report.Client = s.run(ctx, t, lg, report.Cluster)
+	compaction, err := checkForCompaction(report.Cluster)
+	if err != nil {
+		t.Fatalf("failed checking for compaction: %v", err)
+	}
 	forcestopCluster(report.Cluster)
 
 	watchProgressNotifyEnabled := report.Cluster.Cfg.ServerConfig.ExperimentalWatchProgressNotifyInterval != 0
 	validateGotAtLeastOneProgressNotify(t, report.Client, s.watch.requestProgress || watchProgressNotifyEnabled)
-	validateConfig := validate.Config{ExpectRevisionUnique: s.traffic.ExpectUniqueRevision()}
+	validateConfig := validate.Config{
+		ExpectRevisionUnique: s.traffic.ExpectUniqueRevision(),
+		AssumeCompaction:     compaction,
+	}
 	report.Visualize = validate.ValidateAndReturnVisualize(t, lg, validateConfig, report.Client)
 
 	panicked = false
@@ -158,3 +167,30 @@ func forcestopCluster(clus *e2e.EtcdProcessCluster) error {
 	}
 	return clus.ConcurrentStop()
 }
+
+func checkForCompaction(clus *e2e.EtcdProcessCluster) (bool, error) {
+	req := e2e.CURLReq{
+		Endpoint: "/metrics",
+		Expected: expect.ExpectedResponse{
+			// Filter out the etcd_debugging_mvcc_db_compaction_keys_total
+			// metric if it has a value greater than 0 from the response returned
+			// by the /metrics endpoint.
+			Value:         `etcd_debugging_mvcc_db_compaction_keys_total\s+[1-9][0-9]*`,
+			IsRegularExpr: true,
+		},
+	}
+
+	err := e2e.CURLGet(clus, req)
+	if err == nil {
+		return true, nil
+	}
+
+	// If no match was found, compaction did not take place.
+	noCompaction := strings.Contains(err.Error(), "match not found.") ||
+		strings.Contains(err.Error(), "context deadline exceeded")
+	if noCompaction {
+		return false, nil
+	}
+
+	return false, err
+}

tests/robustness/validate/validate.go

@@ -49,6 +49,7 @@ func ValidateAndReturnVisualize(t *testing.T, lg *zap.Logger, cfg Config, report
 
 type Config struct {
 	ExpectRevisionUnique bool
+	AssumeCompaction     bool
 }
 
 func mergeWatchEventHistory(reports []report.ClientReport) ([]model.PersistedEvent, error) {

tests/robustness/validate/watch.go

@@ -34,7 +34,7 @@ func validateWatch(t *testing.T, lg *zap.Logger, cfg Config, reports []report.Cl
 		if eventHistory != nil {
 			validateReliable(t, eventHistory, r)
 			validateResumable(t, eventHistory, r)
-			validatePrevKV(t, r, eventHistory)
+			validatePrevKV(t, r, eventHistory, cfg.AssumeCompaction)
 			validateCreateEvent(t, r, eventHistory)
 		}
 	}
@@ -148,7 +148,7 @@ func validateResumable(t *testing.T, events []model.PersistedEvent, report repor
 
 // validatePrevKV ensures that a watch response (if configured with WithPrevKV()) returns
 // the appropriate response.
-func validatePrevKV(t *testing.T, report report.ClientReport, history []model.PersistedEvent) {
+func validatePrevKV(t *testing.T, report report.ClientReport, history []model.PersistedEvent, compactionOccured bool) {
 	replay := model.NewReplay(history)
 	for _, op := range report.Watch {
 		if !op.Request.WithPrevKV {
@@ -161,16 +161,16 @@ func validatePrevKV(t *testing.T, report report.ClientReport, history []model.Pe
 				if err != nil {
 					t.Error(err)
 				}
-				// TODO(MadhavJivrajani): check if compaction has been run as part
-				// of failpoint injection. If compaction has run, prevKV can be nil
-				// even if it is not a create event.
-				//
+
 				// Considering that Kubernetes opens watches to etcd using WithPrevKV()
-				// option, ideally we would want to explicitly check the condition that
+				// option, we would want to explicitly check the condition that
 				// Kubernetes does while parsing events received from etcd:
 				// https://github.com/kubernetes/kubernetes/blob/a9e4f5b7862e84c4152eabe2e960f3f6fb9a4867/staging/src/k8s.io/apiserver/pkg/storage/etcd3/event.go#L59
-				// i.e. prevKV is nil iff the event is a create event, we cannot reliably
-				// check that without knowing if compaction has run.
+				// i.e. prevKV is nil iff the event is a create event, we can reliably
+				// check that only if compaction has not occured.
+				if !compactionOccured && event.PrevValue == nil && !event.IsCreate {
+					t.Errorf("PrevKV - without compaction, PrevValue cannot be nil if the event is not a create event, event: %+v", event)
+				}
 
 				// We allow PrevValue to be nil since in the face of compaction, etcd does not
 				// guarantee its presence.