Merge #151807 #152177

151807: changefeedccl: prep logging channel migration of changefeed events r=kyle-a-wong a=kyle-a-wong

In v26.1, changefeed events will be moved from the TELEMETRY logging
channel to the CHANGEFEED logging channel.

This commit gates this migration under the cluster setting:
`log.channel_compatibility_mode.enabled` and will log these events
to the CHANGEFEED 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](https://cockroachlabs.atlassian.net/browse/CRDB-53410)
Part of: [CRDB-53412](https://cockroachlabs.atlassian.net/browse/CRDB-53412)
Release note (ops change): changefeed events will be moved to the
CHANGEFEED 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 CHANGEFEED channel so
this shouldn't be tested in a production environment.

---
This is a stacked PR, only the last commit should be reviewed

152177: docs: update cluster settings to use correct product labels r=jhlodin a=jhlodin

The cluster setting labeling still applies pre-Cloud 2.0 product names (dedicated/serverless) resulting in old product names under the **Supported deployments** column of the [cluster settings documentation](https://www.cockroachlabs.com/docs/stable/cluster-settings.html).

Fixes #152179

Co-authored-by: Kyle Wong <[email protected]>
Co-authored-by: Joe Lodin <[email protected]>

Author: 3 people

docs/generated/eventlog.md

@@ -32,6 +32,11 @@ Events in this category are logged to the `TELEMETRY` channel.
 
 An event of type `alter_changefeed` is an event for any ALTER CHANGEFEED statements that are run.
 
+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`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|
@@ -54,6 +59,11 @@ An event of type `alter_changefeed` is an event for any ALTER CHANGEFEED stateme
 
 An event of type `changefeed_canceled` is an event for any changefeed cancellations.
 
+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`.
+
 
 
 
@@ -73,6 +83,11 @@ An event of type `changefeed_canceled` is an event for any changefeed cancellati
 
 An event of type `changefeed_emitted_bytes` is an event representing the bytes emitted by a changefeed over an interval.
 
+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`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|
@@ -99,6 +114,11 @@ An event of type `changefeed_emitted_bytes` is an event representing the bytes e
 An event of type `changefeed_failed` is an event for any changefeed failure since the plan hook
 was triggered.
 
+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`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|
@@ -123,6 +143,11 @@ An event of type `create_changefeed` is an event for any CREATE CHANGEFEED query
 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`.
+
 
 | Field | Description | Sensitive |
 |--|--|--|

docs/generated/settings/BUILD.bazel

@@ -4,7 +4,7 @@ genrule(
     cmd = """
     $(location //pkg/cmd/cockroach-short) gen settings-list \
     --show-class --format=rawhtml --class-header-label 'Supported Deployments' \
-    --class-labels 'Dedicated/Self-Hosted,Dedicated/Self-hosted (read-write); Serverless (read-only),Serverless/Dedicated/Self-Hosted' > $@
+    --class-labels 'Advanced/Self-Hosted,Advanced/Self-hosted (read-write); Basic/Standard (read-only),Basic/Standard/Advanced/Self-Hosted' > $@
     """,
     tools = ["//pkg/cmd/cockroach-short"],
     visibility = [

docs/generated/settings/settings.html

@@ -144,6 +144,7 @@ go_library(
         "//pkg/util/intsets",
         "//pkg/util/json",
         "//pkg/util/log",
+        "//pkg/util/log/channel",
         "//pkg/util/log/eventpb",
         "//pkg/util/log/logcrash",
         "//pkg/util/log/severity",
@@ -280,6 +281,7 @@ go_test(
         "//pkg/server",
         "//pkg/server/serverpb",
         "//pkg/server/telemetry",
+        "//pkg/settings",
         "//pkg/settings/cluster",
         "//pkg/spanconfig",
         "//pkg/spanconfig/spanconfigjob",
@@ -337,6 +339,8 @@ go_test(
         "//pkg/util/leaktest",
         "//pkg/util/log",
         "//pkg/util/log/eventpb",
+        "//pkg/util/log/logpb",
+        "//pkg/util/log/logtestutils",
         "//pkg/util/metamorphic",
         "//pkg/util/mon",
         "//pkg/util/parquet",

pkg/ccl/changefeedccl/BUILD.bazel

@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/sql/types"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
 	"github.com/cockroachdb/errors"
 )
@@ -250,8 +251,8 @@ func alterChangefeedPlanHook(
 		}
 
 		telemetry.Count(telemetryPath)
-
-		logAlterChangefeedTelemetry(ctx, j, jobPayload.Description, targets.Size)
+		shouldMigrate := log.ShouldMigrateEvent(p.ExecCfg().SV())
+		logAlterChangefeedTelemetry(ctx, j, jobPayload.Description, targets.Size, shouldMigrate)
 
 		select {
 		case <-ctx.Done():

pkg/ccl/changefeedccl/alter_changefeed_stmt.go

@@ -54,6 +54,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/ctxgroup"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
+	"github.com/cockroachdb/cockroach/pkg/util/log/channel"
 	"github.com/cockroachdb/cockroach/pkg/util/log/eventpb"
 	"github.com/cockroachdb/cockroach/pkg/util/log/severity"
 	"github.com/cockroachdb/cockroach/pkg/util/randutil"
@@ -78,8 +79,8 @@ func init() {
 	sql.AddPlanHook("changefeed", changefeedPlanHook, changefeedTypeCheck)
 	jobs.RegisterConstructor(
 		jobspb.TypeChangefeed,
-		func(job *jobs.Job, _ *cluster.Settings) jobs.Resumer {
-			r := &changefeedResumer{job: job}
+		func(job *jobs.Job, s *cluster.Settings) jobs.Resumer {
+			r := &changefeedResumer{job: job, sv: &s.SV}
 			r.mu.perNodeAggregatorStats = make(bulk.ComponentAggregatorStats)
 			return r
 		},
@@ -282,7 +283,8 @@ func changefeedPlanHook(
 			p.ExtendedEvalContext().Descs.ReleaseAll(ctx)
 
 			telemetry.Count(`changefeed.create.core`)
-			logCreateChangefeedTelemetry(ctx, jr, changefeedStmt.Select != nil, targets.Size)
+			shouldMigrate := log.ShouldMigrateEvent(p.ExecCfg().SV())
+			logCreateChangefeedTelemetry(ctx, jr, changefeedStmt.Select != nil, targets.Size, shouldMigrate)
 			if err := maybeShowCursorAgeWarning(ctx, p, opts); err != nil {
 				return err
 			}
@@ -384,8 +386,8 @@ func changefeedPlanHook(
 		if err := maybeShowCursorAgeWarning(ctx, p, opts); err != nil {
 			return err
 		}
-
-		logCreateChangefeedTelemetry(ctx, jr, changefeedStmt.Select != nil, targets.Size)
+		shouldMigrate := log.ShouldMigrateEvent(p.ExecCfg().SV())
+		logCreateChangefeedTelemetry(ctx, jr, changefeedStmt.Select != nil, targets.Size, shouldMigrate)
 
 		select {
 		case <-ctx.Done():
@@ -400,7 +402,9 @@ func changefeedPlanHook(
 	rowFnLogErrors := func(ctx context.Context, resultsCh chan<- tree.Datums) error {
 		err := rowFn(ctx, resultsCh)
 		if err != nil {
-			logChangefeedFailedTelemetryDuringStartup(ctx, description, failureTypeForStartupError(err))
+
+			shouldMigrate := log.ShouldMigrateEvent(p.ExecCfg().SV())
+			logChangefeedFailedTelemetryDuringStartup(ctx, description, failureTypeForStartupError(err), shouldMigrate)
 			var e *kvpb.BatchTimestampBeforeGCError
 			if errors.As(err, &e) && opts.HasStartCursor() {
 				err = errors.Wrapf(err,
@@ -1320,6 +1324,7 @@ func validateAndNormalizeChangefeedExpression(
 
 type changefeedResumer struct {
 	job *jobs.Job
+	sv  *settings.Values
 
 	mu struct {
 		syncutil.Mutex
@@ -1792,17 +1797,17 @@ func (b *changefeedResumer) OnFailOrCancel(
 		}
 		numTargets = targets.Size
 	}
-
+	shouldMigrate := log.ShouldMigrateEvent(b.sv)
 	// If this job has failed (not canceled), increment the counter.
 	if jobs.HasErrJobCanceled(
 		errors.DecodeError(ctx, *b.job.Payload().FinalResumeError),
 	) {
 		telemetry.Count(`changefeed.enterprise.cancel`)
-		logChangefeedCanceledTelemetry(ctx, b.job, numTargets)
+		logChangefeedCanceledTelemetry(ctx, b.job, numTargets, shouldMigrate)
 	} else {
 		telemetry.Count(`changefeed.enterprise.fail`)
 		exec.ExecCfg().JobRegistry.MetricsStruct().Changefeed.(*Metrics).Failures.Inc(1)
-		logChangefeedFailedTelemetry(ctx, b.job, changefeedbase.UnknownError, numTargets)
+		logChangefeedFailedTelemetry(ctx, b.job, changefeedbase.UnknownError, numTargets, shouldMigrate)
 	}
 	return nil
 }
@@ -1890,7 +1895,7 @@ func getChangefeedTargetName(
 }
 
 func logCreateChangefeedTelemetry(
-	ctx context.Context, jr *jobs.Record, isTransformation bool, numTargets uint,
+	ctx context.Context, jr *jobs.Record, isTransformation bool, numTargets uint, migrateEvent bool,
 ) {
 	var changefeedEventDetails eventpb.CommonChangefeedEventDetails
 	if jr != nil {
@@ -1903,11 +1908,11 @@ func logCreateChangefeedTelemetry(
 		Transformation:               isTransformation,
 	}
 
-	log.StructuredEvent(ctx, severity.INFO, createChangefeedEvent)
+	getChangefeedEventMigrator(migrateEvent).StructuredEvent(ctx, severity.INFO, createChangefeedEvent)
 }
 
 func logAlterChangefeedTelemetry(
-	ctx context.Context, job *jobs.Job, prevDescription string, numTargets uint,
+	ctx context.Context, job *jobs.Job, prevDescription string, numTargets uint, migrateEvent bool,
 ) {
 	var changefeedEventDetails eventpb.CommonChangefeedEventDetails
 	if job != nil {
@@ -1921,11 +1926,15 @@ func logAlterChangefeedTelemetry(
 		PreviousDescription:          prevDescription,
 	}
 
-	log.StructuredEvent(ctx, severity.INFO, alterChangefeedEvent)
+	getChangefeedEventMigrator(migrateEvent).StructuredEvent(ctx, severity.INFO, alterChangefeedEvent)
 }
 
 func logChangefeedFailedTelemetry(
-	ctx context.Context, job *jobs.Job, failureType changefeedbase.FailureType, numTargets uint,
+	ctx context.Context,
+	job *jobs.Job,
+	failureType changefeedbase.FailureType,
+	numTargets uint,
+	migrateEvent bool,
 ) {
 	var changefeedEventDetails eventpb.CommonChangefeedEventDetails
 	if job != nil {
@@ -1938,21 +1947,26 @@ func logChangefeedFailedTelemetry(
 		FailureType:                  failureType,
 	}
 
-	log.StructuredEvent(ctx, severity.INFO, changefeedFailedEvent)
+	getChangefeedEventMigrator(migrateEvent).StructuredEvent(ctx, severity.INFO, changefeedFailedEvent)
 }
 
 func logChangefeedFailedTelemetryDuringStartup(
-	ctx context.Context, description string, failureType changefeedbase.FailureType,
+	ctx context.Context,
+	description string,
+	failureType changefeedbase.FailureType,
+	migrateEvent bool,
 ) {
 	changefeedFailedEvent := &eventpb.ChangefeedFailed{
 		CommonChangefeedEventDetails: eventpb.CommonChangefeedEventDetails{Description: description},
 		FailureType:                  failureType,
 	}
 
-	log.StructuredEvent(ctx, severity.INFO, changefeedFailedEvent)
+	getChangefeedEventMigrator(migrateEvent).StructuredEvent(ctx, severity.INFO, changefeedFailedEvent)
 }
 
-func logChangefeedCanceledTelemetry(ctx context.Context, job *jobs.Job, numTargets uint) {
+func logChangefeedCanceledTelemetry(
+	ctx context.Context, job *jobs.Job, numTargets uint, migrateEvent bool,
+) {
 	var changefeedEventDetails eventpb.CommonChangefeedEventDetails
 	if job != nil {
 		changefeedDetails := job.Details().(jobspb.ChangefeedDetails)
@@ -1962,7 +1976,7 @@ func logChangefeedCanceledTelemetry(ctx context.Context, job *jobs.Job, numTarge
 	changefeedCanceled := &eventpb.ChangefeedCanceled{
 		CommonChangefeedEventDetails: changefeedEventDetails,
 	}
-	log.StructuredEvent(ctx, severity.INFO, changefeedCanceled)
+	getChangefeedEventMigrator(migrateEvent).StructuredEvent(ctx, severity.INFO, changefeedCanceled)
 }
 
 func makeCommonChangefeedEventDetails(
@@ -2101,3 +2115,12 @@ func maybeUpgradePreProductionReadyExpression(
 		"Please see CDC documentation on the use of new cdc_prev tuple.",
 		tree.AsString(oldExpression), tree.AsString(newExpression))
 }
+
+func getChangefeedEventMigrator(migrateEvent bool) log.StructuredEventMigrator {
+	return log.NewStructuredEventMigrator(
+		func() bool {
+			return migrateEvent
+		},
+		channel.CHANGEFEED,
+	)
+}