kvclient: add some metrics for buffered writes

Gets the ball rolling on some metrics.

Epic: none

Release note: None

Author: arulajmani

docs/generated/metrics/metrics.html

@@ -1949,8 +1949,12 @@
 <tr><td>APPLICATION</td><td>txn.restarts.txnpush</td><td>Number of restarts due to a transaction push failure</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>txn.restarts.unknown</td><td>Number of restarts due to a unknown reasons</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>txn.restarts.writetooold</td><td>Number of restarts due to a concurrent writer committing first</td><td>Restarted Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>APPLICATION</td><td>txn.rollbacks.async.failed</td><td>Number of KV transaction that failed to send abort asynchronously which is not always retried</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
-<tr><td>APPLICATION</td><td>txn.rollbacks.failed</td><td>Number of KV transaction that failed to send final abort</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.rollbacks.async.failed</td><td>Number of KV transactions that failed to send abort asynchronously which is not always retried</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.rollbacks.failed</td><td>Number of KV transactions that failed to send final abort</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.batches.fully_handled</td><td>Number of KV batches that were fully handled by the write buffer (not sent to KV)</td><td>KV Batches</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.disabled_after_buffering</td><td>Number of KV transactions that disabled write buffering after buffering some writes but before an EndTxn request</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.memory_limit_exceeded</td><td>Number of KV transactions that exceeded the write buffering memory limit</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>txn.write_buffering.num_enabled</td><td>Number of KV transactions that enabled buffered writes</td><td>KV Transactions</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>SERVER</td><td>build.timestamp</td><td>Build information</td><td>Build Time</td><td>GAUGE</td><td>TIMESTAMP_SEC</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>SERVER</td><td>go.scheduler_latency</td><td>Go scheduling latency</td><td>Nanoseconds</td><td>HISTOGRAM</td><td>NANOSECONDS</td><td>AVG</td><td>NONE</td></tr>
 <tr><td>SERVER</td><td>log.buffered.messages.dropped</td><td>Count of log messages that are dropped by buffered log sinks. When CRDB attempts to buffer a log message in a buffered log sink whose buffer is already full, it drops the oldest buffered messages to make space for the new message</td><td>Messages</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>

pkg/kv/kvclient/kvcoord/txn_coord_sender.go

@@ -321,7 +321,10 @@ func (tc *TxnCoordSender) initCommonInterceptors(
 	if ds, ok := tcf.wrapped.(*DistSender); ok {
 		riGen.ds = ds
 	}
-	tc.interceptorAlloc.txnWriteBuffer.st = tcf.st
+	tc.interceptorAlloc.txnWriteBuffer = txnWriteBuffer{
+		st:         tcf.st,
+		txnMetrics: &tc.metrics,
+	}
 	tc.interceptorAlloc.txnPipeliner = txnPipeliner{
 		st:                       tcf.st,
 		riGen:                    riGen,

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -176,7 +176,8 @@ type txnWriteBuffer struct {
 
 	bufferSeek bufferedWrite // re-use while seeking
 
-	wrapped lockedSender
+	wrapped    lockedSender
+	txnMetrics *TxnMetrics
 
 	// testingOverrideCPutEvalFn is used to mock the evaluation function for
 	// conditional puts. Intended only for tests.
@@ -189,6 +190,9 @@ func (twb *txnWriteBuffer) setEnabled(enabled bool) {
 		// to ensure to flush the buffer.
 		twb.flushOnNextBatch = true
 	}
+	if enabled {
+		twb.txnMetrics.TxnWriteBufferEnabled.Inc(1)
+	}
 	twb.enabled = enabled
 }
 
@@ -249,7 +253,7 @@ func (twb *txnWriteBuffer) SendLocked(
 	// NB: if bufferedWritesMaxBufferSize is set to 0 then we effectively disable
 	// any buffer limiting.
 	if maxSize != 0 && bufSize > maxSize {
-		// TODO(arul): add some metrics for this case.
+		twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Inc(1)
 		log.VEventf(ctx, 2, "flushing buffer because buffer size (%s) exceeds max size (%s)",
 			humanizeutil.IBytes(bufSize),
 			humanizeutil.IBytes(maxSize))
@@ -279,6 +283,7 @@ func (twb *txnWriteBuffer) SendLocked(
 				return nil, pErr
 			}
 		}
+		twb.txnMetrics.TxnWriteBufferFullyHandledBatches.Inc(1)
 		return br, nil
 	}
 
@@ -1258,6 +1263,13 @@ func (twb *txnWriteBuffer) flushBufferAndSendBatch(
 		return twb.wrapped.SendLocked(ctx, ba) // nothing to flush
 	}
 
+	if _, ok := ba.GetArg(kvpb.EndTxn); !ok {
+		// We're flushing the buffer even though the batch doesn't contain an EndTxn
+		// request. That means we buffered some writes and decided to disable write
+		// buffering mid-way through the transaction, thus necessitating this flush.
+		twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Inc(1)
+	}
+
 	// Flush all buffered writes by pre-pending them to the requests being sent
 	// in the batch.
 	reqs := make([]kvpb.RequestUnion, 0, numBuffered+len(ba.Requests))

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
@@ -25,11 +26,13 @@ import (
 )
 
 func makeMockTxnWriteBuffer(st *cluster.Settings) (txnWriteBuffer, *mockLockedSender) {
+	metrics := MakeTxnMetrics(time.Hour)
 	mockSender := &mockLockedSender{}
 	return txnWriteBuffer{
-		enabled: true,
-		wrapped: mockSender,
-		st:      st,
+		enabled:    true,
+		wrapped:    mockSender,
+		txnMetrics: &metrics,
+		st:         st,
 	}, mockSender
 }
 
@@ -109,6 +112,7 @@ func TestTxnWriteBufferBuffersBlindWrites(t *testing.T) {
 	require.Equal(t, br.Responses[0].GetInner(), &kvpb.PutResponse{})
 	require.Equal(t, br.Responses[1].GetInner(), &kvpb.PutResponse{})
 	require.Equal(t, br.Responses[2].GetInner(), &kvpb.DeleteResponse{})
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferFullyHandledBatches.Count())
 
 	// Verify the writes were buffered correctly.
 	expBufferedWrites := []bufferedWrite{
@@ -1450,6 +1454,8 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
 	// Even though we flushed the Put, it shouldn't make it back to the response.
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.DeleteResponse{}, br.Responses[0].GetInner())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Count())
 
 	// Ensure the buffer is empty at this point.
 	require.Equal(t, 0, len(twb.testingBufferedWritesAsSlice()))
@@ -1497,6 +1503,9 @@ func TestTxnWriteBufferFlushesWhenOverBudget(t *testing.T) {
 	require.NotNil(t, br)
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.EndTxnResponse{}, br.Responses[0].GetInner())
+	// Sanity check the metrics remain the same (and don't increase).
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferMemoryLimitExceeded.Count())
 }
 
 // TestTxnWriteBufferDeleteRange ensures that the txnWriteBuffer correctly
