changefeedccl: more timeutil.Now -> crtime.NowMono conversions

I noticed a few of these when reading through a profile. I was able to
get most calls out of the per-event call path; however there is still
one call to walltime in our metrics recording that results from
calculating the time since the MVCC timestamp.

I doubt this has much of an impact, but it seems good to use crtime when
we can.

Epic: none
Release note: None

Author: stevendanna

pkg/ccl/changefeedccl/BUILD.bazel

@@ -169,6 +169,7 @@ go_library(
         "//pkg/util/uuid",
         "@com_github_apache_pulsar_client_go//pulsar",
         "@com_github_cockroachdb_changefeedpb//:go_default_library",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_redact//:redact",

pkg/ccl/changefeedccl/batching_sink.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/retry"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -265,7 +266,7 @@ type sinkBatch struct {
 	numMessages int
 	numKVBytes  int          // the total amount of uncompressed kv data in the batch
 	keys        intsets.Fast // the set of keys within the batch to provide to parallelIO
-	bufferTime  time.Time    // the earliest time a message was inserted into the batch
+	bufferTime  crtime.Mono  // the earliest time a message was inserted into the batch
 	mvcc        hlc.Timestamp
 
 	alloc  kvevent.Alloc
@@ -306,7 +307,7 @@ func hashToInt(h hash.Hash32, buf []byte) int {
 // Append adds the contents of a kvEvent to the batch, merging its alloc pool.
 func (sb *sinkBatch) Append(ctx context.Context, e *rowEvent) {
 	if sb.isEmpty() {
-		sb.bufferTime = timeutil.Now()
+		sb.bufferTime = crtime.NowMono()
 	}
 
 	sb.buffer.Append(ctx, e.key, e.val, attributes{
@@ -409,7 +410,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 
 		req, send, err := ioEmitter.AdmitRequest(ctx, batchBuffer)
 		if errors.Is(err, ErrNotEnoughQuota) {
-			waitStart := timeutil.Now()
+			waitStart := crtime.NowMono()
 
 			// Quota can only be freed by consuming a result.
 			select {
@@ -431,7 +432,7 @@ func (s *batchingSink) runBatchingWorker(ctx context.Context) {
 				return err
 			}
 
-			s.metrics.recordSinkBackpressure(timeutil.Since(waitStart))
+			s.metrics.recordSinkBackpressure(waitStart.Elapsed())
 		} else if err != nil {
 			return err
 		} else {

pkg/ccl/changefeedccl/cdceval/BUILD.bazel

@@ -43,7 +43,7 @@ go_library(
         "//pkg/util/ctxgroup",
         "//pkg/util/hlc",
         "//pkg/util/log",
-        "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_lib_pq//oid",
     ],

pkg/ccl/changefeedccl/cdceval/expr_eval.go

@@ -25,7 +25,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/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -256,10 +256,10 @@ func sameVersion(currentVersion, newVersion *cdcevent.EventDescriptor) bool {
 // planAndRun plans CDC expression and starts execution pipeline.
 func (e *familyEvaluator) planAndRun(ctx context.Context) (err error) {
 	if log.V(1) {
-		start := timeutil.Now()
+		start := crtime.NowMono()
 		defer func() {
 			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)
+				tree.AsString(e.norm), e.norm.desc.Version, start.Elapsed(), err)
 		}()
 	}
 

pkg/ccl/changefeedccl/cdcutils/BUILD.bazel

@@ -11,8 +11,8 @@ go_library(
         "//pkg/util/log",
         "//pkg/util/metric",
         "//pkg/util/quotapool",
-        "//pkg/util/timeutil",
         "//pkg/util/tracing",
+        "@com_github_cockroachdb_crlib//crtime",
     ],
 )
 

pkg/ccl/changefeedccl/cdcutils/throttle.go

@@ -17,8 +17,8 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/metric"
 	"github.com/cockroachdb/cockroach/pkg/util/quotapool"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/crlib/crtime"
 )
 
 // Throttler is a changefeed IO throttler.
@@ -168,9 +168,9 @@ func (m Metrics) MetricStruct() {}
 func waitQuota(
 	ctx context.Context, n int64, limit *quotapool.RateLimiter, c *metric.Counter,
 ) error {
-	start := timeutil.Now()
+	start := crtime.NowMono()
 	defer func() {
-		c.Inc(int64(timeutil.Since(start)))
+		c.Inc(start.Elapsed().Nanoseconds())
 	}()
 	return limit.WaitN(ctx, n)
 }

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -55,6 +55,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/uuid"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/logtags"
 	"github.com/cockroachdb/redact"
@@ -856,7 +857,7 @@ func (ca *changeAggregator) tick() error {
 		return err
 	}
 
-	queuedNanos := timeutil.Since(event.BufferAddTimestamp()).Nanoseconds()
+	queuedNanos := event.BufferAddTimestamp().Elapsed().Nanoseconds()
 	ca.metrics.QueueTimeNanos.Inc(queuedNanos)
 
 	switch event.Type() {
@@ -935,11 +936,11 @@ func (ca *changeAggregator) noteResolvedSpan(resolved jobspb.ResolvedSpan) error
 	checkpointFrontier := (advanced && forceFlush) || ca.frontierFlushLimiter.canSave(ctx)
 
 	if checkpointFrontier {
-		now := timeutil.Now()
+		now := crtime.NowMono()
 		if err := ca.flushFrontier(ctx); err != nil {
 			return err
 		}
-		ca.frontierFlushLimiter.doneSave(timeutil.Since(now))
+		ca.frontierFlushLimiter.doneSave(now.Elapsed())
 	}
 
 	return nil

pkg/ccl/changefeedccl/checkpoint/BUILD.bazel

@@ -15,7 +15,7 @@ go_library(
         "//pkg/util/hlc",
         "//pkg/util/metric",
         "//pkg/util/metric/aggmetric",
-        "//pkg/util/timeutil",
+        "@com_github_cockroachdb_crlib//crtime",
         "@com_github_cockroachdb_errors//:errors",
     ],
 )

pkg/ccl/changefeedccl/checkpoint/checkpoint.go

@@ -13,7 +13,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/jobs/jobspb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
-	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 )
 
@@ -27,7 +27,7 @@ func Make(
 	maxBytes int64,
 	metrics *Metrics,
 ) *jobspb.TimestampSpansMap {
-	start := timeutil.Now()
+	start := crtime.NowMono()
 
 	spanGroupMap := make(map[hlc.Timestamp]*roachpb.SpanGroup)
 	for s, ts := range spans {
@@ -60,7 +60,7 @@ func Make(
 	}
 
 	if metrics != nil {
-		metrics.CreateNanos.RecordValue(int64(timeutil.Since(start)))
+		metrics.CreateNanos.RecordValue(start.Elapsed().Nanoseconds())
 		metrics.TotalBytes.RecordValue(int64(cp.Size()))
 		metrics.TimestampCount.RecordValue(int64(cp.TimestampCount()))
 		metrics.SpanCount.RecordValue(int64(cp.SpanCount()))

pkg/ccl/changefeedccl/event_processing.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/syncutil"
 	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
+	"github.com/cockroachdb/crlib/crtime"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/redact"
 )
@@ -631,10 +632,9 @@ type parallelEventConsumer struct {
 var _ eventConsumer = (*parallelEventConsumer)(nil)
 
 func (c *parallelEventConsumer) ConsumeEvent(ctx context.Context, ev kvevent.Event) error {
-	startTime := timeutil.Now().UnixNano()
+	start := crtime.NowMono()
 	defer func() {
-		time := timeutil.Now().UnixNano()
-		c.metrics.ParallelConsumerConsumeNanos.RecordValue(time - startTime)
+		c.metrics.ParallelConsumerConsumeNanos.RecordValue(start.Elapsed().Nanoseconds())
 	}()
 
 	bucket := c.getBucketForEvent(ev)
@@ -760,10 +760,9 @@ func (c *parallelEventConsumer) setWorkerError(err error) error {
 // Flush flushes the consumer by blocking until all events are consumed,
 // or until there is an error.
 func (c *parallelEventConsumer) Flush(ctx context.Context) error {
-	startTime := timeutil.Now().UnixNano()
+	start := crtime.NowMono()
 	defer func() {
-		time := timeutil.Now().UnixNano()
-		c.metrics.ParallelConsumerFlushNanos.RecordValue(time - startTime)
+		c.metrics.ParallelConsumerFlushNanos.RecordValue(start.Elapsed().Nanoseconds())
 	}()
 
 	needFlush := func() bool {