log: finalize structured event logging channel migration

jasonlmfong · jasonlmfong · commit 1ae7b24a7204 · 2025-10-07T13:21:23.000-04:00
The following structured events should have their logging channels migrated: CreateChangefeed: TELEMETRY -> CHANGEFEED AlterChangefeed: TELEMETRY -> CHANGEFEED ChangefeedFailed: TELEMETRY -> CHANGEFEED ChangefeedCanceled: TELEMETRY -> CHANGEFEED ChangefeedEmittedBytes: TELEMETRY -> CHANGEFEED SampledQuery: TELEMETRY -> SQL_EXEC SampledTransaction: TELEMETRY -> SQL_EXEC SlowQuery: SQL_PERF -> SQL_EXEC LargeRow: SQL_PERF -> SQL_EXEC TxnRowsWrittenLimit: SQL_PERF -> SQL_EXEC TxnRowsReadLimit: SQL_PERF -> SQL_EXEC SlowQueryInternal: SQL_INTERNAL_PERF -> SQL_EXEC LargeRowInternal: SQL_INTERNAL_PERF -> SQL_EXEC TxnRowsWrittenLimitInternal: SQL_INTERNAL_PERF -> SQL_EXEC TxnRowsReadLimitInternal: SQL_INTERNAL_PERF -> SQL_EXEC This is achieved by setting default of `log.channel_compatibility_mode.enabled` to false This change also updates the protobuff documentation Epic: CRDB-53410 Release note (ops change): This change completes the migration of changfeed events, sample query/txn events, sql_perf events into their respective new logging channels. In order to keep using the old channels, users have to set the setting: `log.channel_compatibility_mode.enabled` to true.
diff --git a/docs/generated/eventlog.md b/docs/generated/eventlog.md
diff --git a/docs/generated/settings/settings-for-tenants.txt b/docs/generated/settings/settings-for-tenants.txt
@@ -106,7 +106,7 @@ kv.transaction.write_pipelining.locking_reads.enabled	boolean	true	if enabled, t
 kv.transaction.write_pipelining.ranged_writes.enabled	boolean	true	if enabled, transactional ranged writes are pipelined through Raft consensus	application
 kv.transaction.write_pipelining.enabled (alias: kv.transaction.write_pipelining_enabled)	boolean	true	if enabled, transactional writes are pipelined through Raft consensus	application
 kv.transaction.write_pipelining.max_batch_size (alias: kv.transaction.write_pipelining_max_batch_size)	integer	128	if non-zero, defines that maximum size batch that will be pipelined through Raft consensus	application
-log.channel_compatibility_mode.enabled	boolean	true	when true, logs will continue to log to the expected logging channels; when false, logs will be moved to new logging channels as part of a logging channel consolidation effort	application
+log.channel_compatibility_mode.enabled	boolean	false	when true, logs will to log to their legacy (pre 26.1) logging channels; when false, logs will be logged to new logging channels	application
 obs.tablemetadata.automatic_updates.enabled	boolean	false	enables automatic updates of the table metadata cache system.table_metadata	application
 obs.tablemetadata.data_valid_duration	duration	20m0s	the duration for which the data in system.table_metadata is considered valid	application
 schedules.backup.gc_protection.enabled	boolean	true	enable chaining of GC protection across backups run as part of a schedule	application
diff --git a/docs/generated/settings/settings.html b/docs/generated/settings/settings.html
@@ -137,7 +137,7 @@
 <tr><td><div id="setting-kv-transaction-write-pipelining-max-batch-size" class="anchored"><code>kv.transaction.write_pipelining.max_batch_size<br />(alias: kv.transaction.write_pipelining_max_batch_size)</code></div></td><td>integer</td><td><code>128</code></td><td>if non-zero, defines that maximum size batch that will be pipelined through Raft consensus</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-kvadmission-store-provisioned-bandwidth" class="anchored"><code>kvadmission.store.provisioned_bandwidth</code></div></td><td>byte size</td><td><code>0 B</code></td><td>if set to a non-zero value, this is used as the provisioned bandwidth (in bytes/s), for each store. It can be overridden on a per-store basis using the --store flag. Note that setting the provisioned bandwidth to a positive value may enable disk bandwidth based admission control, since admission.disk_bandwidth_tokens.elastic.enabled defaults to true</td><td>Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-kvadmission-store-snapshot-ingest-bandwidth-control-enabled" class="anchored"><code>kvadmission.store.snapshot_ingest_bandwidth_control.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>if set to true, snapshot ingests will be subject to disk write control in AC</td><td>Advanced/Self-Hosted</td></tr>
