optimize hot path by caching contexual attribute

[jgao54]

Two related fixes in this PR

1. Cache contextual attribute

All our counters/gauges are implemented with ContextAware*, which when executing Record will attempt to build attributes based on context:

func (a *ContextAwareInt64SyncGauge) Record(ctx context.Context, value int64, options ...metric.RecordOption) {
	newOptions := append([]metric.RecordOption{buildContextualAttributes(ctx)}, options...)
	a.Int64Gauge.Record(ctx, value, newOptions...)
}

This is intentional to standardize attributes (see buildContextualAttributes). However, on hot path like in the PullRecord loop, this can get pretty expensive and wasteful. We have observed flow-worker spending ~14% CPU just rebuilding attributes from the same context and generating the same result:

Fix here is to front the context building with a cache. Benchmark result on my machine shows ~3.6x improvement:

BenchmarkContextAwareGaugeRecord
BenchmarkContextAwareGaugeRecord/cache-hit
BenchmarkContextAwareGaugeRecord/cache-hit-14         	 2929458	       405.3 ns/op
BenchmarkContextAwareGaugeRecord/cache-miss
BenchmarkContextAwareGaugeRecord/cache-miss-14        	  781149	      1489 ns/op

2. Reduce gauge record frequency for MongoDB

A second optimization is in the MongoDB PullRecord loop, where instead of recording the gauge on every change event, only do so once per second. This is because SyncGauge is still not free (as seen in benchmark above), and Record just updates the cached value in the Gauge, not actually exporting it, so is essentially wasted work. This was attributing to another 5% of the cpu:

As a follow-up, worth also apply this change for PG/MySQL, but Mongo tends to be the most bottlenecked on CDC due to slow JSON processing, so prioritizing it here (also considered generalizing this mechanism into the gauge itself, but it's a bit unconventional to build sampling into gauge (more reasonable for traces).

[claude]

Code review

No issues found. Checked for bugs and CLAUDE.md compliance.

[dtunikov]

very nice

[jgao54]

Claude caught the scenario that when there are more than one mirrors in the flow pulling records concurrently, gauge will record in an interleaving manner with different context (since gauge/counter are singletons), and the cache wouldn't help much. Updated PR to have one cached entry per mirror instead. Benchmark shows the result before and after:

Before:

BenchmarkContextAwareGaugeRecord
BenchmarkContextAwareGaugeRecord/cache-hit
BenchmarkContextAwareGaugeRecord/cache-hit-14         	 2871757	       404.3 ns/op
BenchmarkContextAwareGaugeRecord/cache-miss
BenchmarkContextAwareGaugeRecord/cache-miss-14        	  790094	      1488 ns/op
BenchmarkContextAwareGaugeRecord/cache-hit-interleaved-ctx
BenchmarkContextAwareGaugeRecord/cache-hit-interleaved-ctx-14         	  796032	      1465 ns/op

After:

BenchmarkContextAwareGaugeRecord
BenchmarkContextAwareGaugeRecord/cache-hit
BenchmarkContextAwareGaugeRecord/cache-hit-14         	 2719821	       424.3 ns/op
BenchmarkContextAwareGaugeRecord/cache-miss
BenchmarkContextAwareGaugeRecord/cache-miss-14        	  751904	      1596 ns/op
BenchmarkContextAwareGaugeRecord/cache-hit-interleaved-ctx
BenchmarkContextAwareGaugeRecord/cache-hit-interleaved-ctx-14         	 2823506	       426.7 ns/op