contention: periodically log resolved contention events

This commit adds logging to the flushAndResolve method in the contention event
store. This method is called periodically at an interval defined by the
(non-public) cluster setting sql.contention.event_store.resolution_interval.

By logging aggregated contention event information, users now have the ability
to find which waiting statements/txns are experiencing high contention through
logs (previously this was only available via insights or SQL activity).

Additionally, this aggregated event information contains the blocking txn and
contended key - information that was previously only available via querying
crdb_internal.transaction_contention_events.

The logs are aggregated by a composite identifier of
(waiting stmt, waiting txn, blocking txn, key), and the total contention time
observed for that combination. They are structured logs. For example:

```
{"Timestamp":1756224167482848000,"EventType":"aggregated_contention_info",
"WaitingStmtFingerprintId":"\\x000000000000007b",
"WaitingTxnFingerprintId":"\\x00000000000001c8",
"BlockingTxnFingerprintId":"\\x0000000000000315",
"ContendedKey":"test-key-1","Duration":300000000}
```

Since this information is available via logs, it provides a solution for users
who desire more persistence of contention event information.

Part of: https://cockroachlabs.atlassian.net/browse/CRDB-53592
Release note (ops change): Adds logging of contention events.

Author: alyshanjahani-crl

docs/generated/eventlog.md

@@ -409,6 +409,35 @@ is shut down inside the same process as the KV layer.
 | `InstanceID` | The ID of the server instance. | no |
 | `TenantName` | The name of the tenant at the time the event was emitted. | yes |
 
+## Contention events
+
+Aggregated information about contention events.
+
+Events in this category are logged to the `SQL_EXEC` channel.
+
+
+### `aggregated_contention_info`
+
+An event of type `aggregated_contention_info` is recorded periodically when contention events
+are resolved.
+
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `WaitingStmtFingerprintId` |  | no |
+| `WaitingTxnFingerprintId` |  | no |
+| `BlockingTxnFingerprintId` |  | no |
+| `ContendedKey` |  | partially |
+| `Duration` |  | no |
+
+
+#### Common fields
+
+| Field | Description | Sensitive |
+|--|--|--|
+| `Timestamp` | The timestamp of the event. Expressed as nanoseconds since the Unix epoch. | no |
+| `EventType` | The type of the event. | no |
+
 ## Debugging events
 
 Events in this category pertain to debugging operations performed by

pkg/sql/contention/BUILD.bazel

@@ -23,15 +23,19 @@ go_library(
         "//pkg/sql/catalog/descpb",
         "//pkg/sql/contention/contentionutils",
         "//pkg/sql/contentionpb",
+        "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
         "//pkg/util/cache",
         "//pkg/util/log",
+        "//pkg/util/log/eventpb",
+        "//pkg/util/log/logpb",
         "//pkg/util/metric",
         "//pkg/util/stop",
         "//pkg/util/syncutil",
         "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "@com_github_biogo_store//llrb",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
     ],
 )
 