-<tr><td><div id="setting-log-channel-compatibility-mode-enabled" class="anchored"><code>log.channel_compatibility_mode.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>when true, logs will continue to log to the expected logging channels; when false, logs will be moved to new logging channels as part of a logging channel consolidation effort</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
+<tr><td><div id="setting-log-channel-compatibility-mode-enabled" class="anchored"><code>log.channel_compatibility_mode.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when true, logs will to log to their legacy (pre 26.1) logging channels; when false, logs will be logged to new logging channels</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-obs-tablemetadata-automatic-updates-enabled" class="anchored"><code>obs.tablemetadata.automatic_updates.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>enables automatic updates of the table metadata cache system.table_metadata</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-obs-tablemetadata-data-valid-duration" class="anchored"><code>obs.tablemetadata.data_valid_duration</code></div></td><td>duration</td><td><code>20m0s</code></td><td>the duration for which the data in system.table_metadata is considered valid</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
 <tr><td><div id="setting-schedules-backup-gc-protection-enabled" class="anchored"><code>schedules.backup.gc_protection.enabled</code></div></td><td>boolean</td><td><code>true</code></td><td>enable chaining of GC protection across backups run as part of a schedule</td><td>Basic/Standard/Advanced/Self-Hosted</td></tr>
diff --git a/pkg/ccl/changefeedccl/changefeed_stmt.go b/pkg/ccl/changefeedccl/changefeed_stmt.go
@@ -2317,7 +2317,7 @@ func getChangefeedEventMigrator(migrateEvent bool) log.StructuredEventMigrator {
 		func() bool {
 			return migrateEvent
 		},
-		channel.CHANGEFEED,
+		channel.TELEMETRY,
 	)
 }
 
diff --git a/pkg/ccl/changefeedccl/changefeed_test.go b/pkg/ccl/changefeedccl/changefeed_test.go
@@ -12742,9 +12742,9 @@ func TestTableRenameDuringDatabaseLevelChangefeed(t *testing.T) {
 
 func getChangefeedLoggingChannel(sv *settings.Values) logpb.Channel {
 	if log.ShouldMigrateEvent(sv) {
-		return logpb.Channel_CHANGEFEED
+		return logpb.Channel_TELEMETRY
 	}
-	return logpb.Channel_TELEMETRY
+	return logpb.Channel_CHANGEFEED
 }
 
 func TestCreateTableLevelChangefeedWithDBPrivilege(t *testing.T) {
diff --git a/pkg/ccl/telemetryccl/telemetry_logging_test.go b/pkg/ccl/telemetryccl/telemetry_logging_test.go
@@ -167,10 +167,10 @@ type telemetrySpy struct {
 
 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}
 	}
 
-	return []log.Channel{logpb.Channel_TELEMETRY}
+	return []log.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC}
 }
 
 func (l *telemetrySpy) Intercept(entry []byte) {
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
@@ -3123,7 +3123,7 @@ func (ex *connExecutor) handleTxnRowsGuardrails(
 			}
 			migrator := log.NewStructuredEventMigrator(func() bool {
 				return log.ShouldMigrateEvent(ex.planner.ExecCfg().SV())
-			}, logpb.Channel_SQL_EXEC)
+			}, logpb.Channel_SQL_PERF)
 			migrator.StructuredEvent(ctx, severity.INFO, event)
 			logCounter.Inc(1)
 		}
diff --git a/pkg/sql/event_log_test.go b/pkg/sql/event_log_test.go
@@ -765,9 +765,9 @@ func TestPerfLogging(t *testing.T) {
 			db.Exec(t, tc.query)
 		}
 
