sql: prep migration for sampled_query and sampled_transaction events

In v26.1, sampled_query and sampled_transaction events will be moved
from the TELEMETRY logging channel to the SQL_EXEC logging channel.

This commit gates this migration under the cluster setting:
`log.channel_compatibility_mode.enabled` and will log these events
to the SQL_EXEC channel if this setting is set to false. Users can
set this setting to false in their clusters to validate, test, and
identify potential downstream impacts to their logging setups and
pipelines.

Epic: CRDB-53410
Part of: CRDB-53412
Release note (ops change): sampled_query and sampled_transaction
events will be moved to the SQL_EXEC channel in 26.1. In order to
test the impact of these changes, users can set the setting:
`log.channel_compatibility_mode.enabled` to false. Note that this
will cause these logs to start logging in the SQL_EXEC channel so
this shouldn't be tested in a production environment.

Author: kyle-a-wong

docs/generated/eventlog.md

@@ -3364,6 +3364,11 @@ Fields in this struct should be updated in sync with apps_stats.proto.
 An event of type `sampled_query` is the SQL query event logged to the telemetry channel. It
 contains common SQL event/execution details.
 
+Note: in version 26.1, these events will be moved to the `SQL_EXEC` channel.
+To test compatability before this, set the cluster setting
+`log.channel_compatibility_mode.enabled` to false. This will send the
+events to `SQL_EXEC` instead of `TELEMETRY`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|
@@ -3472,6 +3477,11 @@ contains common SQL event/execution details.
 
 An event of type `sampled_transaction` is the event logged to telemetry at the end of transaction execution.
 
+Note: in version 26.1, these events will be moved to the `SQL_EXEC` channel.
+To test compatability before this, set the cluster setting
+`log.channel_compatibility_mode.enabled` to false. This will send the
+events to `SQL_EXEC` instead of `TELEMETRY`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|

pkg/ccl/telemetryccl/BUILD.bazel

