changefeedccl: migrate CDC logs from DEV to CHANGEFEED logging channel

Replace all instances of log.Dev with log.Changefeed in the
changefeedccl directory.

Resolves: #153046
Release note: None

Author: KeithCh

pkg/ccl/changefeedccl/cdceval/expr_eval.go

@@ -258,7 +258,7 @@ func (e *familyEvaluator) planAndRun(ctx context.Context) (err error) {
 	if log.V(1) {
 		start := timeutil.Now()
 		defer func() {
-			log.Dev.Infof(ctx, "Planning for CDC expression %s (v=%d) took %s (err=%v)",
+			log.Changefeed.Infof(ctx, "Planning for CDC expression %s (v=%d) took %s (err=%v)",
 				tree.AsString(e.norm), e.norm.desc.Version, timeutil.Since(start), err)
 		}()
 	}

pkg/ccl/changefeedccl/cdcevent/event.go

@@ -612,7 +612,7 @@ func (d *eventDecoder) DecodeKV(
 	err = changefeedbase.WithTerminalError(errors.Wrapf(err,
 		"error decoding key %s@%s (hex_kv: %x)",
 		keys.PrettyPrint(nil, kv.Key), kv.Value.Timestamp, kvBytes))
-	log.Dev.Errorf(ctx, "terminal error decoding KV: %v", err)
+	log.Changefeed.Errorf(ctx, "terminal error decoding KV: %v", err)
 	return Row{}, err
 }
 
@@ -844,7 +844,7 @@ func MakeRowFromTuple(ctx context.Context, evalCtx *eval.Context, t *tree.DTuple
 		r.AddValueColumn(name, d.ResolvedType())
 		if err := r.SetValueDatumAt(i, d); err != nil {
 			if build.IsRelease() {
-				log.Dev.Warningf(ctx, "failed to set row value from tuple due to error %v", err)
+				log.Changefeed.Warningf(ctx, "failed to set row value from tuple due to error %v", err)
 				_ = r.SetValueDatumAt(i, tree.DNull)
 			} else {
 				panic(err)

pkg/ccl/changefeedccl/cdcevent/event_test.go

@@ -784,10 +784,10 @@ func TestEventColumnOrderingWithSchemaChanges(t *testing.T) {
 			require.NoError(t, err)
 
 			expectedEvents := len(tc.expectMainFamily) + len(tc.expectECFamily)
-			log.Dev.Infof(ctx, "expectedEvents: %d\n", expectedEvents)
+			log.Changefeed.Infof(ctx, "expectedEvents: %d\n", expectedEvents)
 			for i := 0; i < expectedEvents; i++ {
 				v, deleteRange := popRow(t)
-				log.Dev.Infof(ctx, "event[%d]: v=%+v, deleteRange=%+v", i, v, deleteRange)
+				log.Changefeed.Infof(ctx, "event[%d]: v=%+v, deleteRange=%+v", i, v, deleteRange)
 
 				if deleteRange != nil {
 					// Should not see a RangeFeedValue and a RangeFeedDeleteRange

pkg/ccl/changefeedccl/cdctest/nemeses.go

@@ -423,16 +423,16 @@ func RunNemesis(
 		time := timeutil.Now()
 		for i := 0; i < numInserts; i++ {
 			query := queryGen.Generate()
-			log.Dev.Infof(ctx, "Executing query: %s", query)
+			log.Changefeed.Infof(ctx, "Executing query: %s", query)
 			err := timeutil.RunWithTimeout(ctx, "nemeses populate table",
 				insertTimeout, func(ctx context.Context) error {
 					_, err := db.ExecContext(ctx, query)
 					return err
 				})
-			log.Dev.Infof(ctx, "Time taken to execute last query: %s", timeutil.Since(time))
+			log.Changefeed.Infof(ctx, "Time taken to execute last query: %s", timeutil.Since(time))
 			time = timeutil.Now()
 			if err != nil {
-				log.Dev.Infof(ctx, "Skipping query %s because error %s", query, err)
+				log.Changefeed.Infof(ctx, "Skipping query %s because error %s", query, err)
 				continue
 			}
 		}
@@ -454,18 +454,18 @@ func RunNemesis(
 		var id int
 		var ts string
 		if err := rows.Scan(&id, &ts); err != nil {
-			log.Dev.Infof(ctx, "# skipping row because error: %s", err)
+			log.Changefeed.Infof(ctx, "# skipping row because error: %s", err)
 			continue
 		}
-		log.Dev.Infof(ctx, "INSERT INTO foo (id,ts) VALUES (%d, %s);", id, ts)
+		log.Changefeed.Infof(ctx, "INSERT INTO foo (id,ts) VALUES (%d, %s);", id, ts)
 	}
 
 	cfo := newChangefeedOption(testName)
 	changefeedStatement := fmt.Sprintf(
 		`CREATE CHANGEFEED FOR foo %s`,
 		cfo.OptionString(),
 	)
-	log.Dev.Infof(ctx, "Using changefeed options: %s", changefeedStatement)
+	log.Changefeed.Infof(ctx, "Using changefeed options: %s", changefeedStatement)
 	foo, err := f.Feed(changefeedStatement)
 	if err != nil {
 		return nil, err
@@ -922,7 +922,7 @@ var compiledStateTransitions = fsm.Compile(stateTransitions)
 
 func logEvent(fn func(fsm.Args) error) func(fsm.Args) error {
 	return func(a fsm.Args) error {
-		log.Dev.Infof(a.Ctx, "Event: %#v, Payload: %#v\n", a.Event, a.Payload)
+		log.Changefeed.Infof(a.Ctx, "Event: %#v, Payload: %#v\n", a.Event, a.Payload)
 		return fn(a)
 	}
 }
@@ -1084,7 +1084,7 @@ func noteFeedMessage(a fsm.Args) error {
 			if err != nil {
 				return err
 			}
-			log.Dev.Infof(a.Ctx, "%v", string(m.Resolved))
+			log.Changefeed.Infof(a.Ctx, "%v", string(m.Resolved))
 			err = ns.v.NoteResolved(m.Partition, ts)
 			if err != nil {
 				return err
@@ -1096,7 +1096,7 @@ func noteFeedMessage(a fsm.Args) error {
 				return err
 			}
 			ns.availableRows--
-			log.Dev.Infof(a.Ctx, "%s->%s", m.Key, m.Value)
+			log.Changefeed.Infof(a.Ctx, "%s->%s", m.Key, m.Value)
 			return ns.v.NoteRow(m.Partition, string(m.Key), string(m.Value), ts, m.Topic)
 		}
 	}

pkg/ccl/changefeedccl/cdctest/row.go

@@ -95,10 +95,10 @@ func MakeRangeFeedValueReaderExtended(
 			case r := <-rows:
 				rowKey := r.Key.String() + r.Value.String()
 				if _, isDup := dups[rowKey]; isDup {
-					log.Dev.Infof(context.Background(), "Skip duplicate %s", roachpb.PrettyPrintKey(nil, r.Key))
+					log.Changefeed.Infof(context.Background(), "Skip duplicate %s", roachpb.PrettyPrintKey(nil, r.Key))
 					continue
 				}
-				log.Dev.Infof(context.Background(), "Read row %s", roachpb.PrettyPrintKey(nil, r.Key))
+				log.Changefeed.Infof(context.Background(), "Read row %s", roachpb.PrettyPrintKey(nil, r.Key))
 				dups[rowKey] = struct{}{}
 				return r, nil
 			case d := <-deleteRangeC:

pkg/ccl/changefeedccl/cdcutils/throttle.go

@@ -113,7 +113,7 @@ func NodeLevelThrottler(sv *settings.Values, metrics *Metrics) *Throttler {
 		configStr := changefeedbase.NodeSinkThrottleConfig.Get(sv)
 		if configStr != "" {
 			if err := json.Unmarshal([]byte(configStr), &config); err != nil {
-				log.Dev.Errorf(context.Background(),
+				log.Changefeed.Errorf(context.Background(),
 					"failed to parse node throttle config %q: err=%v; throttling disabled", configStr, err)
 			}
 		}

pkg/ccl/changefeedccl/changefeed.go

@@ -164,7 +164,7 @@ func emitResolvedTimestamp(
 		return err
 	}
 	if log.V(2) {
-		log.Dev.Infof(ctx, `resolved %s`, resolved)
+		log.Changefeed.Infof(ctx, `resolved %s`, resolved)
 	}
 	return nil
 }

pkg/ccl/changefeedccl/changefeed_dist.go

@@ -252,7 +252,7 @@ func startDistChangefeed(
 		return err
 	}
 	if log.ExpensiveLogEnabled(ctx, 2) {
-		log.Dev.Infof(ctx, "tracked spans: %s", trackedSpans)
+		log.Changefeed.Infof(ctx, "tracked spans: %s", trackedSpans)
 	}
 	localState.trackedSpans = trackedSpans
 
@@ -265,7 +265,7 @@ func startDistChangefeed(
 	if progress := localState.progress.GetChangefeed(); progress != nil && progress.SpanLevelCheckpoint != nil {
 		spanLevelCheckpoint = progress.SpanLevelCheckpoint
 		if log.V(2) {
-			log.Dev.Infof(ctx, "span-level checkpoint: %s", spanLevelCheckpoint)
+			log.Changefeed.Infof(ctx, "span-level checkpoint: %s", spanLevelCheckpoint)
 		}
 	}
 	p, planCtx, err := makePlan(execCtx, jobID, details, description, initialHighWater,
@@ -413,7 +413,7 @@ func makePlan(
 		evalCtx := execCtx.ExtendedEvalContext()
 		oracle := replicaoracle.NewOracle(replicaOracleChoice, dsp.ReplicaOracleConfig(locFilter))
 		if useBulkOracle.Get(&evalCtx.Settings.SV) {
-			log.Dev.Infof(ctx, "using bulk oracle for DistSQL planning")
+			log.Changefeed.Infof(ctx, "using bulk oracle for DistSQL planning")
 			oracle = kvfollowerreadsccl.NewBulkOracle(dsp.ReplicaOracleConfig(evalCtx.Locality), locFilter, kvfollowerreadsccl.StreakConfig{})
 		}
 		planCtx := dsp.NewPlanningCtxWithOracle(ctx, execCtx.ExtendedEvalContext(), nil, /* planner */
@@ -423,12 +423,12 @@ func makePlan(
 			return nil, nil, err
 		}
 		if log.ExpensiveLogEnabled(ctx, 2) {
-			log.Dev.Infof(ctx, "spans returned by DistSQL: %v", spanPartitions)
+			log.Changefeed.Infof(ctx, "spans returned by DistSQL: %v", spanPartitions)
 		}
 		switch {
 		case distMode == sql.LocalDistribution || rangeDistribution == defaultDistribution:
 		case rangeDistribution == balancedSimpleDistribution:
-			log.Dev.Infof(ctx, "rebalancing ranges using balanced simple distribution")
+			log.Changefeed.Infof(ctx, "rebalancing ranges using balanced simple distribution")
 			sender := execCtx.ExecCfg().DB.NonTransactionalSender()
 			distSender := sender.(*kv.CrossRangeTxnWrapperSender).Wrapped().(*kvcoord.DistSender)
 			ri := kvcoord.MakeRangeIterator(distSender)
@@ -438,7 +438,7 @@ func makePlan(
 				return nil, nil, err
 			}
 			if log.ExpensiveLogEnabled(ctx, 2) {
-				log.Dev.Infof(ctx, "spans after balanced simple distribution rebalancing: %v", spanPartitions)
+				log.Changefeed.Infof(ctx, "spans after balanced simple distribution rebalancing: %v", spanPartitions)
 			}
 		default:
 			return nil, nil, errors.AssertionFailedf("unsupported dist strategy %d and dist mode %d",
@@ -473,7 +473,7 @@ func makePlan(
 		aggregatorSpecs := make([]*execinfrapb.ChangeAggregatorSpec, len(spanPartitions))
 		for i, sp := range spanPartitions {
 			if log.ExpensiveLogEnabled(ctx, 2) {
-				log.Dev.Infof(ctx, "watched spans for node %d: %v", sp.SQLInstanceID, sp)
+				log.Changefeed.Infof(ctx, "watched spans for node %d: %v", sp.SQLInstanceID, sp)
 			}
 
 			watches := make([]execinfrapb.ChangeAggregatorSpec_Watch, len(sp.Spans))
@@ -543,11 +543,11 @@ func makePlan(
 			flowSpecs,
 			execinfrapb.DiagramFlags{},
 		); err != nil {
-			log.Dev.Warningf(ctx, "failed to generate changefeed plan diagram: %s", err)
+			log.Changefeed.Warningf(ctx, "failed to generate changefeed plan diagram: %s", err)
 		} else if diagURL := diagURL.String(); len(diagURL) > maxLenDiagURL {
-			log.Dev.Warningf(ctx, "changefeed plan diagram length is too large to be logged: %d", len(diagURL))
+			log.Changefeed.Warningf(ctx, "changefeed plan diagram length is too large to be logged: %d", len(diagURL))
 		} else {
-			log.Dev.Infof(ctx, "changefeed plan diagram: %s", diagURL)
+			log.Changefeed.Infof(ctx, "changefeed plan diagram: %s", diagURL)
 		}
 
 		return p, planCtx, nil
@@ -656,7 +656,7 @@ func rebalanceSpanPartitions(
 		nRanges, ok := p.NumRanges()
 		// We cannot rebalance if we're missing range information.
 		if !ok {
-			log.Dev.Warning(ctx, "skipping rebalance due to missing range info")
+			log.Changefeed.Warning(ctx, "skipping rebalance due to missing range info")
 			return partitions, nil
 		}
 		builders[i].numRanges = nRanges