-		expectedChannel := tc.channel
-		if !log.ChannelCompatibilityModeEnabled.Get(&s.ClusterSettings().SV) {
-			expectedChannel = logpb.Channel_SQL_EXEC
+		expectedChannel := logpb.Channel_SQL_EXEC
+		if log.ShouldMigrateEvent(&s.ClusterSettings().SV) {
+			expectedChannel = tc.channel
 		}
 
 		var logRe = regexp.MustCompile(tc.logRe)
diff --git a/pkg/sql/exec_log.go b/pkg/sql/exec_log.go
@@ -252,16 +252,16 @@ func (p *planner) maybeLogStatementInternal(
 		// Is the query actually slow?
 		queryDuration > slowLogThreshold) {
 		commonSQLEventDetails := p.getCommonSQLEventDetails()
-		migrator := log.NewStructuredEventMigrator(func() bool {
-			return !log.ChannelCompatibilityModeEnabled.Get(p.ExecCfg().SV())
-		}, logpb.Channel_SQL_EXEC)
 		switch {
 		case execType == executorTypeExec:
 			// Non-internal queries are always logged to the slow query log.
 			event := &eventpb.SlowQuery{
 				CommonSQLEventDetails: commonSQLEventDetails,
 				CommonSQLExecDetails:  execDetails,
 			}
+			migrator := log.NewStructuredEventMigrator(func() bool {
+				return log.ShouldMigrateEvent(p.ExecCfg().SV())
+			}, logpb.Channel_SQL_PERF)
 			migrator.StructuredEvent(ctx, severity.INFO, event)
 		case execType == executorTypeInternal && slowInternalQueryLogEnabled:
 			// Internal queries that surpass the slow query log threshold should only
@@ -270,6 +270,9 @@ func (p *planner) maybeLogStatementInternal(
 				CommonSQLEventDetails: commonSQLEventDetails,
 				CommonSQLExecDetails:  execDetails,
 			}
+			migrator := log.NewStructuredEventMigrator(func() bool {
+				return log.ShouldMigrateEvent(p.ExecCfg().SV())
+			}, logpb.Channel_SQL_INTERNAL_PERF)
 			migrator.StructuredEvent(ctx, severity.INFO,
 				event)
 		}
@@ -435,7 +438,7 @@ func (p *planner) maybeLogStatementInternal(
 
 		migrator := log.NewStructuredEventMigrator(func() bool {
 			return log.ShouldMigrateEvent(p.ExecCfg().SV())
-		}, logpb.Channel_SQL_EXEC)
+		}, logpb.Channel_TELEMETRY)
 
 		migrator.StructuredEvent(ctx, severity.INFO, sampledQuery)
 	}
@@ -524,7 +527,7 @@ func (p *planner) logTransaction(
 
 	migrator := log.NewStructuredEventMigrator(func() bool {
 		return log.ShouldMigrateEvent(p.ExecCfg().SV())
-	}, logpb.Channel_SQL_EXEC)
+	}, logpb.Channel_TELEMETRY)
 
 	migrator.StructuredEvent(ctx, severity.INFO, sampledTxn)
 }
