sql: introduce crdb_internal.historical_contention_events virtual table

This commit introduces `crdb_internal.historical_contention_events`
virtual table. This virtual tables exposes historical contention events
annoated with transaction fingerprint IDs for transactions that finishes
execution. This allows this virtual table to be joined into the
statement statistics and transaction statistics table.
The new virtual table require at least VIEWACTIVITYREDACTED permission
to access. However, in order to view the contending keys, it would
require at least VIEWACTIVITY permission or above.

Resolves cockroachdb#75904

Release note (sql change): introducing
`crdb_internal.historical_contention_events` virtual table, that exposes
historical contention events. The events exposed in the new virtual
table also include transaction fingerprint IDs for both blocking and
waiting transactions. This allows the new virtual table to be joined
into statement statistics and transaction statistics tables.
The new virtual table require at least VIEWACTIVITYREDACTED permission
to access. However, in order to view the contending keys, it would
require at least VIEWACTIVITY permission or above.

Release note (api change): introducing
GET `/_status/historicalcontentionevents` endpoint, that returns
cluster-wide in-memory historical contention events.
The endpoint require at least VIEWACTIVITYREDACTED permission
to access. However, in order to expose the contending keys, it would
require at least VIEWACTIVITY permission or above.

docs/generated/http/full.md

@@ -4342,6 +4342,51 @@ Response object for issuing Transaction ID Resolution.
 
 
 
