storeliveness: add batching metrics to storeliveness transport

Previously, the only metrics on storeliveness transport for batching
is a simple counter that shows how many aggregate storeliveness
messages have been sent.

For paced storeliveness heartbeats verification, we need to compare the
same metrics with the baseline to see if the changes introduced via
pacing would have some negative impact on heartbeat sending -
particularly with respect to batching.

This commit adds the same metrics that are present in paced
storeliveness heartbeats; additionally, since this commit only
introduces metrics, there will not be any impact on current behaviour.

Fixes: None
Release note: None

Author: dodeca12

docs/generated/metrics/metrics.yaml

@@ -17365,6 +17365,22 @@ layers:
       unit: COUNT
       aggregation: AVG
       derivative: NON_NEGATIVE_DERIVATIVE
+    - name: storeliveness.transport.batches-received
+      exported_name: storeliveness_transport_batches_received
+      description: Number of message batches received by the Store Liveness Transport
+      y_axis_label: Batches
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
+    - name: storeliveness.transport.batches-sent
+      exported_name: storeliveness_transport_batches_sent
+      description: Number of message batches sent by the Store Liveness Transport
+      y_axis_label: Batches
+      type: COUNTER
+      unit: COUNT
+      aggregation: AVG
+      derivative: NON_NEGATIVE_DERIVATIVE
     - name: storeliveness.transport.receive-queue-bytes
       exported_name: storeliveness_transport_receive_queue_bytes
       description: Total byte size of pending incoming messages from Store Liveness Transport

pkg/kv/kvserver/storeliveness/metrics.go

@@ -28,6 +28,9 @@ type TransportMetrics struct {
 	MessagesReceived       *metric.Counter
 	MessagesSendDropped    *metric.Counter
 	MessagesReceiveDropped *metric.Counter
+
+	BatchesSent     *metric.Counter
+	BatchesReceived *metric.Counter
 }
 
 func newTransportMetrics() *TransportMetrics {
@@ -39,6 +42,8 @@ func newTransportMetrics() *TransportMetrics {
 		MessagesReceived:       metric.NewCounter(metaMessagesReceived),
 		MessagesSendDropped:    metric.NewCounter(metaMessagesSendDropped),
 		MessagesReceiveDropped: metric.NewCounter(metaMessagesReceiveDropped),
+		BatchesSent:            metric.NewCounter(metaBatchesSent),
+		BatchesReceived:        metric.NewCounter(metaBatchesReceived),
 	}
 }
 
@@ -201,11 +206,22 @@ var (
 		Measurement: "Bytes",
 		Unit:        metric.Unit_BYTES,
 	}
-
 	metaCallbacksProcessingDuration = metric.Metadata{
 		Name:        "storeliveness.callbacks.processing_duration",
 		Help:        "Duration of support withdrawal callback processing",
 		Measurement: "Duration",
 		Unit:        metric.Unit_NANOSECONDS,
 	}
+	metaBatchesSent = metric.Metadata{
+		Name:        "storeliveness.transport.batches-sent",
+		Help:        "Number of message batches sent by the Store Liveness Transport",
+		Measurement: "Batches",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaBatchesReceived = metric.Metadata{
+		Name:        "storeliveness.transport.batches-received",
+		Help:        "Number of message batches received by the Store Liveness Transport",
+		Measurement: "Batches",
+		Unit:        metric.Unit_COUNT,
+	}
 )

pkg/kv/kvserver/storeliveness/transport.go

@@ -168,6 +168,7 @@ func (t *Transport) stream(stream slpb.RPCStoreLiveness_StreamStream) error {
 					if err != nil {
 						return err
 					}
+					t.metrics.BatchesReceived.Inc(1)
 					if !batch.Now.IsEmpty() {
 						t.clock.Update(batch.Now)
 					}
@@ -397,6 +398,8 @@ func (t *Transport) processQueue(
 				t.metrics.MessagesSendDropped.Inc(int64(len(batch.Messages)))
 				return err
 			}
+
+			t.metrics.BatchesSent.Inc(1)
 			t.metrics.MessagesSent.Inc(int64(len(batch.Messages)))
 
 			// Reuse the Messages slice, but zero out the contents to avoid delaying

pkg/roachprod/agents/opentelemetry/cockroachdb_metrics.go

@@ -2446,6 +2446,8 @@ var cockroachdbMetrics = map[string]string{
     "storeliveness_support_from_stores": "storeliveness.support_from.stores",
     "storeliveness_support_withdraw_failures": "storeliveness.support_withdraw.failures",
     "storeliveness_support_withdraw_successes": "storeliveness.support_withdraw.successes",
+    "storeliveness_transport_batches_received": "storeliveness.transport.batches_received",
+    "storeliveness_transport_batches_sent": "storeliveness.transport.batches_sent",
     "storeliveness_transport_receive_dropped": "storeliveness.transport.receive_dropped",
     "storeliveness_transport_receive_queue_bytes": "storeliveness.transport.receive_queue_bytes",
     "storeliveness_transport_receive_queue_size": "storeliveness.transport.receive_queue_size",