Implement Compaction support in robustness test

Signed-off-by: Marek Siarkowicz <[email protected]>

Author: serathius

tests/robustness/client/client.go

@@ -193,9 +193,13 @@ func (c *RecordingClient) Defragment(ctx context.Context) (*clientv3.DefragmentR
 func (c *RecordingClient) Compact(ctx context.Context, rev int64) (*clientv3.CompactResponse, error) {
 	c.kvMux.Lock()
 	defer c.kvMux.Unlock()
+	callTime := time.Since(c.baseTime)
 	resp, err := c.client.Compact(ctx, rev)
+	returnTime := time.Since(c.baseTime)
+	c.kvOperations.AppendCompact(rev, callTime, returnTime, resp, err)
 	return resp, err
 }
+
 func (c *RecordingClient) MemberList(ctx context.Context, opts ...clientv3.OpOption) (*clientv3.MemberListResponse, error) {
 	c.kvMux.Lock()
 	defer c.kvMux.Unlock()

tests/robustness/failpoint/failpoint.go

@@ -39,17 +39,24 @@ var (
 		KillFailpoint, BeforeCommitPanic, AfterCommitPanic, RaftBeforeSavePanic, RaftAfterSavePanic,
 		DefragBeforeCopyPanic, DefragBeforeRenamePanic, BackendBeforePreCommitHookPanic, BackendAfterPreCommitHookPanic,
 		BackendBeforeStartDBTxnPanic, BackendAfterStartDBTxnPanic, BackendBeforeWritebackBufPanic,
-		BackendAfterWritebackBufPanic, CompactBeforeCommitScheduledCompactPanic, CompactAfterCommitScheduledCompactPanic,
-		CompactBeforeSetFinishedCompactPanic, CompactAfterSetFinishedCompactPanic, CompactBeforeCommitBatchPanic,
-		CompactAfterCommitBatchPanic, RaftBeforeLeaderSendPanic, BlackholePeerNetwork, DelayPeerNetwork,
+		BackendAfterWritebackBufPanic,
+		RaftBeforeLeaderSendPanic, BlackholePeerNetwork, DelayPeerNetwork,
 		RaftBeforeFollowerSendPanic, RaftBeforeApplySnapPanic, RaftAfterApplySnapPanic, RaftAfterWALReleasePanic,
 		RaftBeforeSaveSnapPanic, RaftAfterSaveSnapPanic, BlackholeUntilSnapshot,
 		BeforeApplyOneConfChangeSleep,
-		MemberReplace,
 		DropPeerNetwork,
 		RaftBeforeSaveSleep,
 		RaftAfterSaveSleep,
 		ApplyBeforeOpenSnapshot,
+		MemberReplace,
+
+		// Problem with compact
+		//CompactBeforeCommitScheduledCompactPanic,
+		//CompactBeforeCommitBatchPanic,
+		//CompactAfterCommitBatchPanic,
+		//CompactAfterCommitScheduledCompactPanic,
+		//CompactBeforeSetFinishedCompactPanic,
+		//CompactAfterSetFinishedCompactPanic,
 	}
 )
 

tests/robustness/failpoint/trigger.go

@@ -83,6 +83,11 @@ func (t triggerCompact) Trigger(ctx context.Context, _ *testing.T, member e2e.Et
 	return []report.ClientReport{cc.Report()}, nil
 }
 
-func (t triggerCompact) Available(e2e.EtcdProcessClusterConfig, e2e.EtcdProcess) bool {
+func (t triggerCompact) Available(config e2e.EtcdProcessClusterConfig, _ e2e.EtcdProcess) bool {
+	// For multiBatchCompaction we need to guarantee that there are enough revisions between two compaction requests.
+	// With addition of compaction requests to traffic this might be hard if experimental-compaction-batch-limit is too high.
+	if t.multiBatchCompaction {
+		return config.ServerConfig.ExperimentalCompactionBatchLimit <= 10
+	}
 	return true
 }

tests/robustness/model/describe.go

@@ -25,6 +25,9 @@ func describeEtcdResponse(request EtcdRequest, response MaybeEtcdResponse) strin
 	if response.Error != "" {
 		return fmt.Sprintf("err: %q", response.Error)
 	}
+	if response.ClientError != "" {
+		return fmt.Sprintf("err: %q", response.ClientError)
+	}
 	if response.PartialResponse {
 		return fmt.Sprintf("unknown, rev: %d", response.Revision)
 	}
@@ -38,6 +41,8 @@ func describeEtcdResponse(request EtcdRequest, response MaybeEtcdResponse) strin
 			return "ok"
 		}
 		return fmt.Sprintf("ok, rev: %d", response.Revision)
+	case Compact:
+		return "ok"
 	default:
 		return fmt.Sprintf("<! unknown request type: %q !>", request.Type)
 	}
