Merge #147665 #147888 #148410

147665: e2e-tests: add cypress tests for scale cluster r=dhartunian a=dhartunian

This commit adds files to enable the use of Cypress to smoke test DB Console on a cluster to see if it meets basic performance requirements. This can be used on a scale cluster to make sure we can render all the standard pages in a reasonable amount of time.

The test will print out timing information for key elements in the logs. Additionally, screenshots are taken immediately after the timing snapshot to enable easy analysis of what's on the page at time of timer capture. You can then confirm that the page is "useful".

Resolves: #51018

Release note: None

147888: kvcoord: avoid unnecessary flushing in txnWriteBuffer r=miraradeva a=stevendanna

Previously, if the buffer was being flushed midway through a transaction, we were flushing the entire buffer to ensure that future savepoint rollbacks would result in the correct final value being committed.

This was pessimistic, since many transactions don't use explicit savepoints.

Here, we track whether an explicit savepoint has been created and only flush those values required to rollback the earliest active savepoint.

Epic: none
Release note: None

148410: authors: add Sachuman to authors r=Sachuman a=Sachuman

Epic: None
Release note: None

Co-authored-by: David Hartunian <[email protected]>
Co-authored-by: Steven Danna <[email protected]>
Co-authored-by: Sachin Jain <[email protected]>

Author: 4 people

AUTHORS

@@ -456,6 +456,7 @@ Ryan Kuo <[email protected]> taroface <[email protected]
 Ryan Min <[email protected]>
 Ryan Punt <[email protected]> rpunt <[email protected]>
 Ryan Zhao <[email protected]> ryanzhao <[email protected]>
+Sachin Jain <[email protected]>
 Sadaf Qureshi <[email protected]>
 Sai Ravula <[email protected]>
 Sajjad Rizvi <[email protected]>

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -217,6 +217,10 @@ type txnInterceptor interface {
 	// createSavepointLocked().
 	rollbackToSavepointLocked(context.Context, savepoint)
 
+	// releaseSavepointLocked is called when a savepoint is being
+	// released.
+	releaseSavepointLocked(context.Context, *savepoint)
+
 	// closeLocked closes the interceptor. It is called when the TxnCoordSender
 	// shuts down due to either a txn commit or a txn abort. The method will
 	// be called exactly once from cleanupTxnLocked.

pkg/kv/kvclient/kvcoord/txn_coord_sender_savepoints.go

@@ -187,7 +187,15 @@ func (tc *TxnCoordSender) ReleaseSavepoint(ctx context.Context, s kv.SavepointTo
 	}
 
 	sp := s.(*savepoint)
-	return tc.checkSavepointLocked(sp, "release")
+	if err := tc.checkSavepointLocked(sp, "release"); err != nil {
+		return err
+	}
+
+	for _, reqInt := range tc.interceptorStack {
+		reqInt.releaseSavepointLocked(ctx, sp)
+	}
+
+	return nil
 }
 
 // CanUseSavepoint is part of the kv.TxnSender interface.

pkg/kv/kvclient/kvcoord/txn_interceptor_committer.go

@@ -626,6 +626,9 @@ func (tc *txnCommitter) epochBumpedLocked() {}
 // createSavepointLocked is part of the txnInterceptor interface.
 func (*txnCommitter) createSavepointLocked(context.Context, *savepoint) {}
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (*txnCommitter) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (*txnCommitter) rollbackToSavepointLocked(context.Context, savepoint) {}
 

pkg/kv/kvclient/kvcoord/txn_interceptor_heartbeater.go

@@ -302,6 +302,9 @@ func (h *txnHeartbeater) epochBumpedLocked() {}
 // createSavepointLocked is part of the txnInterceptor interface.
 func (*txnHeartbeater) createSavepointLocked(context.Context, *savepoint) {}
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (*txnHeartbeater) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (*txnHeartbeater) rollbackToSavepointLocked(context.Context, savepoint) {}
 

