crosscluster/producer: tag job id/ dest node/ dest proc in event stream

Epic: none

Release note: none

pkg/ccl/crosscluster/producer/BUILD.bazel

@@ -65,6 +65,7 @@ go_library(
         "//pkg/util/tracing",
         "//pkg/util/uuid",
         "@com_github_cockroachdb_errors//:errors",
+        "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_golang_snappy//:snappy",
     ],
 )

pkg/ccl/crosscluster/producer/event_stream.go

@@ -39,6 +39,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/errors"
+	"github.com/cockroachdb/logtags"
 	"github.com/golang/snappy"
 )
 
@@ -105,6 +106,9 @@ 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)
 	if s.errCh != nil {
 		return errors.AssertionFailedf("expected to be started once")
 	}