@@ -67,6 +72,8 @@ func describeEtcdRequest(request EtcdRequest) string {
 		return fmt.Sprintf("leaseRevoke(%d)", request.LeaseRevoke.LeaseID)
 	case Defragment:
 		return fmt.Sprintf("defragment()")
+	case Compact:
+		return fmt.Sprintf("compact(%d)", request.Compact.Revision)
 	default:
 		return fmt.Sprintf("<! unknown request type: %q !>", request.Type)
 	}

tests/robustness/model/deterministic.go

@@ -23,6 +23,8 @@ import (
 	"sort"
 
 	"github.com/anishathalye/porcupine"
+
+	"go.etcd.io/etcd/server/v3/storage/mvcc"
 )
 
 // DeterministicModel assumes a deterministic execution of etcd requests. All
@@ -64,10 +66,11 @@ var DeterministicModel = porcupine.Model{
 }
 
 type EtcdState struct {
-	Revision  int64
-	KeyValues map[string]ValueRevision
-	KeyLeases map[string]int64
-	Leases    map[int64]EtcdLease
+	Revision        int64
+	CompactRevision int64
+	KeyValues       map[string]ValueRevision
+	KeyLeases       map[string]int64
+	Leases          map[int64]EtcdLease
 }
 
 func (s EtcdState) apply(request EtcdRequest, response EtcdResponse) (bool, EtcdState) {
@@ -77,10 +80,12 @@ func (s EtcdState) apply(request EtcdRequest, response EtcdResponse) (bool, Etcd
 
 func freshEtcdState() EtcdState {
 	return EtcdState{
-		Revision:  1,
-		KeyValues: map[string]ValueRevision{},
-		KeyLeases: map[string]int64{},
-		Leases:    map[int64]EtcdLease{},
+		Revision: 1,
+		// Start from CompactRevision equal -1 as etcd allows client to compact revision 0 for some reason.
+		CompactRevision: -1,
+		KeyValues:       map[string]ValueRevision{},
+		KeyLeases:       map[string]int64{},
+		Leases:          map[int64]EtcdLease{},
 	}
 }
 
@@ -100,6 +105,9 @@ func (s EtcdState) Step(request EtcdRequest) (EtcdState, MaybeEtcdResponse) {
 		if request.Range.Revision > s.Revision {
 			return s, MaybeEtcdResponse{Error: ErrEtcdFutureRev.Error()}
 		}
+		if request.Range.Revision < s.CompactRevision {
+			return s, MaybeEtcdResponse{EtcdResponse: EtcdResponse{ClientError: mvcc.ErrCompacted.Error()}}
+		}
 		return s, MaybeEtcdResponse{PartialResponse: true, EtcdResponse: EtcdResponse{Revision: s.Revision}}
 	case Txn:
 		failure := false
@@ -178,6 +186,14 @@ func (s EtcdState) Step(request EtcdRequest) (EtcdState, MaybeEtcdResponse) {
 		return s, MaybeEtcdResponse{EtcdResponse: EtcdResponse{Revision: s.Revision, LeaseRevoke: &LeaseRevokeResponse{}}}
 	case Defragment:
 		return s, MaybeEtcdResponse{EtcdResponse: EtcdResponse{Defragment: &DefragmentResponse{}, Revision: s.Revision}}
+	case Compact:
+		if request.Compact.Revision <= s.CompactRevision {
+			return s, MaybeEtcdResponse{EtcdResponse: EtcdResponse{ClientError: mvcc.ErrCompacted.Error()}}
+		}
+		s.CompactRevision = request.Compact.Revision
+		// Set fake revision as compaction returns non-linearizable revision.
+		// TODO: Model non-linearizable response revision in model.
+		return s, MaybeEtcdResponse{EtcdResponse: EtcdResponse{Compact: &CompactResponse{}, Revision: -1}}
 	default:
 		panic(fmt.Sprintf("Unknown request type: %v", request.Type))
 	}
@@ -237,6 +253,7 @@ const (
 	LeaseGrant  RequestType = "leaseGrant"
 	LeaseRevoke RequestType = "leaseRevoke"
 	Defragment  RequestType = "defragment"
+	Compact     RequestType = "compact"
 )
 
 type EtcdRequest struct {
@@ -246,6 +263,7 @@ type EtcdRequest struct {
 	Range       *RangeRequest
 	Txn         *TxnRequest
 	Defragment  *DefragmentRequest
+	Compact     *CompactRequest
 }
 
 func (r *EtcdRequest) IsRead() bool {
@@ -337,6 +355,8 @@ type EtcdResponse struct {
 	LeaseGrant  *LeaseGrantReponse
 	LeaseRevoke *LeaseRevokeResponse
 	Defragment  *DefragmentResponse
+	Compact     *CompactResponse
+	ClientError string
 	Revision    int64
 }
 
@@ -398,3 +418,10 @@ func ToValueOrHash(value string) ValueOrHash {
 	}
 	return v
 }
+
+type CompactResponse struct {
+}
+
+type CompactRequest struct {
+	Revision int64
+}

tests/robustness/model/history.go

@@ -16,13 +16,15 @@ package model
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/anishathalye/porcupine"
 
 	"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/server/v3/storage/mvcc"
 	"go.etcd.io/etcd/tests/v3/robustness/identity"
 )
 
@@ -259,6 +261,23 @@ func (h *AppendableHistory) AppendDefragment(start, end time.Duration, resp *cli
 	h.appendSuccessful(request, start, end, defragmentResponse(revision))
 }
 
+func (h *AppendableHistory) AppendCompact(rev int64, start, end time.Duration, resp *clientv3.CompactResponse, err error) {
+	request := compactRequest(rev)
+	if err != nil {
+		if strings.Contains(err.Error(), mvcc.ErrCompacted.Error()) {
+			h.appendSuccessful(request, start, end, MaybeEtcdResponse{
+				EtcdResponse: EtcdResponse{ClientError: mvcc.ErrCompacted.Error()},
+			})
+			return
+		}
+		h.appendFailed(request, start, end, err)
+		return
+	}
+	// Set fake revision as compaction returns non-linearizable revision.
+	// TODO: Model non-linearizable response revision in model.
+	h.appendSuccessful(request, start, end, compactResponse(-1))
+}
+
 func (h *AppendableHistory) appendFailed(request EtcdRequest, start, end time.Duration, err error) {
 	op := porcupine.Operation{
 		ClientId: h.streamID,
@@ -444,6 +463,14 @@ func defragmentResponse(revision int64) MaybeEtcdResponse {
 	return MaybeEtcdResponse{EtcdResponse: EtcdResponse{Defragment: &DefragmentResponse{}, Revision: revision}}
 }
 
+func compactRequest(rev int64) EtcdRequest {
+	return EtcdRequest{Type: Compact, Compact: &CompactRequest{Revision: rev}}
+}
+
+func compactResponse(revision int64) MaybeEtcdResponse {
+	return MaybeEtcdResponse{EtcdResponse: EtcdResponse{Compact: &CompactResponse{}, Revision: revision}}
+}
+
 type History struct {
 	operations []porcupine.Operation
 }

tests/robustness/options/server_config_options.go

@@ -26,6 +26,12 @@ func WithSnapshotCount(input ...uint64) e2e.EPClusterOption {
 	}
 }
 
+func WithCompactionBatchLimit(input ...int) e2e.EPClusterOption {
+	return func(c *e2e.EtcdProcessClusterConfig) {
+		c.ServerConfig.ExperimentalCompactionBatchLimit = input[internalRand.Intn(len(input))]
+	}
+}
+
 func WithSnapshotCatchUpEntries(input ...uint64) e2e.EPClusterOption {
 	return func(c *e2e.EtcdProcessClusterConfig) {
 		c.ServerConfig.SnapshotCatchUpEntries = input[internalRand.Intn(len(input))]

tests/robustness/report/wal.go

@@ -184,7 +184,11 @@ func parseEntryNormal(ent raftpb.Entry) (*model.EtcdRequest, error) {
 	case raftReq.ClusterVersionSet != nil:
 		return nil, nil
 	case raftReq.Compaction != nil:
-		return nil, nil
+		request := model.EtcdRequest{
+			Type:    model.Compact,
+			Compact: &model.CompactRequest{Revision: raftReq.Compaction.Revision},
+		}
+		return &request, nil
 	case raftReq.Txn != nil:
 		txn := model.TxnRequest{
 			Conditions:          []model.EtcdCondition{},

tests/robustness/scenarios.go

@@ -91,7 +91,8 @@ func exploratoryScenarios(_ *testing.T) []testScenario {
 		options.WithSnapshotCount(50, 100, 1000),
 		options.WithSubsetOptions(randomizableOptions...),
 		e2e.WithGoFailEnabled(true),
-		e2e.WithCompactionBatchLimit(100),
+		// Set low minimal compaction batch limit to allow for triggering multi batch compaction failpoints.
+		options.WithCompactionBatchLimit(10, 100, 1000),
 		e2e.WithWatchProcessNotifyInterval(100 * time.Millisecond),
 	}
 

tests/robustness/traffic/etcd.go

@@ -38,13 +38,14 @@ var (
 			{choice: List, weight: 15},
 			{choice: StaleGet, weight: 10},
 			{choice: StaleList, weight: 10},
-			{choice: Put, weight: 23},
-			{choice: LargePut, weight: 2},
 			{choice: Delete, weight: 5},
 			{choice: MultiOpTxn, weight: 5},
 			{choice: PutWithLease, weight: 5},
 			{choice: LeaseRevoke, weight: 5},
 			{choice: CompareAndSet, weight: 5},
+			{choice: Put, weight: 15},
+			{choice: LargePut, weight: 5},
+			{choice: Compact, weight: 5},
 		},
 	}
 	EtcdPut = etcdTraffic{
@@ -56,9 +57,10 @@ var (
 			{choice: List, weight: 15},
 			{choice: StaleGet, weight: 10},
 			{choice: StaleList, weight: 10},
-			{choice: Put, weight: 40},
 			{choice: MultiOpTxn, weight: 5},
 			{choice: LargePut, weight: 5},
+			{choice: Put, weight: 35},
+			{choice: Compact, weight: 5},
 		},
 	}
 )
@@ -89,6 +91,7 @@ const (
 	LeaseRevoke   etcdRequestType = "leaseRevoke"
 	CompareAndSet etcdRequestType = "compareAndSet"
 	Defragment    etcdRequestType = "defragment"
+	Compact       etcdRequestType = "compact"
 )
 
 func (t etcdTraffic) Name() string {
@@ -264,6 +267,12 @@ func (c etcdTrafficClient) Request(ctx context.Context, request etcdRequestType,
 		if resp != nil {
 			rev = resp.Header.Revision
 		}
+	case Compact:
+		var resp *clientv3.CompactResponse
+		resp, err = c.client.Compact(opCtx, lastRev)
+		if resp != nil {
+			rev = resp.Header.Revision
+		}
 	default:
 		panic("invalid choice")
 	}

tests/robustness/traffic/kubernetes.go

@@ -37,9 +37,10 @@ var (
 		resource:        "pods",
 		namespace:       "default",
 		writeChoices: []choiceWeight[KubernetesRequestType]{
-			{choice: KubernetesUpdate, weight: 90},
+			{choice: KubernetesUpdate, weight: 85},
 			{choice: KubernetesDelete, weight: 5},
 			{choice: KubernetesCreate, weight: 5},
+			{choice: KubernetesCompact, weight: 5},
 		},
 	}
 )
@@ -167,6 +168,8 @@ func (t kubernetesTraffic) Write(ctx context.Context, kc *kubernetesClient, ids
 				_, err = kc.OptimisticUpdate(writeCtx, key, fmt.Sprintf("%d", ids.NewRequestID()), rev)
 			case KubernetesCreate:
 				err = kc.OptimisticCreate(writeCtx, t.generateKey(), fmt.Sprintf("%d", ids.NewRequestID()))
+			case KubernetesCompact:
+				err = kc.Compact(writeCtx, rev)
 			default:
 				panic(fmt.Sprintf("invalid choice: %q", op))
 			}
@@ -209,9 +212,10 @@ func (t kubernetesTraffic) generateKey() string {
 type KubernetesRequestType string
 
 const (
-	KubernetesDelete KubernetesRequestType = "delete"
-	KubernetesUpdate KubernetesRequestType = "update"
-	KubernetesCreate KubernetesRequestType = "create"
+	KubernetesDelete  KubernetesRequestType = "delete"
+	KubernetesUpdate  KubernetesRequestType = "update"
+	KubernetesCreate  KubernetesRequestType = "create"
+	KubernetesCompact KubernetesRequestType = "compact"
 )
 
 type kubernetesClient struct {
@@ -250,6 +254,11 @@ func (k kubernetesClient) RequestProgress(ctx context.Context) error {
 	return k.client.RequestProgress(clientv3.WithRequireLeader(ctx))
 }
 
+func (k kubernetesClient) Compact(ctx context.Context, rev int64) error {
+	_, err := k.client.Compact(ctx, rev)
+	return err
+}
+
 // Kubernetes optimistically assumes that key didn't change since it was last observed, so it executes operations within a transaction conditioned on key not changing.
 // However, if the keys value changed it wants imminently to read it, thus the Get operation on failure.
 func (k kubernetesClient) optimisticOperationOrGet(ctx context.Context, key string, operation clientv3.Op, expectedRevision int64) (*mvccpb.KeyValue, error) {

tests/robustness/validate/patch_history.go

@@ -192,6 +192,7 @@ func persistedOperationsReturnTime(allOperations []porcupine.Operation, persiste
 				}
 			}
 		case model.LeaseGrant:
+		case model.Compact:
 		default:
 			panic(fmt.Sprintf("Unknown request type: %q", request.Type))
 		}
@@ -216,6 +217,7 @@ func operationReturnTime(operations []porcupine.Operation) map[model.EtcdOperati
 			}
 		case model.Range:
 		case model.LeaseGrant:
+		case model.Compact:
 		default:
 			panic(fmt.Sprintf("Unknown request type: %q", request.Type))
 		}