pkg/kv/kvclient/kvcoord/txn_interceptor_metric_recorder.go

@@ -78,6 +78,9 @@ func (*txnMetricRecorder) epochBumpedLocked() {}
 // createSavepointLocked is part of the txnInterceptor interface.
 func (*txnMetricRecorder) createSavepointLocked(context.Context, *savepoint) {}
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (*txnMetricRecorder) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (*txnMetricRecorder) rollbackToSavepointLocked(context.Context, savepoint) {}
 

pkg/kv/kvclient/kvcoord/txn_interceptor_pipeliner.go

@@ -984,6 +984,9 @@ func (tp *txnPipeliner) epochBumpedLocked() {
 // createSavepointLocked is part of the txnInterceptor interface.
 func (tp *txnPipeliner) createSavepointLocked(context.Context, *savepoint) {}
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (tp *txnPipeliner) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (tp *txnPipeliner) rollbackToSavepointLocked(ctx context.Context, s savepoint) {
 	// Move all the writes in txnPipeliner that are not in the savepoint to the

pkg/kv/kvclient/kvcoord/txn_interceptor_seq_num_allocator.go

@@ -219,6 +219,9 @@ func (s *txnSeqNumAllocator) createSavepointLocked(ctx context.Context, sp *save
 	sp.seqNum = s.writeSeq
 }
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (*txnSeqNumAllocator) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (s *txnSeqNumAllocator) rollbackToSavepointLocked(context.Context, savepoint) {
 	// Nothing to restore. The seq nums keep increasing. The TxnCoordSender has

pkg/kv/kvclient/kvcoord/txn_interceptor_span_refresher.go

@@ -764,6 +764,9 @@ func (sr *txnSpanRefresher) createSavepointLocked(ctx context.Context, s *savepo
 	s.refreshInvalid = sr.refreshInvalid
 }
 
+// releaseSavepointLocked is part of the txnInterceptor interface.
+func (sr *txnSpanRefresher) releaseSavepointLocked(context.Context, *savepoint) {}
+
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (sr *txnSpanRefresher) rollbackToSavepointLocked(ctx context.Context, s savepoint) {
 	if !KeepRefreshSpansOnSavepointRollback.Get(&sr.st.SV) {

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -177,6 +177,12 @@ type txnWriteBuffer struct {
 	// sees the next BatchRequest.
 	flushOnNextBatch bool
 
+	// firstExplicitSavepointSeq tracks the lowest explicit savepoint that hasn't
+	// been released or rolled back. If this savepoint is non-zero, then a
+	// mid-transaction flush must flush all revisions required to roll back this
+	// (or a later) savepoint.
+	firstExplicitSavepointSeq enginepb.TxnSeq
+
 	buffer        btree
 	bufferIDAlloc uint64
 	bufferSize    int64
@@ -605,6 +611,9 @@ func (twb *txnWriteBuffer) importLeafFinalState(context.Context, *roachpb.LeafTx
 
 // epochBumpedLocked implements the txnInterceptor interface.
 func (twb *txnWriteBuffer) epochBumpedLocked() {
+	// Sequence numbers are reset on epoch bumps so any retained savepoint is
+	// wrong.
+	twb.firstExplicitSavepointSeq = 0
 	twb.resetBuffer()
 }
 
@@ -613,8 +622,26 @@ func (twb *txnWriteBuffer) resetBuffer() {
 	twb.bufferSize = 0
 }
 
+func (twb *txnWriteBuffer) hasActiveSavepoint() bool {
+	return twb.firstExplicitSavepointSeq != enginepb.TxnSeq(0)
+}
+
 // createSavepointLocked is part of the txnInterceptor interface.
-func (twb *txnWriteBuffer) createSavepointLocked(context.Context, *savepoint) {}
+func (twb *txnWriteBuffer) createSavepointLocked(ctx context.Context, sp *savepoint) {
+	assertTrue(twb.firstExplicitSavepointSeq <= sp.seqNum,
+		"sequence number in created savepoint lower than retained savepoint")
+
+	if twb.firstExplicitSavepointSeq == enginepb.TxnSeq(0) {
+		twb.firstExplicitSavepointSeq = sp.seqNum
+	}
+}
+
+// rollbackToSavepointLocked is part of the txnInterceptor interface.
+func (twb *txnWriteBuffer) releaseSavepointLocked(ctx context.Context, sp *savepoint) {
+	if twb.firstExplicitSavepointSeq == sp.seqNum {
+		twb.firstExplicitSavepointSeq = 0
+	}
+}
 
 // rollbackToSavepointLocked is part of the txnInterceptor interface.
 func (twb *txnWriteBuffer) rollbackToSavepointLocked(ctx context.Context, s savepoint) {
@@ -647,6 +674,9 @@ func (twb *txnWriteBuffer) rollbackToSavepointLocked(ctx context.Context, s save
 	for _, bw := range toDelete {
 		twb.removeFromBuffer(bw)
 	}
+	if twb.firstExplicitSavepointSeq == s.seqNum {
+		twb.firstExplicitSavepointSeq = 0
+	}
 }
 
 // closeLocked implements the txnInterceptor interface.
@@ -1370,6 +1400,8 @@ func (twb *txnWriteBuffer) flushBufferAndSendBatch(
 		twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Inc(1)
 	}
 
+	midTxnFlushWithExplicitSavepoint := !hasEndTxn && twb.hasActiveSavepoint()
+
 	// Flush all buffered writes by pre-pending them to the requests being sent
 	// in the batch.
 	//
@@ -1379,8 +1411,8 @@ func (twb *txnWriteBuffer) flushBufferAndSendBatch(
 	it := twb.buffer.MakeIter()
 	numRevisionsBuffered := 0
 	for it.First(); it.Valid(); it.Next() {
-		if !hasEndTxn {
-			revs := it.Cur().toAllRevisionRequests()
+		if midTxnFlushWithExplicitSavepoint {
+			revs := it.Cur().toAllRevisionRequests(twb.firstExplicitSavepointSeq)
 			numRevisionsBuffered += len(revs)
 			reqs = append(reqs, revs...)
 		} else {
@@ -1589,13 +1621,26 @@ func (bw *bufferedWrite) toRequest() kvpb.RequestUnion {
 }
 
 // toAllRevisionRequests returns requests for all revisions of the buffered
-// writes for the key. When the buffer is flushed before the end of a
-// transaction, all revisions must be written to storage to ensure that a future
-// savepoint rollback is properly handled.
-func (bw *bufferedWrite) toAllRevisionRequests() []kvpb.RequestUnion {
+// writes that need to be flushed given the minimum sequence number.
+//
+// When the buffer is flushed before the end of a transaction, previous
+// revisions must be written to storage to ensure that a future savepoint
+// rollback is properly handled.
+//
+// The given sequence number is the smallest sequence number associated with an
+// active savepoint.
+//
+// A write below the the minimum sequence number can be elided if there is a
+// subsequent write also below the minimum sequence number.
+func (bw *bufferedWrite) toAllRevisionRequests(minSeq enginepb.TxnSeq) []kvpb.RequestUnion {
 	rus := make([]kvpb.RequestUnion, 0, len(bw.vals))
-	for _, val := range bw.vals {
-		rus = append(rus, val.toRequestUnion(bw.key, bw.exclusionExpectedSinceTimestamp()))
+	maxIdx := len(bw.vals) - 1
+	for i, val := range bw.vals {
+		nextWriteLessThanSeq := (i+1 < maxIdx) && (bw.vals[i+1].seq < minSeq)
+		canElideRevision := val.seq < minSeq && nextWriteLessThanSeq
+		if !canElideRevision {
+			rus = append(rus, val.toRequestUnion(bw.key, bw.exclusionExpectedSinceTimestamp()))
+		}
 	}
 	return rus
 }