@@ -53,18 +57,24 @@ go_test(
         "//pkg/settings/cluster",
         "//pkg/sql/appstatspb",
         "//pkg/sql/contentionpb",
+        "//pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil",
         "//pkg/storage/enginepb",
         "//pkg/testutils",
         "//pkg/testutils/datapathutils",
         "//pkg/util/cache",
         "//pkg/util/encoding",
         "//pkg/util/leaktest",
+        "//pkg/util/log",
+        "//pkg/util/log/eventpb",
+        "//pkg/util/log/logpb",
+        "//pkg/util/log/logtestutils",
         "//pkg/util/randutil",
         "//pkg/util/stop",
         "//pkg/util/timeutil",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_redact//:redact",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/sql/contention/event_store.go

@@ -13,13 +13,18 @@ import (
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
+	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/contention/contentionutils"
 	"github.com/cockroachdb/cockroach/pkg/sql/contentionpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil"
 	"github.com/cockroachdb/cockroach/pkg/util/cache"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logpb"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/redact"
 )
 
 const (
@@ -296,14 +301,14 @@ func (s *eventStore) getEventByEventHash(
 }
 
 // flushAndResolve is the main method called by the resolver goroutine each
-// time the timer fires. This method does two things:
+// time the timer fires. This method does three things:
 //  1. it triggers the batching buffer to flush its content into the intake
 //     goroutine. This is to ensure that in the case where we have very low
 //     rate of contentions, the contention events won't be permanently trapped
 //     in the batching buffer.
 //  2. it invokes the dequeue() method on the resolverQueue. This cause the
 //     resolver to perform txnID resolution. See inline comments on the method
-//     for details.
+//  3. Lastly, it logs the resolved events.
 func (s *eventStore) flushAndResolve(ctx context.Context) error {
 	// This forces the write-buffer flushes its batch into the intake goroutine.
 	// The intake goroutine will asynchronously add all events in the batch
@@ -320,6 +325,9 @@ func (s *eventStore) flushAndResolve(ctx context.Context) error {
 	// before we bubble up the error.
 	s.upsertBatch(result)
 
+	// Aggregate the resolved event information for logging.
+	logResolvedEvents(ctx, result)
+
 	return err
 }
 
@@ -348,6 +356,42 @@ func (s *eventStore) resolutionIntervalWithJitter() time.Duration {
 	return jitteredInterval
 }
 
+// contentionKey is used as a key in the eventsAggregated map to group
+// contention events by their characteristics.
+type contentionKey struct {
+	waitingStmtFingerprintID appstatspb.StmtFingerprintID
+	waitingTxnFingerprintID  appstatspb.TransactionFingerprintID
+	blockingTxnFingerprintID appstatspb.TransactionFingerprintID
+	contendedKey             redact.RedactableString
+}
+
+func logResolvedEvents(ctx context.Context, events []contentionpb.ExtendedContentionEvent) {
+	eventsAggregated := make(map[contentionKey]time.Duration)
+	for _, event := range events {
+		// Create a struct key instead of a composite string
+		key := contentionKey{
+			waitingStmtFingerprintID: event.WaitingStmtFingerprintID,
+			waitingTxnFingerprintID:  event.WaitingTxnFingerprintID,
+			blockingTxnFingerprintID: event.BlockingTxnFingerprintID,
+			contendedKey:             redact.Sprint(event.BlockingEvent.Key),
+		}
+		// Add the contention time to the existing total for this key combination
+		eventsAggregated[key] += event.BlockingEvent.Duration
+	}
+
+	for key, duration := range eventsAggregated {
+		// Create an AggregatedContentionInfo event for structured logging
+		event := &eventpb.AggregatedContentionInfo{
+			WaitingStmtFingerprintId: "\\x" + sqlstatsutil.EncodeStmtFingerprintIDToString(key.waitingStmtFingerprintID),
+			WaitingTxnFingerprintId:  "\\x" + sqlstatsutil.EncodeTxnFingerprintIDToString(key.waitingTxnFingerprintID),
+			BlockingTxnFingerprintId: "\\x" + sqlstatsutil.EncodeTxnFingerprintIDToString(key.blockingTxnFingerprintID),
+			ContendedKey:             key.contendedKey,
+			Duration:                 duration.Nanoseconds(),
+		}
+		log.StructuredEvent(ctx, logpb.Severity_INFO, event)
+	}
+}
+
 func entryBytes(event *contentionpb.ExtendedContentionEvent) int {
 	// Since we store the event's hash as the key to the unordered cache,
 	// this is means we are storing another copy of uint64 (8 bytes).

pkg/sql/contention/event_store_test.go

@@ -19,12 +19,18 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/contentionpb"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logpb"
+	"github.com/cockroachdb/cockroach/pkg/util/log/logtestutils"
 	"github.com/cockroachdb/cockroach/pkg/util/stop"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/redact"
 	"github.com/stretchr/testify/require"
 )
 
@@ -249,6 +255,171 @@ func BenchmarkEventStoreIntake(b *testing.B) {
 	}
 }
 
+func TestLogResolvedEvents(t *testing.T) {
+	ctx := context.Background()
+
+	logCommonDetails := logpb.CommonEventDetails{
+		Timestamp: timeutil.Now().UnixNano(),
+		EventType: "aggregated_contention_info",
+	}
+	// Set up log interception to capture structured AggregatedContentionInfo events
+	spy := logtestutils.NewStructuredLogSpy(
+		t,
+		[]logpb.Channel{logpb.Channel_SQL_EXEC},
+		[]string{"aggregated_contention_info"},
+		func(entry logpb.Entry) (eventpb.AggregatedContentionInfo, error) {
+			le, err := logtestutils.FromLogEntry[eventpb.AggregatedContentionInfo](entry)
+			if err != nil {
+				return le, err
+			}
+			le.CommonEventDetails = logCommonDetails
+			return le, nil
+		},
+	)
+	cleanup := log.InterceptWith(ctx, spy)
+	defer cleanup()
+
+	fingerprintIDToHex := func(id uint64) string {
+		return "\\x" + sqlstatsutil.EncodeStmtFingerprintIDToString(appstatspb.StmtFingerprintID(id))
+	}
+	txnFingerprintIDToHex := func(id uint64) string {
+		return "\\x" + sqlstatsutil.EncodeTxnFingerprintIDToString(appstatspb.TransactionFingerprintID(id))
+	}
+
+	type testCase struct {
+		events   []contentionpb.ExtendedContentionEvent
+		expected []eventpb.AggregatedContentionInfo
+	}
+
+	testCases := []testCase{
+		{
+			// Two events with same waiting (stmt, txn), blocking txn and key
+			events: []contentionpb.ExtendedContentionEvent{
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  456,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-1"),
+						Duration: 100 * time.Millisecond,
+					},
+				},
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  456,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-1"),
+						Duration: 200 * time.Millisecond,
+					},
+				},
+			},
+			expected: []eventpb.AggregatedContentionInfo{
+				{
+					CommonEventDetails:       logCommonDetails,
+					WaitingStmtFingerprintId: fingerprintIDToHex(123),
+					WaitingTxnFingerprintId:  txnFingerprintIDToHex(456),
+					BlockingTxnFingerprintId: txnFingerprintIDToHex(789),
+					ContendedKey:             redact.Sprint("\"test-key-1\""),
+					Duration:                 300000000,
+				},
+			},
+		},
+		{
+			// Different key.
+			events: []contentionpb.ExtendedContentionEvent{
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  456,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-1"),
+						Duration: 100 * time.Millisecond,
+					},
+				},
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  456,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-2"),
+						Duration: 200 * time.Millisecond,
+					},
+				},
+			},
+			expected: []eventpb.AggregatedContentionInfo{
+				{
+					CommonEventDetails:       logCommonDetails,
+					WaitingStmtFingerprintId: fingerprintIDToHex(123),
+					WaitingTxnFingerprintId:  txnFingerprintIDToHex(456),
+					BlockingTxnFingerprintId: txnFingerprintIDToHex(789),
+					ContendedKey:             redact.Sprint("\"test-key-1\""),
+					Duration:                 100000000,
+				},
+				{
+					CommonEventDetails:       logCommonDetails,
+					WaitingStmtFingerprintId: fingerprintIDToHex(123),
+					WaitingTxnFingerprintId:  txnFingerprintIDToHex(456),
+					BlockingTxnFingerprintId: txnFingerprintIDToHex(789),
+					ContendedKey:             redact.Sprint("\"test-key-2\""),
+					Duration:                 200000000,
+				},
+			},
+		},
+		{
+			// Different fingerprint IDs.
+			events: []contentionpb.ExtendedContentionEvent{
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  456,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-1"),
+						Duration: 100 * time.Millisecond,
+					},
+				},
+				{
+					WaitingStmtFingerprintID: 123,
+					WaitingTxnFingerprintID:  654,
+					BlockingTxnFingerprintID: 789,
+					BlockingEvent: kvpb.ContentionEvent{
+						Key:      []byte("test-key-1"),
+						Duration: 200 * time.Millisecond,
+					},
+				},
+			},
+			expected: []eventpb.AggregatedContentionInfo{
+				{
+					CommonEventDetails:       logCommonDetails,
+					WaitingStmtFingerprintId: fingerprintIDToHex(123),
+					WaitingTxnFingerprintId:  txnFingerprintIDToHex(456),
+					BlockingTxnFingerprintId: txnFingerprintIDToHex(789),
+					ContendedKey:             redact.Sprint("\"test-key-1\""),
+					Duration:                 100000000,
+				},
+				{
+					CommonEventDetails:       logCommonDetails,
+					WaitingStmtFingerprintId: fingerprintIDToHex(123),
+					WaitingTxnFingerprintId:  txnFingerprintIDToHex(654),
+					BlockingTxnFingerprintId: txnFingerprintIDToHex(789),
+					ContendedKey:             redact.Sprint("\"test-key-1\""),
+					Duration:                 200000000,
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		spy.Reset()
+		logResolvedEvents(ctx, tc.events)
+		contentionLogs := spy.GetLogs(logpb.Channel_SQL_EXEC)
+		require.Len(t, contentionLogs, len(tc.expected))
+		for _, exepectedContentionLog := range tc.expected {
+			require.Contains(t, contentionLogs, exepectedContentionLog)
+		}
+	}
+}
+
 func eventSliceToMap(
 	events []contentionpb.ExtendedContentionEvent,
 ) map[uuid.UUID]contentionpb.ExtendedContentionEvent {

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/json_encoding.go

@@ -455,6 +455,12 @@ func EncodeUint64ToBytes(id uint64) []byte {
 	return encoding.EncodeUint64Ascending(result, id)
 }
 
-func encodeStmtFingerprintIDToString(id appstatspb.StmtFingerprintID) string {
+// EncodeStmtFingerprintIDToString returns the hex string representation of a statement fingerprint ID.
+func EncodeStmtFingerprintIDToString(id appstatspb.StmtFingerprintID) string {
+	return hex.EncodeToString(EncodeUint64ToBytes(uint64(id)))
+}
+
+// EncodeTxnFingerprintIDToString returns the hex string representation of a transaction fingerprint ID.
+func EncodeTxnFingerprintIDToString(id appstatspb.TransactionFingerprintID) string {
 	return hex.EncodeToString(EncodeUint64ToBytes(uint64(id)))
 }

pkg/sql/sqlstats/persistedsqlstats/sqlstatsutil/json_impl.go

@@ -289,7 +289,7 @@ func (s *stmtFingerprintID) decodeJSON(js json.JSON) error {
 
 func (s *stmtFingerprintID) encodeJSON() (json.JSON, error) {
 	return json.FromString(
-		encodeStmtFingerprintIDToString((appstatspb.StmtFingerprintID)(*s))), nil
+		EncodeStmtFingerprintIDToString((appstatspb.StmtFingerprintID)(*s))), nil
 }
 
 type innerTxnStats appstatspb.TransactionStatistics

pkg/util/log/eventpb/BUILD.bazel

@@ -49,6 +49,7 @@ proto_library(
     srcs = [
         "changefeed_events.proto",
         "cluster_events.proto",
+        "contention_events.proto",
         "ddl_events.proto",
         "debug_events.proto",
         "events.proto",

pkg/util/log/eventpb/PROTOS.bzl

@@ -20,6 +20,7 @@ EVENTPB_PROTOS = [
     "storage_events.proto",
     "telemetry.proto",
     "changefeed_events.proto",
+    "contention_events.proto",
 ]
 
 EVENTPB_PROTO_DEPS = [ "//pkg/util/log/logpb:event.proto", ] + EVENTPB_PROTOS