+## HistoricalContentionEvents
+
+`GET /_status/historicalcontentionevents`
+
+HistoricalContentionEvents returns a list of un-aggregated contention
+events sorted by the collection timestamp.
+
+Support status: [reserved](#support-status)
+
+#### Request Parameters
+
+
+
+
+
+
+
+| Field | Type | Label | Description | Support status |
+| ----- | ---- | ----- | ----------- | -------------- |
+| node_id | [string](#cockroach.server.serverpb.HistoricalContentionEventsRequest-string) |  |  | [reserved](#support-status) |
+
+
+
+
+
+
+
+#### Response Parameters
+
+
+
+
+
+
+
+| Field | Type | Label | Description | Support status |
+| ----- | ---- | ----- | ----------- | -------------- |
+| events | [cockroach.sql.contentionpb.ExtendedContentionEvent](#cockroach.server.serverpb.HistoricalContentionEventsResponse-cockroach.sql.contentionpb.ExtendedContentionEvent) | repeated |  | [reserved](#support-status) |
+
+
+
+
+
+
+
 ## RequestCA
 
 `GET /_join/v1/ca`

pkg/ccl/logictestccl/testdata/logic_test/crdb_internal_tenant

@@ -61,6 +61,7 @@ crdb_internal  gossip_alerts                    table  NULL  NULL  NULL
 crdb_internal  gossip_liveness                  table  NULL  NULL  NULL
 crdb_internal  gossip_network                   table  NULL  NULL  NULL
 crdb_internal  gossip_nodes                     table  NULL  NULL  NULL
+crdb_internal  historical_contention_events     table  NULL  NULL  NULL
 crdb_internal  index_columns                    table  NULL  NULL  NULL
 crdb_internal  index_usage_statistics           table  NULL  NULL  NULL
 crdb_internal  invalid_objects                  table  NULL  NULL  NULL

pkg/cli/testdata/zip/partial1

@@ -39,6 +39,7 @@ debug zip --concurrency=1 --cpu-profile-duration=0s /dev/null
 [cluster] retrieving SQL data for crdb_internal.invalid_objects... writing output: debug/crdb_internal.invalid_objects.txt... done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics... writing output: debug/crdb_internal.index_usage_statistics.txt... done
 [cluster] retrieving SQL data for crdb_internal.table_indexes... writing output: debug/crdb_internal.table_indexes.txt... done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events... writing output: debug/crdb_internal.historical_contention_events.txt... done
 [cluster] requesting nodes... received response... converting to JSON... writing binary output: debug/nodes.json... done
 [cluster] requesting liveness... received response... converting to JSON... writing binary output: debug/liveness.json... done
 [node 1] node status... converting to JSON... writing binary output: debug/nodes/1/status.json... done

pkg/cli/testdata/zip/partial1_excluded

@@ -39,6 +39,7 @@ debug zip /dev/null --concurrency=1 --exclude-nodes=2 --cpu-profile-duration=0
 [cluster] retrieving SQL data for crdb_internal.invalid_objects... writing output: debug/crdb_internal.invalid_objects.txt... done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics... writing output: debug/crdb_internal.index_usage_statistics.txt... done
 [cluster] retrieving SQL data for crdb_internal.table_indexes... writing output: debug/crdb_internal.table_indexes.txt... done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events... writing output: debug/crdb_internal.historical_contention_events.txt... done
 [cluster] requesting nodes... received response... converting to JSON... writing binary output: debug/nodes.json... done
 [cluster] requesting liveness... received response... converting to JSON... writing binary output: debug/liveness.json... done
 [node 1] node status... converting to JSON... writing binary output: debug/nodes/1/status.json... done

pkg/cli/testdata/zip/partial2

@@ -39,6 +39,7 @@ debug zip --concurrency=1 --cpu-profile-duration=0 /dev/null
 [cluster] retrieving SQL data for crdb_internal.invalid_objects... writing output: debug/crdb_internal.invalid_objects.txt... done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics... writing output: debug/crdb_internal.index_usage_statistics.txt... done
 [cluster] retrieving SQL data for crdb_internal.table_indexes... writing output: debug/crdb_internal.table_indexes.txt... done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events... writing output: debug/crdb_internal.historical_contention_events.txt... done
 [cluster] requesting nodes... received response... converting to JSON... writing binary output: debug/nodes.json... done
 [cluster] requesting liveness... received response... converting to JSON... writing binary output: debug/liveness.json... done
 [node 1] node status... converting to JSON... writing binary output: debug/nodes/1/status.json... done

pkg/cli/testdata/zip/testzip

@@ -39,6 +39,7 @@ debug zip --concurrency=1 --cpu-profile-duration=1s /dev/null
 [cluster] retrieving SQL data for crdb_internal.invalid_objects... writing output: debug/crdb_internal.invalid_objects.txt... done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics... writing output: debug/crdb_internal.index_usage_statistics.txt... done
 [cluster] retrieving SQL data for crdb_internal.table_indexes... writing output: debug/crdb_internal.table_indexes.txt... done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events... writing output: debug/crdb_internal.historical_contention_events.txt... done
 [cluster] requesting nodes... received response... converting to JSON... writing binary output: debug/nodes.json... done
 [cluster] requesting liveness... received response... converting to JSON... writing binary output: debug/liveness.json... done
 [cluster] requesting CPU profiles

pkg/cli/testdata/zip/testzip_concurrent

@@ -73,6 +73,9 @@ zip
 [cluster] retrieving SQL data for crdb_internal.default_privileges...
 [cluster] retrieving SQL data for crdb_internal.default_privileges: done
 [cluster] retrieving SQL data for crdb_internal.default_privileges: writing output: debug/crdb_internal.default_privileges.txt...
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events...
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events: done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events: writing output: debug/crdb_internal.historical_contention_events.txt...
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics...
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics: done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics: writing output: debug/crdb_internal.index_usage_statistics.txt...

pkg/cli/testdata/zip/testzip_tenant

@@ -53,6 +53,7 @@ debug zip --concurrency=1 /dev/null
 [cluster] retrieving SQL data for crdb_internal.invalid_objects... writing output: debug/crdb_internal.invalid_objects.txt... done
 [cluster] retrieving SQL data for crdb_internal.index_usage_statistics... writing output: debug/crdb_internal.index_usage_statistics.txt... done
 [cluster] retrieving SQL data for crdb_internal.table_indexes... writing output: debug/crdb_internal.table_indexes.txt... done
+[cluster] retrieving SQL data for crdb_internal.historical_contention_events... writing output: debug/crdb_internal.historical_contention_events.txt... done
 [cluster] requesting nodes... received response... converting to JSON... writing binary output: debug/nodes.json... done
 [cluster] requesting liveness... received response...
 [cluster] requesting liveness: last request failed: rpc error: ...

pkg/cli/zip_cluster_wide.go

@@ -110,6 +110,7 @@ var debugZipTablesPerCluster = []string{
 	"crdb_internal.invalid_objects",
 	"crdb_internal.index_usage_statistics",
 	"crdb_internal.table_indexes",
+	"crdb_internal.historical_contention_events",
 }
 
 // getNodesList constructs a NodesListResponse using the Nodes API. We need this while building

pkg/rpc/auth_tenant.go

@@ -107,6 +107,9 @@ func (a tenantAuthorizer) authorize(
 	case "/cockroach.server.serverpb.Status/CancelLocalQuery":
 		return a.authTenant(tenID)
 
+	case "/cockroach.server.serverpb.Status/HistoricalContentionEvents":
+		return a.authTenant(tenID)
+
 	case "/cockroach.roachpb.Internal/GetSpanConfigs":
 		return a.authGetSpanConfigs(tenID, req.(*roachpb.GetSpanConfigsRequest))
 

pkg/server/admin.go

@@ -3293,6 +3293,30 @@ func (c *adminPrivilegeChecker) requireViewActivityAndNoViewActivityRedactedPerm
 	return nil
 }
 
+func (c *adminPrivilegeChecker) hasViewActivityPermission(
+	ctx context.Context,
+) (hasViewActivityPermission bool, _ error) {
+	userName, isAdmin, err := c.getUserAndRole(ctx)
+	if err != nil {
+		return false /* hasViewActivityPermission */, serverError(ctx, err)
+	}
+
+	if isAdmin {
+		return true /* hasViewActivityPermission */, nil
+	}
+
+	hasViewActivity, err := c.hasRoleOption(ctx, userName, roleoption.VIEWACTIVITY)
+	if err != nil {
+		return false /* hasViewActivityPermission */, serverError(ctx, err)
+	}
+
+	if hasViewActivity {
+		return true /* hasViewActivityPermission */, nil
+	}
+
+	return false /* hasViewActivityPermission */, nil
+}
+
 // Note that the function returns plain errors, and it is the caller's
 // responsibility to convert them to serverErrors.
 func (c *adminPrivilegeChecker) getUserAndRole(

pkg/server/serverpb/status.go

@@ -37,6 +37,7 @@ type SQLStatusServer interface {
 	TableIndexStats(context.Context, *TableIndexStatsRequest) (*TableIndexStatsResponse, error)
 	UserSQLRoles(ctx context.Context, request *UserSQLRolesRequest) (*UserSQLRolesResponse, error)
 	TxnIDResolution(context.Context, *TxnIDResolutionRequest) (*TxnIDResolutionResponse, error)
+	HistoricalContentionEvents(context.Context, *HistoricalContentionEventsRequest) (*HistoricalContentionEventsResponse, error)
 }
 
 // OptionalNodesStatusServer is a StatusServer that is only optionally present

pkg/server/serverpb/status.proto

@@ -1509,6 +1509,16 @@ message TxnIDResolutionResponse {
     (gogoproto.nullable) = false];
 }
 
+message HistoricalContentionEventsRequest {
+  string node_id = 1 [(gogoproto.customname) = "NodeID"];
+}
+
+message HistoricalContentionEventsResponse {
+  repeated cockroach.sql.contentionpb.ExtendedContentionEvent events = 1 [
+    (gogoproto.nullable) = false
+  ];
+}
+
 service Status {
   // Certificates retrieves a copy of the TLS certificates.
   rpc Certificates(CertificatesRequest) returns (CertificatesResponse) {
@@ -1917,4 +1927,12 @@ service Status {
   //   Client is responsible to perform retries if the requested transaction ID
   //   is not returned in the RPC response.
   rpc TxnIDResolution(TxnIDResolutionRequest) returns (TxnIDResolutionResponse) {}
+
+  // HistoricalContentionEvents returns a list of un-aggregated contention
+  // events sorted by the collection timestamp.
+  rpc HistoricalContentionEvents(HistoricalContentionEventsRequest) returns (HistoricalContentionEventsResponse) {
+    option (google.api.http) = {
+      get: "/_status/historicalcontentionevents"
+    };
+  }
 }

pkg/server/status.go

@@ -401,6 +401,27 @@ func (b *baseStatusServer) localTxnIDResolution(
 	return resp
 }
 
+func (b *baseStatusServer) localHistoricalContentionEvents(
+	shouldRedactContendingKey bool,
+) *serverpb.HistoricalContentionEventsResponse {
+	registry := b.sqlServer.execCfg.ContentionRegistry
+
+	resp := &serverpb.HistoricalContentionEventsResponse{
+		Events: make([]contentionpb.ExtendedContentionEvent, 0),
+	}
+	// Ignore error returned by ForEachEvent() since if our own callback doesn't
+	// return error, ForEachEvent() also doesn't return error.
+	_ = registry.ForEachEvent(func(event *contentionpb.ExtendedContentionEvent) error {
+		if shouldRedactContendingKey {
+			event.BlockingEvent.Key = []byte{}
+		}
+		resp.Events = append(resp.Events, *event)
+		return nil
+	})
+
+	return resp
+}
+
 // A statusServer provides a RESTful status API.
 type statusServer struct {
 	*baseStatusServer
@@ -3087,3 +3108,76 @@ func (s *statusServer) TxnIDResolution(
 
 	return statusClient.TxnIDResolution(ctx, req)
 }
+
+func (s *statusServer) HistoricalContentionEvents(
+	ctx context.Context, req *serverpb.HistoricalContentionEventsRequest,
+) (*serverpb.HistoricalContentionEventsResponse, error) {
+	ctx = s.AnnotateCtx(propagateGatewayMetadata(ctx))
+
+	if err := s.privilegeChecker.requireViewActivityOrViewActivityRedactedPermission(ctx); err != nil {
+		return nil, err
+	}
+
+	hasViewActivity, err := s.privilegeChecker.hasViewActivityPermission(ctx)
+	if err != nil {
+		return nil, err
+	}
+	shouldRedactContendingKeys := !hasViewActivity
+
+	if s.gossip.NodeID.Get() == 0 {
+		return nil, status.Errorf(codes.Unavailable, "nodeID not set")
+	}
+
+	if len(req.NodeID) > 0 {
+		requestedNodeID, local, err := s.parseNodeID(req.NodeID)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, err.Error())
+		}
+		if local {
+			return s.localHistoricalContentionEvents(shouldRedactContendingKeys), nil
+		}
+
+		statusClient, err := s.dialNode(ctx, requestedNodeID)
+		if err != nil {
+			return nil, err
+		}
+		return statusClient.HistoricalContentionEvents(ctx, req)
+	}
+
+	dialFn := func(ctx context.Context, nodeID roachpb.NodeID) (interface{}, error) {
+		statusClient, err := s.dialNode(ctx, nodeID)
+		return statusClient, err
+	}
+
+	localReq := &serverpb.HistoricalContentionEventsRequest{
+		NodeID: "local",
+	}
+
+	rpcCallFn := func(ctx context.Context, client interface{}, _ roachpb.NodeID) (interface{}, error) {
+		statusClient := client.(serverpb.StatusClient)
+		return statusClient.HistoricalContentionEvents(ctx, localReq)
+	}
+
+	resp := &serverpb.HistoricalContentionEventsResponse{
+		Events: make([]contentionpb.ExtendedContentionEvent, 0),
+	}
+
+	if err := s.iterateNodes(ctx, fmt.Sprintf("historical contention events for node %s", req.NodeID),
+		dialFn,
+		rpcCallFn,
+		func(nodeID roachpb.NodeID, nodeResp interface{}) {
+			historicalContentionEvents := nodeResp.(*serverpb.HistoricalContentionEventsResponse)
+			resp.Events = append(resp.Events, historicalContentionEvents.Events...)
+		},
+		func(nodeID roachpb.NodeID, nodeFnError error) {
+		},
+	); err != nil {
+		return nil, err
+	}
+
+	sort.Slice(resp.Events, func(i, j int) bool {
+		return resp.Events[i].CollectionTs.Before(resp.Events[j].CollectionTs)
+	})
+
+	return resp, nil
+}