Merge #141821

141821: kvserver: add basic FlushLockTable request r=arulajmani a=stevendanna

This adds a new request FlushLockTableRequest. When issues, any
unreplicated locks in the request header's span will be written as
replicated locks to disk and removed from the in-memory lock table.s

The goal of this request is two fold:

1) By modelling this as a request, we can easily issue flush
   operations during KVNemesis at random. This will help us uncover
   problems with our current flush semantics.

2) We may use this flush request from a new queue that will
  periodically ensure that the size of the lock table is
  under control.

Epic: [CRDB-42764](https://cockroachlabs.atlassian.net/browse/CRDB-42764)

Release note: None

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

docs/generated/metrics/metrics.html

@@ -2980,6 +2980,18 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: distsender.rpc.flushlocktable.sent
+      exported_name: distsender_rpc_flushlocktable_sent
+      description: |-
+        Number of FlushLockTable requests processed.
+
+        This counts the requests in batches handed to DistSender, not the RPCs
+        sent to individual Ranges as a result.
+      y_axis_label: RPCs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: distsender.rpc.gc.sent
       exported_name: distsender_rpc_gc_sent
       description: |-
@@ -15676,6 +15688,14 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: rpc.method.flushlocktable.recv
+      exported_name: rpc_method_flushlocktable_recv
+      description: Number of FlushLockTable requests processed
+      y_axis_label: RPCs
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: rpc.method.gc.recv
       exported_name: rpc_method_gc_recv
       description: Number of GC requests processed

docs/generated/metrics/metrics.yaml

@@ -294,6 +294,7 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *kvpb.MigrateRequest:
 			case *kvpb.QueryResolvedTimestampRequest:
 			case *kvpb.BarrierRequest:
+			case *kvpb.FlushLockTableRequest:
 			case *kvpb.LinkExternalSSTableRequest:
 			case *kvpb.ExciseRequest:
 			default:
@@ -1141,6 +1142,27 @@ func (b *Batch) barrier(s, e interface{}, withLAI bool) {
 	b.initResult(1, 0, notRaw, nil)
 }
 
+func (b *Batch) flushLockTable(s, e interface{}) {
+	begin, err := marshalKey(s)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	end, err := marshalKey(e)
+	if err != nil {
+		b.initResult(0, 0, notRaw, err)
+		return
+	}
+	req := &kvpb.FlushLockTableRequest{
+		RequestHeader: kvpb.RequestHeader{
+			Key:    begin,
+			EndKey: end,
+		},
+	}
+	b.appendReqs(req)
+	b.initResult(1, 0, notRaw, nil)
+}
+
 func (b *Batch) bulkRequest(
 	numKeys int, requestFactory func() (req kvpb.RequestUnion, kvSize int),
 ) {

pkg/kv/batch.go

@@ -899,6 +899,23 @@ func (db *DB) Barrier(ctx context.Context, begin, end interface{}) (hlc.Timestam
 	return resp.Timestamp, nil
 }
 
+func (db *DB) FlushLockTable(ctx context.Context, begin, end interface{}) error {
+	b := &Batch{}
+	b.flushLockTable(begin, end)
+	if err := getOneErr(db.Run(ctx, b), b); err != nil {
+		return err
+	}
+	if l := len(b.response.Responses); l != 1 {
+		return errors.Errorf("got %d responses for FlushLockTable", l)
+	}
+	resp := b.response.Responses[0].GetFlushLockTable()
+	if resp == nil {
+		return errors.Errorf("unexpected response %T for FlushLockTable",
+			b.response.Responses[0].GetInner())
+	}
+	return nil
+}
+
 // BarrierWithLAI is like Barrier, but also returns the lease applied index and
 // range descriptor at which the barrier was applied. In this case, the barrier
 // can't span multiple ranges, otherwise a RangeKeyMismatchError is returned.

pkg/kv/db.go

@@ -143,6 +143,8 @@ func applyOp(ctx context.Context, env *Env, db *kv.DB, op *Operation) {
 			_, err = db.Barrier(ctx, o.Key, o.EndKey)
 		}
 		o.Result = resultInit(ctx, err)
+	case *FlushLockTableOperation:
+		o.Result = resultInit(ctx, db.FlushLockTable(ctx, o.Key, o.EndKey))
 	case *ClosureTxnOperation:
 		// Use a backoff loop to avoid thrashing on txn aborts. Don't wait between
 		// epochs of the same transaction to avoid waiting while holding locks.
@@ -450,6 +452,16 @@ func applyClientOp(
 			})
 		})
 		o.Result = resultInit(ctx, err)
