kvserver: Expose WAL logical and total bytes written

This change exposes the `WAL.BytesWritten` and `WAL.BytesIn` metric from
Pebble to the Cockroach Side.

Fixes: cockroachdb#104531
Release note: None

pkg/kv/kvserver/metrics.go

@@ -2013,7 +2013,18 @@ Note that the measurement does not include the duration for replicating the eval
 		Measurement: "Flush Utilization",
 		Unit:        metric.Unit_PERCENT,
 	}
-
+	metaWALBytesWritten = metric.Metadata{
+		Name:        "storage.wal.bytes_written",
+		Help:        "The number of bytes the storage engine has written to the WAL",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
+	metaWALBytesIn = metric.Metadata{
+		Name:        "storage.wal.bytes_in",
+		Help:        "The number of logical bytes the storage engine has written to the WAL",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
 	metaStorageFsyncLatency = metric.Metadata{
 		Name:        "storage.wal.fsync.latency",
 		Help:        "The write ahead log fsync latency",
@@ -2250,6 +2261,8 @@ type StoreMetrics struct {
 	RaftSchedulerLatency      metric.IHistogram
 	RaftTimeoutCampaign       *metric.Counter
 	RaftStorageReadBytes      *metric.Counter
+	WALBytesWritten           *metric.Gauge
+	WALBytesIn                *metric.Gauge
 
 	// Raft message metrics.
 	//
@@ -2813,6 +2826,8 @@ func newStoreMetrics(histogramWindow time.Duration) *StoreMetrics {
 		BatchCommitL0StallDuration:    metric.NewGauge(metaBatchCommitL0StallDuration),
 		BatchCommitWALRotWaitDuration: metric.NewGauge(metaBatchCommitWALRotDuration),
 		BatchCommitCommitWaitDuration: metric.NewGauge(metaBatchCommitCommitWaitDuration),
+		WALBytesWritten:               metric.NewGauge(metaWALBytesWritten),
+		WALBytesIn:                    metric.NewGauge(metaWALBytesIn),
 
 		// Ingestion metrics
 		IngestCount: metric.NewGauge(metaIngestCount),
@@ -3172,6 +3187,8 @@ func (sm *StoreMetrics) updateEngineMetrics(m storage.Metrics) {
 	sm.FlushableIngestTableCount.Update(int64(m.Flush.AsIngestTableCount))
 	sm.FlushableIngestTableSize.Update(int64(m.Flush.AsIngestBytes))
 	sm.IngestCount.Update(int64(m.Ingest.Count))
+	sm.WALBytesWritten.Update(int64(m.WAL.BytesWritten))
+	sm.WALBytesIn.Update(int64(m.WAL.BytesIn))
 	sm.BatchCommitCount.Update(int64(m.BatchCommitStats.Count))
 	sm.BatchCommitDuration.Update(int64(m.BatchCommitStats.TotalDuration))
 	sm.BatchCommitSemWaitDuration.Update(int64(m.BatchCommitStats.SemaphoreWaitDuration))
@@ -3180,6 +3197,7 @@ func (sm *StoreMetrics) updateEngineMetrics(m storage.Metrics) {
 	sm.BatchCommitL0StallDuration.Update(int64(m.BatchCommitStats.L0ReadAmpWriteStallDuration))
 	sm.BatchCommitWALRotWaitDuration.Update(int64(m.BatchCommitStats.WALRotationDuration))
 	sm.BatchCommitCommitWaitDuration.Update(int64(m.BatchCommitStats.CommitWaitDuration))
+
 	// Update the maximum number of L0 sub-levels seen.
 	sm.l0SublevelsTracker.Lock()
 	sm.l0SublevelsTracker.swag.Record(timeutil.Now(), float64(m.Levels[0].Sublevels))

pkg/ui/workspaces/db-console/src/views/cluster/containers/nodeGraphs/dashboards/storage.tsx

@@ -212,6 +212,26 @@ export default function (props: GraphDashboardProps) {
       </Axis>
     </LineGraph>,
 
+
+    <LineGraph
+      title="WAL Bytes Written"
+      sources={storeSources}
+      tenantSource={tenantSource}
+      tooltip={`Bytes written to WAL files ${tooltipSelection}.`}
+    >
+      <Axis units={AxisUnits.Bytes} label="written bytes">
+        {_.map(nodeIDs, nid => (
+          <Metric
+            key={nid}
+            name="cr.store.rocksdb.wal-bytes-written"
+            title={getNodeNameById(nid)}
+            sources={storeIDsForNode(storeIDsByNodeID, nid)}
+            nonNegativeRate
+          />
+        ))}
+      </Axis>
+    </LineGraph>,
+
     <LineGraph
       title="Compactions"
       sources={storeSources}