@@ -20,6 +20,7 @@ go_test(
         "//pkg/security/securityassets",
         "//pkg/security/securitytest",
         "//pkg/server",
+        "//pkg/settings",
         "//pkg/sql",
         "//pkg/sql/sem/tree",
         "//pkg/sql/sqltestutils",

pkg/ccl/telemetryccl/telemetry_logging_test.go

@@ -13,6 +13,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"regexp"
+	"slices"
 	"strings"
 	"testing"
 
@@ -21,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/cloud/nodelocal"
 	"github.com/cockroachdb/cockroach/pkg/jobs"
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
+	"github.com/cockroachdb/cockroach/pkg/settings"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -155,20 +157,29 @@ type expectedSampleQueryEvent struct {
 }
 
 type telemetrySpy struct {
-	t *testing.T
+	t  *testing.T
+	sv *settings.Values
 
 	sampledQueries    []eventpb.SampledQuery
 	sampledQueriesRaw []logpb.Entry
 	recoveryEvents    []eventpb.RecoveryEvent
 }
 
+func (l *telemetrySpy) channelsToIntercept() []log.Channel {
+	if log.ShouldMigrateEvent(l.sv) {
+		return []log.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC}
+	}
+
+	return []log.Channel{logpb.Channel_TELEMETRY}
+}
+
 func (l *telemetrySpy) Intercept(entry []byte) {
 	var rawLog logpb.Entry
 	if err := json.Unmarshal(entry, &rawLog); err != nil {
 		l.t.Errorf("failed unmarshaling %s: %s", entry, err)
 	}
 
-	if rawLog.Channel != logpb.Channel_TELEMETRY {
+	if !slices.Contains(l.channelsToIntercept(), rawLog.Channel) {
 		return
 	}
 
@@ -204,12 +215,6 @@ func TestBulkJobTelemetryLogging(t *testing.T) {
 
 	ctx := context.Background()
 
-	spy := &telemetrySpy{
-		t: t,
-	}
-	cleanup := log.InterceptWith(ctx, spy)
-	defer cleanup()
-
 	st := logtestutils.StubTime{}
 	sqm := logtestutils.StubQueryStats{}
 	sts := logtestutils.StubTracingStatus{}
@@ -229,6 +234,15 @@ func TestBulkJobTelemetryLogging(t *testing.T) {
 			ExternalIODir: dir,
 		},
 	})
+
+	spy := &telemetrySpy{
+		t:  t,
+		sv: &testCluster.Server(0).ClusterSettings().SV,
+	}
+
+	cleanup := log.InterceptWith(ctx, spy)
+	defer cleanup()
+
 	sqlDB := testCluster.ServerConn(0)
 	defer func() {
 		testCluster.Stopper().Stop(context.Background())

pkg/sql/exec_log.go

@@ -358,6 +358,7 @@ func (p *planner) maybeLogStatementInternal(
 
 		*sampledQuery = eventpb.SampledQuery{
 			CommonSQLExecDetails:     execDetails,
+			CommonSQLEventDetails:    p.getCommonSQLEventDetails(),
 			SkippedQueries:           skippedQueries,
 			CostEstimate:             p.curPlan.instrumentation.costEstimate,
 			Distribution:             p.curPlan.instrumentation.distribution.String(),
@@ -436,7 +437,11 @@ func (p *planner) maybeLogStatementInternal(
 			SchemaChangerMode:                     p.curPlan.instrumentation.schemaChangerMode.String(),
 		}
 
-		p.logEventsOnlyExternally(ctx, sampledQuery)
+		migrator := log.NewStructuredEventMigrator(func() bool {
+			return log.ShouldMigrateEvent(p.ExecCfg().SV())
+		}, logpb.Channel_SQL_EXEC)
+
+		migrator.StructuredEvent(ctx, severity.INFO, sampledQuery)
 	}
 }
 
@@ -521,7 +526,11 @@ func (p *planner) logTransaction(
 		}
 	}
 
-	log.StructuredEvent(ctx, severity.INFO, sampledTxn)
+	migrator := log.NewStructuredEventMigrator(func() bool {
+		return log.ShouldMigrateEvent(p.ExecCfg().SV())
+	}, logpb.Channel_SQL_EXEC)
+
+	migrator.StructuredEvent(ctx, severity.INFO, sampledTxn)
 }
 
 func (p *planner) logEventsOnlyExternally(ctx context.Context, entries ...logpb.EventPayload) {

pkg/sql/telemetry_datadriven_test.go

@@ -60,13 +60,12 @@ func TestTelemetryLoggingDataDriven(t *testing.T) {
 
 	sc := log.Scope(t)
 	defer sc.Close(t)
-
 	appName := "telemetry-logging-datadriven"
 	ignoredAppname := "telemetry-datadriven-ignored-appname"
 	ctx := context.Background()
 	stmtSpy := logtestutils.NewStructuredLogSpy(
 		t,
-		[]logpb.Channel{logpb.Channel_TELEMETRY},
+		[]logpb.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC},
 		[]string{"sampled_query"},
 		logtestutils.FormatEntryAsJSON,
 		func(_ logpb.Entry, logStr string) bool {
@@ -79,7 +78,7 @@ func TestTelemetryLoggingDataDriven(t *testing.T) {
 
 	txnsSpy := logtestutils.NewStructuredLogSpy(
 		t,
-		[]logpb.Channel{logpb.Channel_TELEMETRY},
+		[]logpb.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC},
 		[]string{"sampled_transaction"},
 		logtestutils.FormatEntryAsJSON,
 		func(_ logpb.Entry, logStr string) bool {
@@ -209,13 +208,13 @@ func TestTelemetryLoggingDataDriven(t *testing.T) {
 				}
 
 				newStmtLogCount := stmtSpy.Count()
-				sb.WriteString(strings.Join(stmtSpy.GetLastNLogs(logpb.Channel_TELEMETRY, newStmtLogCount-stmtLogCount), "\n"))
+				sb.WriteString(strings.Join(stmtSpy.GetLastNLogs(getSampleQueryLoggingChannel(&s.ClusterSettings().SV), newStmtLogCount-stmtLogCount), "\n"))
 				if newStmtLogCount > stmtLogCount {
 					sb.WriteString("\n")
 				}
 
 				newTxnLogCount := txnsSpy.Count()
-				sb.WriteString(strings.Join(txnsSpy.GetLastNLogs(logpb.Channel_TELEMETRY, newTxnLogCount-txnLogCount), "\n"))
+				sb.WriteString(strings.Join(txnsSpy.GetLastNLogs(getSampleQueryLoggingChannel(&s.ClusterSettings().SV), newTxnLogCount-txnLogCount), "\n"))
 				return sb.String()
 			case "reset-last-sampled":
 				telemetryLogging.resetLastSampledTime()