diff --git a/pkg/sql/row/helper.go b/pkg/sql/row/helper.go
@@ -477,7 +477,7 @@ func (rh *RowHelper) CheckRowSize(
 		}
 		migrator := log.NewStructuredEventMigrator(func() bool {
 			return rh.migrateLargeRowLog
-		}, logpb.Channel_SQL_EXEC)
+		}, logpb.Channel_SQL_PERF)
 		migrator.StructuredEvent(ctx, severity.INFO, event)
 	}
 	if shouldErr {
diff --git a/pkg/sql/telemetry_logging_test.go b/pkg/sql/telemetry_logging_test.go
@@ -1630,7 +1630,7 @@ func TestTelemetryLoggingStmtPosInTxn(t *testing.T) {
 
 func getSampleQueryLoggingChannel(sv *settings.Values) logpb.Channel {
 	if log.ShouldMigrateEvent(sv) {
-		return logpb.Channel_SQL_EXEC
+		return logpb.Channel_TELEMETRY
 	}
-	return logpb.Channel_TELEMETRY
+	return logpb.Channel_SQL_EXEC
 }
diff --git a/pkg/util/log/eventpb/changefeed_events.proto b/pkg/util/log/eventpb/changefeed_events.proto
@@ -10,23 +10,19 @@ option go_package = "github.com/cockroachdb/cockroach/pkg/util/log/eventpb";
 import "gogoproto/gogo.proto";
 import "util/log/eventpb/events.proto";
 
-// TODO (#151948): Update channel from TELEMETRY to CHANGEFEED once the
-// `log.channel_compatibility_mode_enabled` cluster setting is set to false
-// by default.
-
 // Category: Changefeed telemetry events
-// Channel: TELEMETRY
+// Channel: CHANGEFEED
 //
 // Events in this category pertain to changefeed usage and metrics.
 
 // CreateChangefeed is an event for any CREATE CHANGEFEED query that
 // successfully starts running. Failed CREATE statements will show up as
 // ChangefeedFailed events.
 //
-// Note: in version 26.1, these events will be moved to the `CHANGEFEED` channel.
-// To test compatability before this, set the cluster setting
-// `log.channel_compatibility_mode.enabled` to false. This will send the
-// events to `CHANGEFEED` instead of `TELEMETRY`.
+// Note: in version 26.1, these events moved to the `CHANGEFEED` channel.
+// To test compatability prior to this, set the cluster setting
+// `log.channel_compatibility_mode.enabled` to true. This will send the
+// events to `TELEMETRY` instead of `CHANGEFEED`.
 message CreateChangefeed {
   CommonChangefeedEventDetails common = 1 [(gogoproto.nullable) = false, (gogoproto.jsontag) = "", (gogoproto.embed) = true];
 
diff --git a/pkg/util/log/eventpb/eventlog_channels_generated.go b/pkg/util/log/eventpb/eventlog_channels_generated.go
diff --git a/pkg/util/log/eventpb/sql_audit_events.proto b/pkg/util/log/eventpb/sql_audit_events.proto
diff --git a/pkg/util/log/eventpb/telemetry.proto b/pkg/util/log/eventpb/telemetry.proto
diff --git a/pkg/util/log/migrator.go b/pkg/util/log/migrator.go

Original file line number	Diff line number	Diff line change
`@@ -2317,7 +2317,7 @@ func getChangefeedEventMigrator(migrateEvent bool) log.StructuredEventMigrator {`
`2317`	`2317`	`func() bool {`
`2318`	`2318`	`return migrateEvent`
`2319`	`2319`	`},`
`2320`		`- channel.CHANGEFEED,`
	`2320`	`+ channel.TELEMETRY,`
`2321`	`2321`	`)`
`2322`	`2322`	`}`
`2323`	`2323`
Original file line number	Diff line number	Diff line change
`@@ -12742,9 +12742,9 @@ func TestTableRenameDuringDatabaseLevelChangefeed(t *testing.T) {`
`12742`	`12742`
`12743`	`12743`	`func getChangefeedLoggingChannel(sv *settings.Values) logpb.Channel {`
`12744`	`12744`	`if log.ShouldMigrateEvent(sv) {`
`12745`		`- return logpb.Channel_CHANGEFEED`
	`12745`	`+ return logpb.Channel_TELEMETRY`
`12746`	`12746`	`}`
`12747`		`- return logpb.Channel_TELEMETRY`
	`12747`	`+ return logpb.Channel_CHANGEFEED`
`12748`	`12748`	`}`
`12749`	`12749`
`12750`	`12750`	`func TestCreateTableLevelChangefeedWithDBPrivilege(t *testing.T) {`
Original file line number	Diff line number	Diff line change
`@@ -167,10 +167,10 @@ type telemetrySpy struct {`
`167`	`167`
`168`	`168`	`func (l *telemetrySpy) channelsToIntercept() []log.Channel {`
`169`	`169`	`if log.ShouldMigrateEvent(l.sv) {`
`170`		`- return []log.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC}`
	`170`	`+ return []log.Channel{logpb.Channel_TELEMETRY}`
`171`	`171`	`}`
`172`	`172`
`173`		`- return []log.Channel{logpb.Channel_TELEMETRY}`
	`173`	`+ return []log.Channel{logpb.Channel_TELEMETRY, logpb.Channel_SQL_EXEC}`
`174`	`174`	`}`
`175`	`175`
`176`	`176`	`func (l *telemetrySpy) Intercept(entry []byte) {`
Original file line number	Diff line number	Diff line change
`@@ -3123,7 +3123,7 @@ func (ex *connExecutor) handleTxnRowsGuardrails(`
`3123`	`3123`	`}`
`3124`	`3124`	`migrator := log.NewStructuredEventMigrator(func() bool {`
`3125`	`3125`	`return log.ShouldMigrateEvent(ex.planner.ExecCfg().SV())`
`3126`		`- }, logpb.Channel_SQL_EXEC)`
	`3126`	`+ }, logpb.Channel_SQL_PERF)`
`3127`	`3127`	`migrator.StructuredEvent(ctx, severity.INFO, event)`
`3128`	`3128`	`logCounter.Inc(1)`
`3129`	`3129`	`}`
Original file line number	Diff line number	Diff line change
`@@ -765,9 +765,9 @@ func TestPerfLogging(t *testing.T) {`
`765`	`765`	`db.Exec(t, tc.query)`
`766`	`766`	`}`
`767`	`767`
`768`		`- expectedChannel := tc.channel`
`769`		`- if !log.ChannelCompatibilityModeEnabled.Get(&s.ClusterSettings().SV) {`
`770`		`- expectedChannel = logpb.Channel_SQL_EXEC`
	`768`	`+ expectedChannel := logpb.Channel_SQL_EXEC`
	`769`	`+ if log.ShouldMigrateEvent(&s.ClusterSettings().SV) {`
	`770`	`+ expectedChannel = tc.channel`
`771`	`771`	`}`
`772`	`772`
`773`	`773`	`var logRe = regexp.MustCompile(tc.logRe)`
Original file line number	Diff line number	Diff line change
`@@ -477,7 +477,7 @@ func (rh *RowHelper) CheckRowSize(`
`477`	`477`	`}`
`478`	`478`	`migrator := log.NewStructuredEventMigrator(func() bool {`
`479`	`479`	`return rh.migrateLargeRowLog`
`480`		`- }, logpb.Channel_SQL_EXEC)`
	`480`	`+ }, logpb.Channel_SQL_PERF)`
`481`	`481`	`migrator.StructuredEvent(ctx, severity.INFO, event)`
`482`	`482`	`}`
`483`	`483`	`if shouldErr {`