+	case *FlushLockTableOperation:
+		_, _, err := dbRunWithResultAndTimestamp(ctx, db, func(b *kv.Batch) {
+			b.AddRawRequest(&kvpb.FlushLockTableRequest{
+				RequestHeader: kvpb.RequestHeader{
+					Key:    o.Key,
+					EndKey: o.EndKey,
+				},
+			})
+		})
+		o.Result = resultInit(ctx, err)
 	case *BatchOperation:
 		b := &kv.Batch{}
 		applyBatchOp(ctx, b, db.Run, o)
@@ -570,6 +582,8 @@ func applyBatchOp(
 			panic(errors.AssertionFailedf(`AddSSTable cannot be used in batches`))
 		case *BarrierOperation:
 			panic(errors.AssertionFailedf(`Barrier cannot be used in batches`))
+		case *FlushLockTableOperation:
+			panic(errors.AssertionFailedf(`FlushLockOperation cannot be used in batches`))
 		default:
 			panic(errors.AssertionFailedf(`unknown batch operation type: %T %v`, subO, subO))
 		}

pkg/kv/kvnemesis/applier.go

@@ -267,6 +267,10 @@ type ClientOperationConfig struct {
 	AddSSTable int
 	// Barrier is an operation that waits for in-flight writes to complete.
 	Barrier int
+
+	// FlushLockTable is an operation that moves unreplicated locks in the
+	// in-memory lock table into the
+	FlushLockTable int
 }
 
 // BatchOperationConfig configures the relative probability of generating a
