Merge #154234

craig[bot] · stevendanna · craig[bot] · commit ac68bcaf4a54 · 2025-09-27T09:09:03.000Z
154234: *: use logtags.BuildBuffer() r=shubhamdhama a=stevendanna

This handles most of the places where we are adding multiple tags to the context. Most of these are not called frequently, but it still seems better to use the more efficient API since people often look to existing code as an example.

Epic: none
Release note: None

Co-authored-by: Steven Danna &lt;danna@cockroachlabs.com&gt;
diff --git a/pkg/crosscluster/logical/logical_replication_writer_processor.go b/pkg/crosscluster/logical/logical_replication_writer_processor.go
@@ -272,9 +272,11 @@ func newLogicalReplicationWriterProcessor(
 //
 // Start implements the RowSource interface.
 func (lrw *logicalReplicationWriterProcessor) Start(ctx context.Context) {
-	ctx = logtags.AddTag(ctx, "job", lrw.spec.JobID)
-	ctx = logtags.AddTag(ctx, "src-node", lrw.spec.PartitionSpec.PartitionID)
-	ctx = logtags.AddTag(ctx, "proc", lrw.ProcessorID)
+	tags := logtags.BuildBuffer()
+	tags.Add("job", lrw.spec.JobID)
+	tags.Add("src-node", lrw.spec.PartitionSpec.PartitionID)
+	tags.Add("proc", lrw.ProcessorID)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 	lrw.agg = tracing.TracingAggregatorForContext(ctx)
 	var listeners []tracing.EventListener
 	if lrw.agg != nil {
diff --git a/pkg/crosscluster/logical/offline_initial_scan_processor.go b/pkg/crosscluster/logical/offline_initial_scan_processor.go
@@ -189,11 +189,11 @@ func (o *offlineInitialScanProcessor) setup(ctx context.Context) error {
 }
 
 func (o *offlineInitialScanProcessor) Start(ctx context.Context) {
-	tags := &logtags.Buffer{}
-	tags = tags.Add("job", o.spec.JobID)
-	tags = tags.Add("src-node", o.spec.PartitionSpec.PartitionID)
-	tags = tags.Add("proc", o.ProcessorID)
-	ctx = logtags.AddTags(ctx, tags)
+	tags := logtags.BuildBuffer()
+	tags.Add("job", o.spec.JobID)
+	tags.Add("src-node", o.spec.PartitionSpec.PartitionID)
+	tags.Add("proc", o.ProcessorID)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 
 	ctx = o.StartInternal(ctx, offlineInitialScanProcessorName)
 
diff --git a/pkg/crosscluster/physical/stream_ingestion_processor.go b/pkg/crosscluster/physical/stream_ingestion_processor.go
@@ -396,8 +396,10 @@ func newStreamIngestionDataProcessor(
 //
 // Start implements the RowSource interface.
 func (sip *streamIngestionProcessor) Start(ctx context.Context) {
-	ctx = logtags.AddTag(ctx, "job", sip.spec.JobID)
-	ctx = logtags.AddTag(ctx, "proc", sip.ProcessorID)
+	tags := logtags.BuildBuffer()
+	tags.Add("job", sip.spec.JobID)
+	tags.Add("proc", sip.ProcessorID)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 	log.Dev.Infof(ctx, "starting ingest proc")
 	sip.agg = tracing.TracingAggregatorForContext(ctx)
 
diff --git a/pkg/crosscluster/producer/event_stream.go b/pkg/crosscluster/producer/event_stream.go
@@ -100,9 +100,11 @@ func (s *eventStream) Start(ctx context.Context, txn *kv.Txn) (retErr error) {
 	// false.  However, this generator never terminates without an error,
 	// so this method should be called once.  Be defensive and return an error
 	// if this method is called again.
-	ctx = logtags.AddTag(ctx, "id", s.streamID)
-	ctx = logtags.AddTag(ctx, "dst-node", s.spec.ConsumerNode)
-	ctx = logtags.AddTag(ctx, "dst-proc", s.spec.ConsumerProc)
+	tags := logtags.BuildBuffer()
+	tags.Add("id", s.streamID)
+	tags.Add("dst-node", s.spec.ConsumerNode)
+	tags.Add("dst-proc", s.spec.ConsumerProc)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 	if s.errCh != nil {
 		return errors.AssertionFailedf("expected to be started once")
 	}
diff --git a/pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go b/pkg/kv/kvclient/kvcoord/dist_sender_mux_rangefeed.go
@@ -369,14 +369,14 @@ func (m *rangefeedMuxer) startNodeMuxRangeFeed(
 	stream *future.Future[muxStreamOrError],
 ) (retErr error) {
 
-	tags := &logtags.Buffer{}
-	tags = tags.Add("mux_n", nodeID)
+	tags := logtags.BuildBuffer()
+	tags.Add("mux_n", nodeID)
 	// Add "generation" number to the context so that log messages and stacks can
 	// differentiate between multiple instances of mux rangefeed goroutine
 	// (this can happen when one was shutdown, then re-established).
-	tags = tags.Add("gen", atomic.AddInt64(&m.seqID, 1))
+	tags.Add("gen", atomic.AddInt64(&m.seqID, 1))
 
-	ctx = logtags.AddTags(ctx, tags)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 	ctx, restore := pprofutil.SetProfilerLabelsFromCtxTags(ctx)
 	defer restore()
 
diff --git a/pkg/rpc/context.go b/pkg/rpc/context.go
@@ -2023,12 +2023,12 @@ func (rpcCtx *Context) wrapCtx(
 	if remoteNodeID == 0 {
 		rnodeID = redact.SafeString("?")
 	}
-	l := &logtags.Buffer{}
-	l = l.Add(RemoteNodeTag, rnodeID)
-	l = l.Add(RemoteAddressTag, target)
-	l = l.Add(Class, class)
-	l = l.Add(RpcTag, nil)
-	return logtags.AddTags(ctx, l)
+	l := logtags.BuildBuffer()
+	l.Add(RemoteNodeTag, rnodeID)
+	l.Add(RemoteAddressTag, target)
+	l.Add(Class, class)
+	l.Add(RpcTag, nil)
+	return logtags.AddTags(ctx, l.Finish())
 }
 
 // grpcDialRaw connects to the remote node.
diff --git a/pkg/server/authserver/authentication.go b/pkg/server/authserver/authentication.go
@@ -260,9 +260,10 @@ func (s *authenticationServer) UserLogin(
 // DemoLogin is the same as UserLogin but using the GET method.
 // It is only available for 'cockroach demo' and test clusters.
 func (s *authenticationServer) DemoLogin(w http.ResponseWriter, req *http.Request) {
-	ctx := context.Background()
-	ctx = logtags.AddTag(ctx, "client", log.SafeOperational(req.RemoteAddr))
-	ctx = logtags.AddTag(ctx, "demologin", nil)
+	tags := logtags.BuildBuffer()
+	tags.Add("client", log.SafeOperational(req.RemoteAddr))
+	tags.Add("demologin", nil)
+	ctx := logtags.WithTags(context.Background(), tags.Finish())
 
 	fail := func(err error) {
 		w.WriteHeader(500)
diff --git a/pkg/server/node.go b/pkg/server/node.go
@@ -2266,14 +2266,14 @@ func (n *Node) muxRangeFeed(muxStream kvpb.RPCInternal_MuxRangeFeedStream) error
 				continue
 			}
 
-			tags := &logtags.Buffer{}
-			tags = tags.Add("r", req.RangeID)
-			tags = tags.Add("sm", req.Replica.StoreID)
-			tags = tags.Add("sid", req.StreamID)
+			tags := logtags.BuildBuffer()
+			tags.Add("r", req.RangeID)
+			tags.Add("sm", req.Replica.StoreID)
+			tags.Add("sid", req.StreamID)
 			if req.ConsumerID != 0 {
-				tags = tags.Add("cid", req.ConsumerID)
+				tags.Add("cid", req.ConsumerID)
 			}
-			streamCtx := logtags.AddTags(ctx, tags)
+			streamCtx := logtags.AddTags(ctx, tags.Finish())
 
 			streamSink := sm.NewStream(req.StreamID, req.RangeID)
 
diff --git a/pkg/server/server_controller_channel_orchestrator.go b/pkg/server/server_controller_channel_orchestrator.go
@@ -223,8 +223,10 @@ func (o *channelOrchestrator) startControlledServer(
 	// stopper will have its own tracer which is incompatible with the
 	// tracer attached to the incoming context.
 	tenantCtx := logtags.WithTags(context.Background(), logtags.FromContext(ctx))
-	tenantCtx = logtags.AddTag(tenantCtx, "tenant-orchestration", nil)
-	tenantCtx = logtags.AddTag(tenantCtx, "tenant", tenantName)
+	tags := logtags.BuildBuffer()
+	tags.Add("tenant-orchestration", nil)
+	tags.Add("tenant", tenantName)
+	tenantCtx = logtags.AddTags(tenantCtx, tags.Finish())
 
 	// ctlStopper is a stopper uniquely responsible for the control
 	// loop. It is separate from the tenantStopper defined below so
diff --git a/pkg/server/testserver.go b/pkg/server/testserver.go
@@ -2731,9 +2731,11 @@ func newClientRPCContext(
 	cid *base.ClusterIDContainer,
 	s serverutils.ApplicationLayerInterface,
 ) *rpc.Context {
-	ctx = logtags.AddTag(ctx, "testclient", nil)
-	ctx = logtags.AddTag(ctx, "user", user)
-	ctx = logtags.AddTag(ctx, "nsql", s.SQLInstanceID())
+	tags := logtags.BuildBuffer()
+	tags.Add("testclient", nil)
+	tags.Add("user", user)
+	tags.Add("nsql", s.SQLInstanceID())
+	ctx = logtags.AddTags(ctx, tags.Finish())
 
 	stopper := s.AppStopper()
 	if ctx.Done() == nil {
diff --git a/pkg/sql/conn_executor_exec.go b/pkg/sql/conn_executor_exec.go
@@ -393,11 +393,11 @@ func (ex *connExecutor) execStmtInOpenState(
 	}
 
 	if len(stmt.QueryTags) > 0 {
-		tags := &logtags.Buffer{}
+		tags := logtags.BuildBuffer()
 		for _, tag := range stmt.QueryTags {
-			tags = tags.Add("querytag-"+tag.Key, tag.Value)
+			tags.Add("querytag-"+tag.Key, tag.Value)
 		}
-		ctx = logtags.AddTags(ctx, tags)
+		ctx = logtags.AddTags(ctx, tags.Finish())
 	}
 
 	var queryTimeoutTicker *time.Timer
diff --git a/pkg/sql/pgwire/server.go b/pkg/sql/pgwire/server.go
@@ -952,8 +952,10 @@ func (s *Server) ServeConn(
 	// been overridden by a status parameter).
 	connDetails.RemoteAddress = sArgs.RemoteAddr.String()
 	sp := tracing.SpanFromContext(ctx)
-	ctx = logtags.AddTag(ctx, "client", log.SafeOperational(connDetails.RemoteAddress))
-	ctx = logtags.AddTag(ctx, preServeStatus.ConnType.String(), nil)
+	tags := logtags.BuildBuffer()
+	tags.Add("client", log.SafeOperational(connDetails.RemoteAddress))
+	tags.Add(preServeStatus.ConnType.String(), nil)
+	ctx = logtags.AddTags(ctx, tags.Finish())
 	sp.SetTag("conn_type", attribute.StringValue(preServeStatus.ConnType.String()))
 	sp.SetTag("client", attribute.StringValue(connDetails.RemoteAddress))
 
diff --git a/pkg/sql/schema_changer.go b/pkg/sql/schema_changer.go
@@ -731,19 +731,19 @@ func startGCJob(
 }
 
 func (sc *SchemaChanger) execLogTags() *logtags.Buffer {
-	buf := &logtags.Buffer{}
-	buf = buf.Add("scExec", nil)
+	buf := logtags.BuildBuffer()
+	buf.Add("scExec", nil)
 
-	buf = buf.Add("id", sc.descID)
+	buf.Add("id", sc.descID)
 	if sc.mutationID != descpb.InvalidMutationID {
-		buf = buf.Add("mutation", sc.mutationID)
+		buf.Add("mutation", sc.mutationID)
 	}
 	if sc.droppedDatabaseID != descpb.InvalidID {
-		buf = buf.Add("db", sc.droppedDatabaseID)
+		buf.Add("db", sc.droppedDatabaseID)
 	} else if !sc.droppedSchemaIDs.Empty() {
-		buf = buf.Add("schema", sc.droppedSchemaIDs)
+		buf.Add("schema", sc.droppedSchemaIDs)
 	}
-	return buf
+	return buf.Finish()
 }
 
 // notFirstInLine checks if that this schema changer is at the front of the line
diff --git a/pkg/sql/type_change.go b/pkg/sql/type_change.go
@@ -417,7 +417,7 @@ func (t *typeSchemaChanger) exec(ctx context.Context) error {
 					continue
 				}
 				rows, err := txn.QueryBufferedEx(ctx, "select-invalid-instances", txn.KV(),
-					sessiondata.NodeUserSessionDataOverride, `SELECT id FROM system.sql_instances 
+					sessiondata.NodeUserSessionDataOverride, `SELECT id FROM system.sql_instances
  							WHERE crdb_region = $1`, member.PhysicalRepresentation)
 				if err != nil {
 					return err
@@ -1430,10 +1430,10 @@ func (t *typeSchemaChanger) execWithRetry(ctx context.Context) error {
 }
 
 func (t *typeSchemaChanger) logTags() *logtags.Buffer {
-	buf := &logtags.Buffer{}
-	buf = buf.Add("typeChangeExec", nil)
-	buf = buf.Add("type", t.typeID)
-	return buf
+	buf := logtags.BuildBuffer()
+	buf.Add("typeChangeExec", nil)
+	buf.Add("type", t.typeID)
+	return buf.Finish()
 }
 
 // typeChangeResumer is the anchor struct for the type change job.
diff --git a/pkg/storage/pebble.go b/pkg/storage/pebble.go
@@ -910,9 +910,11 @@ func newPebble(ctx context.Context, cfg engineConfig) (p *Pebble, err error) {
 	logCtx := logtags.WithTags(context.Background(), logtags.FromContext(ctx))
 	// The store id, could not necessarily be determined when this function
 	// is called. Therefore, we use a container for the store id.
+	tags := logtags.BuildBuffer()
 	storeIDContainer := &base.StoreIDContainer{}
-	logCtx = logtags.AddTag(logCtx, "s", storeIDContainer)
-	logCtx = logtags.AddTag(logCtx, "pebble", nil)
+	tags.Add("s", storeIDContainer)
+	tags.Add("pebble", nil)
+	logCtx = logtags.AddTags(logCtx, tags.Finish())
 
 	cfg.opts.Local.ReadaheadConfig = objstorageprovider.NewReadaheadConfig()
 	updateReadaheadFn := func(ctx context.Context) {
