obsservice: export StatementInsightsStatistics

pkg/obs/event_exporter.go

@@ -220,6 +220,12 @@ func NewEventsExporter(
 				Version: "1.0",
 			},
 		},
+		obspb.StatementInsightsStatsEvent: {
+			instrumentationScope: otel_pb.InstrumentationScope{
+				Name:    string(obspb.StatementInsightsStatsEvent),
+				Version: "1.0",
+			},
+		},
 	}
 	s.buf.mu.memAccount = memMonitor.MakeBoundAccount()
 	return s

pkg/obsservice/cmd/obsservice/main.go

@@ -85,7 +85,8 @@ from one or more CockroachDB clusters.`,
 
 		// Run the event ingestion in the background.
 		eventRouter := router.NewEventRouter(map[obspb.EventType]obslib.EventConsumer{
-			obspb.EventlogEvent: &obsutil.StdOutConsumer{},
+			obspb.EventlogEvent:               &obsutil.StdOutConsumer{},
+			obspb.StatementInsightsStatsEvent: &obsutil.StdOutConsumer{},
 		})
 		ingester := ingest.MakeEventIngester(ctx, eventRouter, nil)
 

pkg/obsservice/obspb/BUILD.bazel

@@ -23,6 +23,8 @@ proto_library(
         "//pkg/obsservice/obspb/opentelemetry-proto/logs/v1:v1_proto",
         "//pkg/obsservice/obspb/opentelemetry-proto/resource/v1:v1_proto",
         "@com_github_gogo_protobuf//gogoproto:gogo_proto",
+        "@com_google_protobuf//:duration_proto",
+        "@com_google_protobuf//:timestamp_proto",
     ],
 )
 

pkg/obsservice/obspb/event_types.go

@@ -14,7 +14,8 @@ type EventType string
 const (
 	// EventlogEvent represents general events about the cluster that historically
 	// have been persisted inside CRDB in the system.eventlog table.
-	EventlogEvent EventType = "eventlog"
+	EventlogEvent               EventType = "eventlog"
+	StatementInsightsStatsEvent EventType = "stmtinsightsstats"
 )
 
 // EventlogEventTypeAttribute represents the key of the attribute containing

pkg/obsservice/obspb/obsservice.proto

@@ -11,7 +11,8 @@ syntax = "proto3";
 package cockroach.obspb;
 
 import "gogoproto/gogo.proto";
-
+import "google/protobuf/duration.proto";
+import "google/protobuf/timestamp.proto";
 import "obsservice/obspb/opentelemetry-proto/logs/v1/logs.proto";
 import "obsservice/obspb/opentelemetry-proto/common/v1/common.proto";
 import "obsservice/obspb/opentelemetry-proto/resource/v1/resource.proto";
@@ -36,3 +37,107 @@ message Event {
   // The LogRecord containing the specific event information.
   opentelemetry.proto.logs.v1.LogRecord log_record = 3;
 }
+
+message StatementInsightsStatistics {
+  enum Status {
+    Completed = 0;
+    Failed = 1;
+  }
+
+  enum Problem {
+    // This statement execution was not itself problematic, but it happened in
+    // the same transaction as another problematic statement.
+    None = 0;
+
+    // This statement execution was slow, either above a fixed latency threshold
+    // or detected as an anomaly.
+    SlowExecution = 1;
+
+    // This statement execution failed completely, due to contention, resource
+    // saturation, or syntax errors.
+    FailedExecution = 2;
+  }
+
+  enum Cause {
+    // We have not identified specific reasons why this statement was slow.
+    // We will not explicitly use this zero value; it is here defensively.
+    Unset = 0;
+
+    // This statement was slow because we picked the wrong plan, possibly due to
+    // outdated statistics, the statement using different literals or search
+    // conditions, or a change in the database schema.
+    PlanRegression = 1;
+
+    // This statement was slow because a good plan was not available, whether
+    // due to outdated statistics or missing indexes.
+    SuboptimalPlan = 2;
+
+    // This statement was slow because of contention.
+    HighContention = 3;
+
+    // This statement was slow because of being retried multiple times, again due
+    // to contention. The "high" threshold may be configured by the
+    // `sql.insights.high_retry_count.threshold` cluster setting.
+    HighRetryCount = 4;
+  }
+
+  bytes id = 1 [(gogoproto.customname) = "ID"];
+  uint64 fingerprint_id = 2 [(gogoproto.customname) = "FingerprintID"];
+  bytes transaction_id = 3 [(gogoproto.customname) = "TransactionID"];
+  uint64 txn_fingerprint_id = 4 [(gogoproto.customname) = "TxnFingerprintID"];
+  bytes session_id = 5 [(gogoproto.customname) = "SessionID"];
+  string query = 6;
+  Status status = 7;
+  google.protobuf.Timestamp start_time = 8 [(gogoproto.stdtime) = true];
+  google.protobuf.Timestamp end_time = 9 [(gogoproto.stdtime) = true];
+  bool full_scan = 10;
+  bool implicit_txn = 11;
+  string user = 12;
+  string user_priority = 13;
+  string application_name = 14;
+  string database = 15;
+  string plan_gist = 16;
+  int64 rows_read = 17;
+  int64 rows_written = 18;
+  int64 retries = 19;
+  string auto_retry_reason = 20;
+  // Nodes is the ordered list of nodes ids on which the statement was executed.
+  repeated int64 nodes = 21;
+  google.protobuf.Duration contention = 22 [(gogoproto.stdduration) = true];
+  repeated string index_recommendations = 23;
+  Problem problem = 24;
+  repeated Cause causes = 25;
+  int64 cpu_sql_nanos = 26 [(gogoproto.customname) = "CPUSQLNanos"];
+  string error_code = 27;
+  // IdleLatSeconds is the time spent in seconds in an open transaction waiting
+  // for the client to send the statement.
+  double idle_lat_seconds = 28;
+  // ParseLat is the time in seconds to transform the SQL string into an AST.
+  double parse_lat_seconds = 29;
+  // PlanLatSeconds is the time spent in seconds to transform the AST into a logical query plan.
+  double plan_lat_seconds = 30;
+  // RunLatSeconds is the time in seconds to run the query and fetch/compute the result rows.
+  double run_lat_seconds = 31;
+  // ServiceLatSeconds is the time in seconds to service the query, from start of parse to end of execute.
+  double service_lat_seconds = 32;
+  repeated ContentionEvent contention_events = 33 [(gogoproto.nullable) = true];
+}
+
+message ContentionEvent {
+  // Key is the key that this and the other transaction conflicted on.
+  bytes key = 1;
+  string pretty_key = 2;
+  bytes id = 3 [(gogoproto.customname) = "BlockingTxnID"];
+  uint64 blocking_txn_fingerprint_id = 4 [(gogoproto.customname) = "BlockingTxnFingerprintID"];
+  // Duration spent contending against the other transaction.
+  google.protobuf.Duration duration = 5 [(gogoproto.stdduration) = true];
+  bytes waiting_txn_id = 6 [(gogoproto.customname) = "WaitingTxnID"];
+  uint64 waiting_txn_fingerprint_id = 7 [(gogoproto.customname) = "WaitingTxnFingerprintID"];
+  google.protobuf.Timestamp collection_ts = 8 [(gogoproto.stdtime) = true];
+  uint64 waiting_stmt_fingerprint_id = 9 [(gogoproto.customname) = "WaitingStmtFingerprintID"];
+  bytes waiting_stmt_id = 10 [(gogoproto.customname) = "WaitingStmtID"];
+  string database_name = 11;
+  string schema_name = 12;
+  string index_name = 13;
+  string table_name = 14;
+}

pkg/sql/sqlstats/insights/BUILD.bazel

@@ -19,13 +19,15 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/sqlstats/insights",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/obsservice/obspb",
         "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/sql/appstatspb",
         "//pkg/sql/clusterunique",
         "//pkg/sql/contention/contentionutils",
         "//pkg/util/cache",
         "//pkg/util/intsets",
+        "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/quantile",
         "//pkg/util/stop",
@@ -48,14 +50,19 @@ go_test(
         "store_test.go",
     ],
     args = ["-test.timeout=295s"],
+    data = glob(["testdata/**"]),
     embed = [":insights"],
     deps = [
+        "//pkg/obsservice/obspb",
         "//pkg/settings/cluster",
         "//pkg/sql/appstatspb",
         "//pkg/sql/clusterunique",
+        "//pkg/util/leaktest",
         "//pkg/util/stop",
         "//pkg/util/uint128",
         "//pkg/util/uuid",
+        "@com_github_cockroachdb_datadriven//:datadriven",
+        "@com_github_kr_pretty//:pretty",
         "@com_github_stretchr_testify//require",
     ],
 )

pkg/sql/sqlstats/insights/registry.go

@@ -11,11 +11,14 @@
 package insights
 
 import (
+	"context"
 	"sync"
 
+	"github.com/cockroachdb/cockroach/pkg/obsservice/obspb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/clusterunique"
 	"github.com/cockroachdb/cockroach/pkg/util/intsets"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/redact"
 )
 
@@ -192,3 +195,85 @@ func newRegistry(st *cluster.Settings, detector detector, sink sink) *lockingReg
 		sink:       sink,
 	}
 }
+
+func (s *Statement) CopyTo(
+	ctx context.Context, t *Transaction, session *Session, other *obspb.StatementInsightsStatistics,
+) {
+	other.ApplicationName = t.ApplicationName
+	other.AutoRetryReason = s.AutoRetryReason
+	other.Contention = s.Contention
+	other.CPUSQLNanos = s.CPUSQLNanos
+	other.Database = s.Database
+	other.EndTime = &s.EndTime
+	other.ErrorCode = s.ErrorCode
+	other.FingerprintID = uint64(s.FingerprintID)
+	other.FullScan = s.FullScan
+	other.ImplicitTxn = t.ImplicitTxn
+	other.IndexRecommendations = s.IndexRecommendations
+	other.Nodes = s.Nodes
+	other.PlanGist = s.PlanGist
+	other.Query = s.Query
+	other.Retries = s.Retries
+	other.RowsRead = s.RowsRead
+	other.RowsWritten = s.RowsWritten
+	other.ServiceLatSeconds = s.LatencyInSeconds
+	other.StartTime = &s.StartTime
+	other.TxnFingerprintID = uint64(t.FingerprintID)
+	other.User = t.User
+	other.UserPriority = t.UserPriority
+
+	var err error
+	other.ID, err = s.ID.MarshalJSON()
+	if err != nil {
+		log.Errorf(ctx, "marshalling statement insights ID for Insights exporter")
+	}
+	other.TransactionID, err = t.ID.MarshalJSON()
+	if err != nil {
+		log.Errorf(ctx, "marshalling transaction insights ID for Insights exporter")
+	}
+	other.SessionID, err = session.ID.MarshalJSON()
+	if err != nil {
+		log.Errorf(ctx, "marshalling sessions ID for Insights exporter")
+	}
+
+	switch s.Status {
+	case Statement_Completed:
+		other.Status = obspb.StatementInsightsStatistics_Completed
+	case Statement_Failed:
+		other.Status = obspb.StatementInsightsStatistics_Failed
+	default:
+		other.Status = obspb.StatementInsightsStatistics_Completed
+	}
+
+	switch s.Problem {
+	case Problem_FailedExecution:
+		other.Problem = obspb.StatementInsightsStatistics_FailedExecution
+	case Problem_SlowExecution:
+		other.Problem = obspb.StatementInsightsStatistics_SlowExecution
+	default:
+		other.Problem = obspb.StatementInsightsStatistics_None
+	}
+
+	other.Causes = []obspb.StatementInsightsStatistics_Cause{}
+	for c := range s.Causes {
+		switch int32(c) {
+		case Cause_value["SuboptimalPlan"]:
+			other.Causes = append(other.Causes, obspb.StatementInsightsStatistics_SuboptimalPlan)
+		case Cause_value["HighRetryCount"]:
+			other.Causes = append(other.Causes, obspb.StatementInsightsStatistics_HighRetryCount)
+		case Cause_value["PlanRegression"]:
+			other.Causes = append(other.Causes, obspb.StatementInsightsStatistics_PlanRegression)
+		case Cause_value["HighContention"]:
+			other.Causes = append(other.Causes, obspb.StatementInsightsStatistics_HighContention)
+		default:
+			other.Causes = append(other.Causes, obspb.StatementInsightsStatistics_Unset)
+		}
+	}
+
+	// TODO(maryliag): add information about Contention Events
+	// and Idle/Parse/Run Latencies.
+	//other.ContentionEvents
+	//other.IdleLatSeconds
+	//other.ParseLatSeconds
+	//other.RunLatSeconds
+}

pkg/sql/sqlstats/insights/registry_test.go

@@ -13,14 +13,20 @@ package insights
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"sort"
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/pkg/obsservice/obspb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/appstatspb"
 	"github.com/cockroachdb/cockroach/pkg/sql/clusterunique"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
+	"github.com/cockroachdb/cockroach/pkg/util/uint128"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/datadriven"
+	"github.com/kr/pretty"
 	"github.com/stretchr/testify/require"
 )
 
@@ -377,3 +383,54 @@ func TestRegistry(t *testing.T) {
 		require.Equal(t, transaction.Status, Transaction_Status(statementNotIgnored.Status))
 	})
 }
+
+func TestInsightsConversion(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	ctx := context.Background()
+	session := Session{ID: clusterunique.IDFromBytes([]byte("aaaaaaaaaaaaaaaaaaaaaaaaaaa"))}
+	contentionDuration := 10 * time.Second
+
+	// Construct by hand an Insight struct. The values don't matter, but the same values
+	// being included in the transformation result do. This will fail whenever someone makes a change to
+	// obspb.StatementInsightsStatistics, forcing folks to update the transformation logic accordingly.
+	stmt := Statement{
+		AutoRetryReason:      "myRetryReason",
+		Causes:               []Cause{Cause_HighContention, Cause_SuboptimalPlan},
+		Contention:           &contentionDuration,
+		CPUSQLNanos:          500,
+		Database:             "myDB",
+		EndTime:              time.Date(2023, time.October, 31, 18, 33, 39, 0, time.UTC),
+		ErrorCode:            "myErrorCode",
+		FingerprintID:        12345,
+		FullScan:             true,
+		ID:                   clusterunique.ID{Uint128: uint128.Uint128{Lo: 12, Hi: 987}},
+		IndexRecommendations: []string{"rec1", "rec2"},
+		Nodes:                []int64{2, 4, 8},
+		PlanGist:             "myPlanGist",
+		Problem:              Problem_SlowExecution,
+		Query:                "myQuery",
+		Retries:              2,
+		RowsRead:             100,
+		RowsWritten:          2,
+		LatencyInSeconds:     2,
+		StartTime:            time.Date(2023, time.October, 31, 18, 31, 39, 0, time.UTC),
+		Status:               Statement_Completed,
+	}
+	txn := Transaction{
+		ApplicationName: "myApp",
+		FingerprintID:   appstatspb.TransactionFingerprintID(123),
+		ID:              uuid.UUID{2},
+		ImplicitTxn:     true,
+		User:            "myUser",
+		UserPriority:    "1",
+	}
+
+	datadriven.RunTest(t, "testdata/collectedstmtinsightsstats_transform", func(t *testing.T, d *datadriven.TestData) string {
+		res := new(obspb.StatementInsightsStatistics)
+		stmt.CopyTo(ctx, &txn, &session, res)
+		var buf bytes.Buffer
+		_, err := fmt.Fprintf(&buf, "%# v\n", pretty.Formatter(res))
+		require.NoError(t, err)
+		return buf.String()
+	})
+}