@@ -407,6 +411,7 @@ func newAllOperationsConfig() GeneratorConfig {
 		DeleteRangeUsingTombstone:                          1,
 		AddSSTable:                                         1,
 		Barrier:                                            1,
+		FlushLockTable:                                     1,
 	}
 	batchOpConfig := BatchOperationConfig{
 		Batch: 4,
@@ -542,6 +547,11 @@ func NewDefaultConfig() GeneratorConfig {
 	config.Ops.ClosureTxn.CommitBatchOps.Barrier = 0
 	config.Ops.ClosureTxn.TxnClientOps.Barrier = 0
 	config.Ops.ClosureTxn.TxnBatchOps.Ops.Barrier = 0
+
+	config.Ops.Batch.Ops.FlushLockTable = 0
+	config.Ops.ClosureTxn.CommitBatchOps.FlushLockTable = 0
+	config.Ops.ClosureTxn.TxnClientOps.FlushLockTable = 0
+	config.Ops.ClosureTxn.TxnBatchOps.Ops.FlushLockTable = 0
 	return config
 }
 
@@ -839,6 +849,7 @@ func (g *generator) registerClientOps(allowed *[]opGen, c *ClientOperationConfig
 	addOpGen(allowed, randDelRangeUsingTombstone, c.DeleteRangeUsingTombstone)
 	addOpGen(allowed, randAddSSTable, c.AddSSTable)
 	addOpGen(allowed, randBarrier, c.Barrier)
+	addOpGen(allowed, randFlushLockTable, c.FlushLockTable)
 }
 
 func (g *generator) registerBatchOps(allowed *[]opGen, c *BatchOperationConfig) {
@@ -1144,6 +1155,19 @@ func randBarrier(g *generator, rng *rand.Rand) Operation {
 	return barrier(key, endKey, withLAI)
 }
 
+func randFlushLockTable(g *generator, rng *rand.Rand) Operation {
+	// FlushLockTable can't span multiple ranges. We want to test a combination of
+	// requests that span the entire range and those that span part of a range.
+	key, endKey := randRangeSpan(rng, g.currentSplits)
+
+	wholeRange := rng.Float64() < 0.5
+	if !wholeRange {
+		key = randKeyBetween(rng, key, endKey)
+	}
+
+	return flushLockTable(key, endKey)
+}
+
 func randScan(g *generator, rng *rand.Rand) Operation {
 	key, endKey := randSpan(rng)
 	return scan(key, endKey)
@@ -1986,6 +2010,13 @@ func barrier(key, endKey string, withLAI bool) Operation {
 	}}
 }
 
+func flushLockTable(key, endKey string) Operation {
+	return Operation{FlushLockTable: &FlushLockTableOperation{
+		Key:    []byte(key),
+		EndKey: []byte(endKey),
+	}}
+}
+
 func createSavepoint(id int) Operation {
 	return Operation{SavepointCreate: &SavepointCreateOperation{ID: int32(id)}}
 }

pkg/kv/kvnemesis/generator.go

@@ -248,6 +248,8 @@ func TestRandStep(t *testing.T) {
 				client.AddSSTable++
 			case *BarrierOperation:
 				client.Barrier++
+			case *FlushLockTableOperation:
+				client.FlushLockTable++
 			case *BatchOperation:
 				batch.Batch++
 				countClientOps(&batch.Ops, nil, o.Ops...)
@@ -284,7 +286,8 @@ func TestRandStep(t *testing.T) {
 			*DeleteRangeOperation,
 			*DeleteRangeUsingTombstoneOperation,
 			*AddSSTableOperation,
-			*BarrierOperation:
+			*BarrierOperation,
+			*FlushLockTableOperation:
 			countClientOps(&counts.DB, &counts.Batch, step.Op)
 		case *ClosureTxnOperation:
 			countClientOps(&counts.ClosureTxn.TxnClientOps, &counts.ClosureTxn.TxnBatchOps, o.Ops...)

pkg/kv/kvnemesis/generator_test.go

@@ -38,6 +38,8 @@ func (op Operation) Result() *Result {
 		return &o.Result
 	case *BarrierOperation:
 		return &o.Result
+	case *FlushLockTableOperation:
+		return &o.Result
 	case *SplitOperation:
 		return &o.Result
 	case *MergeOperation:
@@ -138,6 +140,8 @@ func (op Operation) format(w *strings.Builder, fctx formatCtx) {
 		o.format(w, fctx)
 	case *BarrierOperation:
 		o.format(w, fctx)
+	case *FlushLockTableOperation:
+		o.format(w, fctx)
 	case *SplitOperation:
 		o.format(w, fctx)
 	case *MergeOperation:
@@ -368,6 +372,11 @@ func (op BarrierOperation) format(w *strings.Builder, fctx formatCtx) {
 	op.Result.format(w)
 }
 
+func (op FlushLockTableOperation) format(w *strings.Builder, fctx formatCtx) {
+	fmt.Fprintf(w, `%s.FlushLockTable(ctx, %s, %s)`, fctx.receiver, fmtKey(op.Key), fmtKey(op.EndKey))
+	op.Result.format(w)
+}
+
 func (op SplitOperation) format(w *strings.Builder, fctx formatCtx) {
 	fmt.Fprintf(w, `%s.AdminSplit(ctx, %s, hlc.MaxTimestamp)`, fctx.receiver, fmtKey(op.Key))
 	op.Result.format(w)

pkg/kv/kvnemesis/operations.go

@@ -98,6 +98,12 @@ message BarrierOperation {
   Result result = 4 [(gogoproto.nullable) = false];
 }
 
+message FlushLockTableOperation {
+  bytes key = 1;
+  bytes end_key = 2;
+  Result result = 3 [(gogoproto.nullable) = false];
+}
+
 message SplitOperation {
   bytes key = 1;
   Result result = 2 [(gogoproto.nullable) = false];
@@ -183,6 +189,7 @@ message Operation {
   SavepointReleaseOperation savepoint_release = 21;
   SavepointRollbackOperation savepoint_rollback = 22;
   BarrierOperation barrier = 23;
+  FlushLockTableOperation flush_lock_table = 24;
 }
 
 enum ResultType {

pkg/kv/kvnemesis/operations.proto

@@ -735,6 +735,14 @@ func (v *validator) processOp(op Operation) {
 		}
 		// We don't yet actually check the barrier guarantees here, i.e. that all
 		// concurrent writes are applied by the time it completes. Maybe later.
+	case *FlushLockTableOperation:
+		execTimestampStrictlyOptional = true
+		if resultHasErrorType(t.Result, &kvpb.RangeKeyMismatchError{}) {
+			// FlushLockTableOperation may race with a split.
+		} else {
+			// Fail or retry on other errors, depending on type.
+			v.checkNonAmbError(op, t.Result, exceptUnhandledRetry)
+		}
 	case *ScanOperation:
 		if _, isErr := v.checkError(op, t.Result); isErr {
 			break