eventpb: add storage event types; log periodically per-store

Add the `StoreStats` event type, a per-store event emitted to the
`TELEMETRY` logging channel. This event type will be computed from the
Pebble metrics for each store.

Emit a `StoreStats` event periodically, by default, once per hour, per
store.

Touches #85589.

Release note: None.

Release justification: low risk, high benefit changes to existing
functionality.

Makefile

@@ -1629,6 +1629,7 @@ EVENTPB_PROTOS = \
   pkg/util/log/eventpb/cluster_events.proto \
   pkg/util/log/eventpb/job_events.proto \
   pkg/util/log/eventpb/health_events.proto \
+  pkg/util/log/eventpb/storage_events.proto \
   pkg/util/log/eventpb/telemetry.proto
 
 EVENTLOG_PROTOS = pkg/util/log/logpb/event.proto $(EVENTPB_PROTOS)

pkg/kv/kvserver/BUILD.bazel

@@ -186,6 +186,7 @@ go_library(
         "//pkg/util/iterutil",
         "//pkg/util/limit",
         "//pkg/util/log",
+        "//pkg/util/log/logcrash",
         "//pkg/util/metric",
         "//pkg/util/metric/aggmetric",
         "//pkg/util/mon",

pkg/kv/kvserver/store.go

@@ -70,6 +70,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/cockroach/pkg/util/limit"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logcrash"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/mon"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
@@ -125,6 +126,13 @@ var storeSchedulerConcurrency = envutil.EnvOrDefaultInt(
 var logSSTInfoTicks = envutil.EnvOrDefaultInt(
 	"COCKROACH_LOG_SST_INFO_TICKS_INTERVAL", 60)
 
+// By default, telemetry events are emitted once per hour, per store:
+// (10s tick interval) * 6 * 60 = 3600s = 1h.
+var logStoreTelemetryTicks = envutil.EnvOrDefaultInt(
+	"COCKROACH_LOG_STORE_TELEMETRY_TICKS_INTERVAL",
+	6*60,
+)
+
 // bulkIOWriteLimit is defined here because it is used by BulkIOWriteLimiter.
 var bulkIOWriteLimit = settings.RegisterByteSizeSetting(
 	settings.TenantWritable,
@@ -3343,6 +3351,17 @@ func (s *Store) ComputeMetrics(ctx context.Context, tick int) error {
 		// will not contain the log prefix.
 		log.Infof(ctx, "\n%s", m.Metrics)
 	}
+	// Periodically emit a store stats structured event to the TELEMETRY channel,
+	// if reporting is enabled. These events are intended to be emitted at low
+	// frequency. Trigger on tick 1 for the same reasons as above.
+	if logcrash.DiagnosticsReportingEnabled.Get(&s.ClusterSettings().SV) &&
+		tick%logStoreTelemetryTicks == 1 {
+		// The stats event is populated from a subset of the Metrics.
+		e := m.AsStoreStatsEvent()
+		e.NodeId = int32(s.NodeID())
+		e.StoreId = int32(s.StoreID())
+		log.StructuredEvent(ctx, &e)
+	}
 	return nil
 }
 

pkg/storage/BUILD.bazel

@@ -69,6 +69,7 @@ go_library(
         "//pkg/util/humanizeutil",
         "//pkg/util/iterutil",
         "//pkg/util/log",
+        "//pkg/util/log/eventpb",
         "//pkg/util/mon",
         "//pkg/util/protoutil",
         "//pkg/util/syncutil",

pkg/storage/engine.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/iterutil"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/protoutil"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
@@ -1026,6 +1027,64 @@ func (m *Metrics) CompactedBytes() (read, written uint64) {
 	return read, written
 }
 
+// AsStoreStatsEvent converts a Metrics struct into an eventpb.StoreStats event,
+// suitable for logging to the telemetry channel.
+func (m *Metrics) AsStoreStatsEvent() eventpb.StoreStats {
+	e := eventpb.StoreStats{
+		CacheSize:                  m.BlockCache.Size,
+		CacheCount:                 m.BlockCache.Count,
+		CacheHits:                  m.BlockCache.Hits,
+		CacheMisses:                m.BlockCache.Misses,
+		CompactionCountDefault:     m.Compact.DefaultCount,
+		CompactionCountDeleteOnly:  m.Compact.DeleteOnlyCount,
+		CompactionCountElisionOnly: m.Compact.ElisionOnlyCount,
+		CompactionCountMove:        m.Compact.MoveCount,
+		CompactionCountRead:        m.Compact.ReadCount,
+		CompactionCountRewrite:     m.Compact.RewriteCount,
+		CompactionNumInProgress:    m.Compact.NumInProgress,
+		CompactionMarkedFiles:      int64(m.Compact.MarkedFiles),
+		FlushCount:                 m.Flush.Count,
+		MemtableSize:               m.MemTable.Size,
+		MemtableCount:              m.MemTable.Count,
+		MemtableZombieCount:        m.MemTable.ZombieCount,
+		MemtableZombieSize:         m.MemTable.ZombieSize,
+		WalLiveCount:               m.WAL.Files,
+		WalLiveSize:                m.WAL.Size,
+		WalObsoleteCount:           m.WAL.ObsoleteFiles,
+		WalObsoleteSize:            m.WAL.ObsoletePhysicalSize,
+		WalPhysicalSize:            m.WAL.PhysicalSize,
+		WalBytesIn:                 m.WAL.BytesIn,
+		WalBytesWritten:            m.WAL.BytesWritten,
+		TableObsoleteCount:         m.Table.ObsoleteCount,
+		TableObsoleteSize:          m.Table.ObsoleteSize,
+		TableZombieCount:           m.Table.ZombieCount,
+		TableZombieSize:            m.Table.ZombieSize,
+	}
+	for i, l := range m.Levels {
+		if l.NumFiles == 0 {
+			continue
+		}
+		e.Levels = append(e.Levels, eventpb.LevelStats{
+			Level:           uint32(i),
+			NumFiles:        l.NumFiles,
+			SizeBytes:       l.Size,
+			Score:           float32(l.Score),
+			BytesIn:         l.BytesIn,
+			BytesIngested:   l.BytesIngested,
+			BytesMoved:      l.BytesMoved,
+			BytesRead:       l.BytesRead,
+			BytesCompacted:  l.BytesCompacted,
+			BytesFlushed:    l.BytesFlushed,
+			TablesCompacted: l.TablesCompacted,
+			TablesFlushed:   l.TablesFlushed,
+			TablesIngested:  l.TablesIngested,
+			TablesMoved:     l.TablesMoved,
+			NumSublevels:    l.Sublevels,
+		})
+	}
+	return e
+}
+
 // EnvStats is a set of RocksDB env stats, including encryption status.
 type EnvStats struct {
 	// TotalFiles is the total number of files reported by rocksdb.

pkg/util/log/eventpb/BUILD.bazel

@@ -58,6 +58,7 @@ proto_library(
         "role_events.proto",
         "session_events.proto",
         "sql_audit_events.proto",
+        "storage_events.proto",
         "telemetry.proto",
         "zone_events.proto",
     ],

pkg/util/log/eventpb/PROTOS.bzl

@@ -17,6 +17,7 @@ EVENTPB_PROTOS = [
     "cluster_events.proto",
     "job_events.proto",
     "health_events.proto",
+    "storage_events.proto",
     "telemetry.proto",
 ]
 

pkg/util/log/eventpb/event_test.go

@@ -58,6 +58,13 @@ func TestEventJSON(t *testing.T) {
 		// `includeempty` annotation, so nothing is emitted, despite the presence of
 		// zero values.
 		{&SchemaSnapshotMetadata{SnapshotID: "", NumRecords: 0}, ""},
+
+		// Primitive fields with an `includeempty` annotation will emit their zero
+		// value.
+		{
+			&StoreStats{Levels: []LevelStats{{Level: 0, NumFiles: 1}, {Level: 6, NumFiles: 2}}},
+			`"Levels":[{"Level":0,"NumFiles":1},{"Level":6,"NumFiles":2}]`,
+		},
 	}
 
 	for _, tc := range testCases {

pkg/util/log/eventpb/eventpbgen/gen.go

@@ -660,6 +660,18 @@ func (m *{{.GoType}}) AppendJSONFields(printComma bool, b redact.RedactableBytes
    {{ if not .AllowZeroValue -}}
    }
    {{- end }}
+   {{- else if eq .FieldType "array_of_LevelStats"}}
+   if len(m.{{.FieldName}}) > 0 {
+     if printComma { b = append(b, ',')}; printComma = true
+     b = append(b, "\"{{.FieldName}}\":["...)
+     for i, l := range m.{{.FieldName}} {
+       if i > 0 { b = append(b, ',') }
+       b = append(b, '{')
+			 printComma, b = l.AppendJSONFields(false, b)
+       b = append(b, '}')
+     }
+     b = append(b, ']')
+   }
    {{- else if eq .FieldType "protobuf"}}
    if m.{{.FieldName}} != nil {
      if printComma { b = append(b, ',')}; printComma = true