changefeedccl: add timer for inner batching sink client flush

[asg0451]

Add a timer for the inner flushes inside sink
clients. Where possible (batching_sink users,
webhook v1, pubsub v1, cloud storage), this is
distinct from changefeed.batch_hist_nanos in that
in only captures time spent flushing data to the
downstream endpoint, not time spent buffering
data.

This also fixes a bug where timers were not
correctly registered with the metric system.

Part of: #127784

Release note (enterprise change): Added a timer
for inner sink client flushes. Fixed a bug where
timers were not correctly registered with the
metric system.

[blathers-crl]

It looks like your PR touches production code but doesn't add or edit any test code. Did you consider adding tests to your PR?

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}

[cockroach-teamcity]

This change is 

[wenyihu6]

Do we want to do the same thing across other sinks as well and rename this metric to SinkClientFlush? Perhaps in a follow up PR.

[rharding6373]

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asg0451 and @wenyihu6)

---

pkg/ccl/changefeedccl/timers/timers.go line 48 at r2 (raw file):

		Encode:                    b.Histogram(histogramOptsFor("changefeed.stage.encode.latency", "Latency of the changefeed stage: encoding data")),
		EmitRow:                   b.Histogram(histogramOptsFor("changefeed.stage.emit_row.latency", "Latency of the changefeed stage: emitting row to sink")),
		BatchingSinkClientFlush:   b.Histogram(histogramOptsFor("changefeed.stage.batching_sink_client_flush.latency", "Latency of the changefeed stage: flushing the inner sink client, for sinks using batching_sink")),

I agree with Wenyi that it would be great if this metric were agnostic to the batching sink, so we could extend it to all sinks in the future. Sink client latency is applicable to all cases.

[asg0451]

The reason I added this to batching sink only is that in the other cases BatchHistNanos is sufficient for this. it's only in batching sink-wrapped sinks that it is counting an additional queueing point

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @rharding6373 and @wenyihu6)

---

pkg/ccl/changefeedccl/timers/timers.go line 48 at r2 (raw file):

Previously, rharding6373 (Rachael Harding) wrote…

I agree with Wenyi that it would be great if this metric were agnostic to the batching sink, so we could extend it to all sinks in the future. Sink client latency is applicable to all cases.

The reason I added this to batching sink only is that in the other cases BatchHistNanos is sufficient for this. it's only in batching sink-wrapped sinks that it is counting an additional queueing point

[wenyihu6]

If that's the case, how do you feel about if we just rename BatchHistNanos to SinkClientFlush and keep SinkClientFlush consistent across different sink types - as in they should always measure client sink flush? And we can add something else as a separate metrics for BatchingSink.BatchHistNanos to track the time between we add something to the batch for the first time until the batch is flushed.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @rharding6373)

[asg0451]

i'm open to something like that, let's discuss later today if there are no other topics

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @rharding6373)

[rharding6373]

I only have a nit about the name of the metric itself as exposed to users. Thanks for going above and beyond the batching sinks in this PR!

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @asg0451 and @wenyihu6)

---

pkg/ccl/changefeedccl/timers/timers.go line 48 at r2 (raw file):

Previously, asg0451 (Miles Frankel) wrote…

The reason I added this to batching sink only is that in the other cases BatchHistNanos is sufficient for this. it's only in batching sink-wrapped sinks that it is counting an additional queueing point

minor nit: would sink_client be descriptive enough to understand the metric, or do we need sink_client_send? sink_client_send reads awkwardly imo (I'm indifferent about DownstreamClientSend, since it's internal)

[asg0451]

Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @rharding6373 and @wenyihu6)

---

pkg/ccl/changefeedccl/timers/timers.go line 48 at r2 (raw file):

Previously, rharding6373 (Rachael Harding) wrote…

minor nit: would sink_client be descriptive enough to understand the metric, or do we need sink_client_send? sink_client_send reads awkwardly imo (I'm indifferent about DownstreamClientSend, since it's internal)

i'm not 100% on this name -- i'm trying to avoid ambiguity with the batching sink's SinkClient, as this metric is not specific to that. hence "send".... if you have a better idea that's not ambiguous on that axis lmk

[asg0451]

bors r=rharding6373

[craig]

Build succeeded:

• lint
• linux_amd64_build
• macos_arm64_build
• check_generated_code
• macos_amd64_build
• acceptance
• local_roachtest
• linux_amd64_fips_build
• local_roachtest_fips
• linux_arm64_build
• windows_build
• examples_orms
• unit_tests
• docker_image_amd64

[blathers-crl]

Encountered an error creating backports. Some common things that can go wrong:

1. The backport branch might have already existed.
2. There was a merge conflict.
3. The backport branch contained merge commits.

You might need to create your backport manually using the backport tool.

---

error creating merge commit from d7ccd77 to blathers/backport-release-23.2-132572: POST https://api.github.com/repos/cockroachdb/cockroach/merges: 409 Merge conflict []

you may need to manually resolve merge conflicts with the backport tool.

Backport to branch 23.2.x failed. See errors above.

---

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}