Expose Ingestion Count on Cockroach Side

Fixes: #103744

This change exposes the `Ingest.Count` metric from Pebble to the
Cockroach side.

Related PR (Pebble Changes): cockroachdb/pebble#2568
Release note: None

docs/generated/eventlog.md

@@ -2708,6 +2708,7 @@ Note that because stats are scoped to the lifetime of the process, counters
 | `FlushIngestCount` |  | no |
 | `FlushIngestTableCount` |  | no |
 | `FlushIngestTableBytes` |  | no |
+| `IngestCount` | ingest_count is the number of successful ingest operations (counter). | no |
 | `MemtableSize` | memtable_size is the total size allocated to all memtables and (large) batches, in bytes (gauge). | no |
 | `MemtableCount` | memtable_count is the count of memtables (gauge). | no |
 | `MemtableZombieCount` | memtable_zombie_count is the count of memtables no longer referenced by the current DB state, but still in use by an iterator (gauge). | no |

pkg/kv/kvserver/metrics.go

@@ -449,6 +449,14 @@ var (
 		Unit:        metric.Unit_COUNT,
 	}
 
+	//Ingest metrics
+	metaIngestCount = metric.Metadata{
+		Name:        "storage.ingest.count",
+		Help:        "Number of successful ingestions performed",
+		Measurement: "Events",
+		Unit:        metric.Unit_COUNT,
+	}
+
 	// Pebble metrics.
 	metaRdbBlockCacheHits = metric.Metadata{
 		Name:        "rocksdb.block.cache.hits",
@@ -2325,6 +2333,9 @@ type StoreMetrics struct {
 	MaxLockWaitDurationNanos       *metric.Gauge
 	MaxLockWaitQueueWaitersForLock *metric.Gauge
 
+	// Ingestion metrics
+	IngestCount *metric.Gauge
+
 	// Closed timestamp metrics.
 	ClosedTimestampMaxBehindNanos *metric.Gauge
 
@@ -2764,6 +2775,9 @@ func newStoreMetrics(histogramWindow time.Duration) *StoreMetrics {
 		BatchCommitWALRotWaitDuration: metric.NewGauge(metaBatchCommitWALRotDuration),
 		BatchCommitCommitWaitDuration: metric.NewGauge(metaBatchCommitCommitWaitDuration),
 
+		// Ingestion metrics
+		IngestCount: metric.NewGauge(metaIngestCount),
+
 		RdbCheckpoints: metric.NewGauge(metaRdbCheckpoints),
 
 		// Disk health metrics.
@@ -3114,6 +3128,7 @@ func (sm *StoreMetrics) updateEngineMetrics(m storage.Metrics) {
 	sm.FlushableIngestCount.Update(int64(m.Flush.AsIngestCount))
 	sm.FlushableIngestTableCount.Update(int64(m.Flush.AsIngestTableCount))
 	sm.FlushableIngestTableSize.Update(int64(m.Flush.AsIngestBytes))
+	sm.IngestCount.Update(int64(m.Ingest.Count))
 	sm.BatchCommitCount.Update(int64(m.BatchCommitStats.Count))
 	sm.BatchCommitDuration.Update(int64(m.BatchCommitStats.TotalDuration))
 	sm.BatchCommitSemWaitDuration.Update(int64(m.BatchCommitStats.SemaphoreWaitDuration))
@@ -3122,7 +3137,6 @@ 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/storage/engine.go

@@ -1189,6 +1189,7 @@ func (m *Metrics) AsStoreStatsEvent() eventpb.StoreStats {
 		FlushIngestCount:           m.Flush.AsIngestCount,
 		FlushIngestTableCount:      m.Flush.AsIngestTableCount,
 		FlushIngestTableBytes:      m.Flush.AsIngestBytes,
+		IngestCount:                m.Ingest.Count,
 		MemtableSize:               m.MemTable.Size,
 		MemtableCount:              m.MemTable.Count,
 		MemtableZombieCount:        m.MemTable.ZombieCount,

pkg/util/log/eventpb/storage_events.proto

@@ -75,6 +75,11 @@ message StoreStats {
   uint64 flush_ingest_table_count = 36 [(gogoproto.jsontag) = ",omitempty"];
   uint64 flush_ingest_table_bytes = 37 [(gogoproto.jsontag) = ",omitempty"];
 
+  // Ingest stats.
+
+  // ingest_count is the number of successful ingest operations (counter).
+  uint64 ingest_count = 38 [(gogoproto.jsontag) = ",omitempty"];
+
   // Memtable stats.
 
   // memtable_size is the total size allocated to all memtables and (large)