@@ -1563,6 +1572,7 @@ func TestTxnWriteBufferDeleteRange(t *testing.T) {
 	// Even though we flushed some writes, it shouldn't make it back to the response.
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.DeleteRangeResponse{}, br.Responses[0].GetInner())
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 
 	// Ensure the buffer is empty at this point.
 	require.Equal(t, 0, len(twb.testingBufferedWritesAsSlice()))
@@ -1765,6 +1775,7 @@ func TestTxnWriteBufferFlushesAfterDisabling(t *testing.T) {
 	// The buffer should be flushed and disabled.
 	require.False(t, twb.enabled)
 	require.False(t, twb.flushOnNextBatch)
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 
 	// Both Put and Del should make it to the server in a single bath.
 	require.Equal(t, numCalledBefore+1, mockSender.NumCalled())
@@ -1813,6 +1824,8 @@ func TestTxnWriteBufferFlushesAfterDisabling(t *testing.T) {
 	require.NotNil(t, br)
 	require.Len(t, br.Responses, 1)
 	require.IsType(t, &kvpb.EndTxnResponse{}, br.Responses[0].GetInner())
+	// Sanity check the metric remains the same (and doesn't increase).
+	require.Equal(t, int64(1), twb.txnMetrics.TxnWriteBufferDisabledAfterBuffering.Count())
 }
 
 // TestTxnWriteBufferClearsBufferOnEpochBump tests that the txnWriteBuffer

pkg/kv/kvclient/kvcoord/txn_metrics.go

@@ -55,6 +55,12 @@ type TxnMetrics struct {
 	// End transaction failure counters.
 	RollbacksFailed      *metric.Counter
 	AsyncRollbacksFailed *metric.Counter
+
+	// Write buffering metrics.
+	TxnWriteBufferEnabled                *metric.Counter
+	TxnWriteBufferDisabledAfterBuffering *metric.Counter
+	TxnWriteBufferMemoryLimitExceeded    *metric.Counter
+	TxnWriteBufferFullyHandledBatches    *metric.Counter
 }
 
 var (
@@ -272,16 +278,40 @@ var (
 	}
 	metaRollbacksFailed = metric.Metadata{
 		Name:        "txn.rollbacks.failed",
-		Help:        "Number of KV transaction that failed to send final abort",
+		Help:        "Number of KV transactions that failed to send final abort",
 		Measurement: "KV Transactions",
 		Unit:        metric.Unit_COUNT,
 	}
 	metaAsyncRollbacksFailed = metric.Metadata{
 		Name:        "txn.rollbacks.async.failed",
-		Help:        "Number of KV transaction that failed to send abort asynchronously which is not always retried",
+		Help:        "Number of KV transactions that failed to send abort asynchronously which is not always retried",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferEnabled = metric.Metadata{
+		Name:        "txn.write_buffering.num_enabled",
+		Help:        "Number of KV transactions that enabled buffered writes",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferDisabledAfterBuffering = metric.Metadata{
+		Name:        "txn.write_buffering.disabled_after_buffering",
+		Help:        "Number of KV transactions that disabled write buffering after buffering some writes but before an EndTxn request",
+		Measurement: "KV Transactions",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaTxnWriteBufferLimitExceeded = metric.Metadata{
+		Name:        "txn.write_buffering.memory_limit_exceeded",
+		Help:        "Number of KV transactions that exceeded the write buffering memory limit",
 		Measurement: "KV Transactions",
 		Unit:        metric.Unit_COUNT,
 	}
+	metaTxnWriteBufferFullyHandledBatches = metric.Metadata{
+		Name:        "txn.write_buffering.batches.fully_handled",
+		Help:        "Number of KV batches that were fully handled by the write buffer (not sent to KV)",
+		Measurement: "KV Batches",
+		Unit:        metric.Unit_COUNT,
+	}
 )
 
 // MakeTxnMetrics returns a TxnMetrics struct that contains metrics whose
@@ -321,15 +351,19 @@ func MakeTxnMetrics(histogramWindow time.Duration) TxnMetrics {
 			SigFigs:      3,
 			BucketConfig: metric.Count1KBuckets,
 		}),
-		RestartsWriteTooOld:            telemetry.NewCounterWithMetric(metaRestartsWriteTooOld),
-		RestartsSerializable:           telemetry.NewCounterWithMetric(metaRestartsSerializable),
-		RestartsAsyncWriteFailure:      telemetry.NewCounterWithMetric(metaRestartsAsyncWriteFailure),
-		RestartsCommitDeadlineExceeded: telemetry.NewCounterWithMetric(metaRestartsCommitDeadlineExceeded),
-		RestartsReadWithinUncertainty:  telemetry.NewCounterWithMetric(metaRestartsReadWithinUncertainty),
-		RestartsTxnAborted:             telemetry.NewCounterWithMetric(metaRestartsTxnAborted),
-		RestartsTxnPush:                telemetry.NewCounterWithMetric(metaRestartsTxnPush),
-		RestartsUnknown:                telemetry.NewCounterWithMetric(metaRestartsUnknown),
-		RollbacksFailed:                metric.NewCounter(metaRollbacksFailed),
-		AsyncRollbacksFailed:           metric.NewCounter(metaAsyncRollbacksFailed),
+		RestartsWriteTooOld:                  telemetry.NewCounterWithMetric(metaRestartsWriteTooOld),
+		RestartsSerializable:                 telemetry.NewCounterWithMetric(metaRestartsSerializable),
+		RestartsAsyncWriteFailure:            telemetry.NewCounterWithMetric(metaRestartsAsyncWriteFailure),
+		RestartsCommitDeadlineExceeded:       telemetry.NewCounterWithMetric(metaRestartsCommitDeadlineExceeded),
+		RestartsReadWithinUncertainty:        telemetry.NewCounterWithMetric(metaRestartsReadWithinUncertainty),
+		RestartsTxnAborted:                   telemetry.NewCounterWithMetric(metaRestartsTxnAborted),
+		RestartsTxnPush:                      telemetry.NewCounterWithMetric(metaRestartsTxnPush),
+		RestartsUnknown:                      telemetry.NewCounterWithMetric(metaRestartsUnknown),
+		RollbacksFailed:                      metric.NewCounter(metaRollbacksFailed),
+		AsyncRollbacksFailed:                 metric.NewCounter(metaAsyncRollbacksFailed),
+		TxnWriteBufferEnabled:                metric.NewCounter(metaTxnWriteBufferEnabled),
+		TxnWriteBufferDisabledAfterBuffering: metric.NewCounter(metaTxnWriteBufferDisabledAfterBuffering),
+		TxnWriteBufferMemoryLimitExceeded:    metric.NewCounter(metaTxnWriteBufferLimitExceeded),
+		TxnWriteBufferFullyHandledBatches:    metric.NewCounter(metaTxnWriteBufferFullyHandledBatches),
 	}
 }

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -1659,7 +1659,16 @@ func TestTxnBasicBufferedWrites(t *testing.T) {
 		require.NoError(t, txn.Commit(ctx))
 
 		err := s.DB.Txn(ctx, func(ctx context.Context, txn *kv.Txn) error {
+			metrics := txn.Sender().(*kvcoord.TxnCoordSender).Metrics()
+			before := metrics.TxnWriteBufferEnabled.Count()
 			txn.SetBufferedWritesEnabled(true)
+			after := metrics.TxnWriteBufferEnabled.Count()
+			require.Equal(t, after-before, int64(1))
+
+			// Ensure that enabling buffered writes after we've already done so
+			// doesn't increment the metric.
+			txn.SetBufferedWritesEnabled(true)
+			require.Equal(t, after, metrics.TxnWriteBufferEnabled.Count())
 
 			// Put transactional value.
 			if err := txn.Put(ctx, keyA, value1); err != nil {