google: sync export with bernot-dev/release-3.13.0-gmp - #330
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the Google Cloud monitoring exporter configurations, updates metric metadata structures to use MetricFamily, and simplifies histogram distribution building in transform.go. It also introduces a comprehensive testing framework under google/export/gcm/promtest. Feedback on these changes includes addressing potential panics in test helpers by avoiding global signal handlers and asynchronous t.Logf calls, resolving variable shadowing of the testing object, properly clearing and recycling completed distributions from the cache, correcting a misleading error message, and optimizing test retry polling intervals to reduce test suite latency.
| func (l *localExportWithGCM) start(t testing.TB, _ e2e.Environment) (v1.API, map[string]string) { | ||
| t.Helper() | ||
|
|
||
| ctx, cancel := context.WithCancel(signals.SetupSignalHandler()) |
There was a problem hiding this comment.
Using signals.SetupSignalHandler() in a test helper is highly discouraged. SetupSignalHandler registers global signal handlers and panics if called more than once in the same test binary (which happens if multiple tests use this helper or if tests are run with -count). It should be replaced with context.Background().
| ctx, cancel := context.WithCancel(signals.SetupSignalHandler()) | |
| ctx, cancel := context.WithCancel(context.Background()) |
| go func() { | ||
| if err := l.e.Run(); err != nil { | ||
| t.Logf("running exporter: %s", err) | ||
| } | ||
| }() |
There was a problem hiding this comment.
Calling t.Logf inside an asynchronous goroutine that may run during or after the test cleanup phase can cause a panic (panic: Logf called after test finished). It is safer to log to os.Stderr or use a standard logger.
| go func() { | |
| if err := l.e.Run(); err != nil { | |
| t.Logf("running exporter: %s", err) | |
| } | |
| }() | |
| go func() { | |
| if err := l.e.Run(); err != nil { | |
| fmt.Fprintf(os.Stderr, "running exporter: %s\n", err) | |
| } | |
| }() |
| t := timestamp.FromTime(time.Now()) | ||
| _, parsedTimestamp, val := tp.Series() | ||
| if parsedTimestamp != nil { | ||
| t = *parsedTimestamp | ||
| } | ||
| metadata[currMeta.MetricFamily] = currMeta | ||
|
|
||
| lset := labels.New() | ||
| tp.Labels(&lset) | ||
| l.labelsByRef[storage.SeriesRef(ref)] = lset | ||
|
|
||
| batch = append(batch, record.RefSample{ | ||
| Ref: chunks.HeadSeriesRef(ref), V: val, T: t, | ||
| }) |
There was a problem hiding this comment.
The variable t (representing the timestamp) shadows the testing.TB parameter t of the enclosing function injectScrapes. This makes it impossible to use the testing object t for assertions or logging within this block. Rename the timestamp variable to ts to avoid shadowing.
| t := timestamp.FromTime(time.Now()) | |
| _, parsedTimestamp, val := tp.Series() | |
| if parsedTimestamp != nil { | |
| t = *parsedTimestamp | |
| } | |
| metadata[currMeta.MetricFamily] = currMeta | |
| lset := labels.New() | |
| tp.Labels(&lset) | |
| l.labelsByRef[storage.SeriesRef(ref)] = lset | |
| batch = append(batch, record.RefSample{ | |
| Ref: chunks.HeadSeriesRef(ref), V: val, T: t, | |
| }) | |
| ts := timestamp.FromTime(time.Now()) | |
| _, parsedTimestamp, val := tp.Series() | |
| if parsedTimestamp != nil { | |
| ts = *parsedTimestamp | |
| } | |
| metadata[currMeta.MetricFamily] = currMeta | |
| lset := labels.New() | |
| tp.Labels(&lset) | |
| l.labelsByRef[storage.SeriesRef(ref)] = lset | |
| batch = append(batch, record.RefSample{ | |
| Ref: chunks.HeadSeriesRef(ref), V: val, T: ts, | |
| }) |
| if !dist.complete() { | ||
| continue | ||
| } | ||
| dp, err := dist.build(dist.lset) | ||
| dp, err := dist.build(e.lset) | ||
| if err != nil { | ||
| return nil, samples[consumed:], err | ||
| } | ||
| if dp != nil && dist.proto != nil { | ||
| //nolint:govet | ||
| ts := *dist.proto | ||
| ts.Points = []*monitoring_pb.Point{{ | ||
| Interval: &monitoring_pb.TimeInterval{ | ||
| StartTime: getTimestamp(dist.resetTimestamp), | ||
| EndTime: getTimestamp(dist.timestamp), | ||
| }, | ||
| Value: &monitoring_pb.TypedValue{ | ||
| Value: &monitoring_pb.TypedValue_DistributionValue{DistributionValue: dp}, | ||
| }, | ||
| }} | ||
| b.histResultsBuf = append(b.histResultsBuf, hashedSeries{hash: dist.hash, proto: &ts}) | ||
| return nil, 0, samples[consumed:], err | ||
| } | ||
| return dp, dist.resetTimestamp, samples[consumed:], nil |
There was a problem hiding this comment.
When a distribution is completed and returned, it is not removed from b.dists. This means stale state remains in the map, which can cause subsequent scrapes of the same histogram in the same batch to fail or be skipped due to mismatching timestamps. Additionally, immediately recycling the completed distribution back to the pool reduces memory pressure. We should delete the completed distribution from b.dists and call putDistribution(dist) before returning.
| if !dist.complete() { | |
| continue | |
| } | |
| dp, err := dist.build(dist.lset) | |
| dp, err := dist.build(e.lset) | |
| if err != nil { | |
| return nil, samples[consumed:], err | |
| } | |
| if dp != nil && dist.proto != nil { | |
| //nolint:govet | |
| ts := *dist.proto | |
| ts.Points = []*monitoring_pb.Point{{ | |
| Interval: &monitoring_pb.TimeInterval{ | |
| StartTime: getTimestamp(dist.resetTimestamp), | |
| EndTime: getTimestamp(dist.timestamp), | |
| }, | |
| Value: &monitoring_pb.TypedValue{ | |
| Value: &monitoring_pb.TypedValue_DistributionValue{DistributionValue: dp}, | |
| }, | |
| }} | |
| b.histResultsBuf = append(b.histResultsBuf, hashedSeries{hash: dist.hash, proto: &ts}) | |
| return nil, 0, samples[consumed:], err | |
| } | |
| return dp, dist.resetTimestamp, samples[consumed:], nil | |
| if !dist.complete() { | |
| continue | |
| } | |
| dp, err := dist.build(e.lset) | |
| if err != nil { | |
| delete(b.dists, e.protos.cumulative.hash) | |
| putDistribution(dist) | |
| return nil, 0, samples[consumed:], err | |
| } | |
| resetTimestamp := dist.resetTimestamp | |
| delete(b.dists, e.protos.cumulative.hash) | |
| putDistribution(dist) | |
| return dp, resetTimestamp, samples[consumed:], nil |
| if it.currTime.Add(after).After(time.Now()) { | ||
| it.t.Fatal(errors.New("sum of all scrape 'after' parameters can be beyond 10 hours")) | ||
| } |
There was a problem hiding this comment.
The error message "sum of all scrape 'after' parameters can be beyond 10 hours" is misleading. Since it.currTime is initialized to time.Now().Add(-1 * time.Hour), the maximum simulated time that can be added is 1 hour before it goes into the future. The error message should accurately reflect that the simulated time cannot go into the future.
| if it.currTime.Add(after).After(time.Now()) { | |
| it.t.Fatal(errors.New("sum of all scrape 'after' parameters can be beyond 10 hours")) | |
| } | |
| if it.currTime.Add(after).After(time.Now()) { | |
| it.t.Fatal(errors.New("sum of all scrape 'after' parameters cannot go into the future")) | |
| } |
| ctx, cancel := context.WithTimeout(t.Context(), timeout) | ||
| t.Cleanup(cancel) | ||
|
|
||
| if err := runutil.Retry(10*time.Second, ctx.Done(), func() error { |
There was a problem hiding this comment.
Polling every 10 seconds in a test retry loop is extremely slow and can significantly increase test execution times. Since the Prometheus scrape interval in the test configuration is 5 seconds, polling every 1 second is much more efficient and allows the test to complete almost immediately after the last scrape is processed.
| if err := runutil.Retry(10*time.Second, ctx.Done(), func() error { | |
| if err := runutil.Retry(1*time.Second, ctx.Done(), func() error { |
|
|
||
| var lastDiff string | ||
| var sameDiffTimes int | ||
| if err := runutil.Retry(10*time.Second, ctx.Done(), func() error { |
There was a problem hiding this comment.
Polling GCM every 10 seconds in the retry loop is very slow. Polling more frequently (e.g., every 2 seconds) would allow the test to detect completion much faster and reduce overall test suite latency.
| if err := runutil.Retry(10*time.Second, ctx.Done(), func() error { | |
| if err := runutil.Retry(2*time.Second, ctx.Done(), func() error { |
| @@ -0,0 +1,47 @@ | |||
| package config | |||
There was a problem hiding this comment.
Do we have to change file location?
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package promtest |
There was a problem hiding this comment.
Why do we duplicate the e2e test from "internal/promqle2etest"?
There was a problem hiding this comment.
Is this rename necessary?
| value *monitoring_pb.TypedValue | ||
| resetTimestamp int64 | ||
| ) | ||
| if entry.metadata.Type == model.MetricTypeHistogram { |
There was a problem hiding this comment.
We might want to upgrade to this in 3.x
| }, | ||
| }, | ||
| }, | ||
| { |
| secret := e.Object.(*corev1.Secret) | ||
| w.s = secret | ||
| secret, ok := e.Object.(*corev1.Secret) | ||
| if ok && secret != nil && secret.Name == config.Name { |
There was a problem hiding this comment.
Is this intended? Fixing some particular bug we didn't know of?
If yes - can we have a dedicated PR and test for it?
There was a problem hiding this comment.
Not critical, but there was a potential bug/flake detected. AI Summary:
The check secret.Name == config.Name in google/secrets/watch.go was added to solve a critical issue with Kubernetes Mock/Fake clients during unit tests.
The Problem with Fake Clients
Under production conditions, the watcher registers a watch with a FieldSelector limiting events to only the target secret name (fields.OneTermEqualSelector(metav1.ObjectNameField, config.Name)). The real Kubernetes API server ensures we only receive events for that specific secret.
However, in unit tests (defined in kubernetes_test.go), we use Kubernetes' fake.Clientset to mock the API server.
- The fake clientset's watch implementation does not filter events by field selectors.
- If a namespace contains multiple secrets (e.g.,
s1ands2in namespacens1), and a watcher is set up fors1, any updates tos2will still trigger a watch event insides1's watcher.
What the Code Accomplishes
If we don't have the secret.Name == config.Name check:
- When
s2is updated, the watcher fors1receives the event. - In the
updatemethod, it setsw.s = secret(assigning thes2secret object tos1's cache). - The next time the application fetches the secret value for
s1, it will query the cacheds2object, either returning a wrong value or failing with a key-not-found error.
By adding secret.Name == config.Name in both update and secret, we guarantee that:
- We ignore any unexpected events from other secrets in the same namespace (primarily during unit testing).
- We only serve the secret value if the cached secret name actually matches the name of the secret we configured.
There was a problem hiding this comment.
Great, do you mind asking AI to generate test for this then?
TAG=agy CONV=2002b781-f79e-47f1-823f-25677b6b5cb8
2 participants
This is only for preview on what changed vs release-2.53.5-gmp so far against 4ac9c11 commit
Executive Summary & Stats
Syncs the
google/directory withbernot-dev/release-3.13.0-gmp. In total, 15 files changed undergoogle/, with 1,056 insertions and 982 deletions:google/{ => export}/config/config.go0(100% similarity move)google/export/export.go+23 / -28google/export/transform.go+60 / -91google/export/export_test.go+33 / -33google/export/series_cache_test.go+7 / -7google/export/export_bench_test.go-195google/export/transform_test.go+8 / -618google/export/gcm/export_gcm_test.go+117google/export/gcm/promtest/local_export.go+227google/export/gcm/promtest/prometheus.go+188google/export/gcm/promtest/promtest.go+332google/export/gcm/promtest/skip.go+43google/internal/promqle2etest/backend_export_gcm.go+4 / -4google/internal/promqle2etest/Makefile+1 / -1google/secrets/watch.go+12 / -6Architectural Breakdown of Changes
1. Prometheus v3 Upstream API Migration
Between Prometheus 2.53.x and 3.13.x, upstream API signatures and struct definitions changed:
MetricMetadataField Rename: The fieldMetric stringinMetricMetadatawas renamed toMetricFamily stringto align with OpenMetrics / Prometheus v3 terminology.parser.ParseMetricSelector(s)inMatchers.Setwas migrated toparser.NewParser(parser.Options{}).ParseMetricSelector(s).textparse.New(...)andtextparse.Parsermethod calls inbackend_export_gcm.gowere updated to accepttextparse.ParserOptions{}andtp.Labels(&lset)instead oftp.Metric(&lset).2. Package Organization (
configmove)google/config/config.gowas moved intogoogle/export/config/config.goto cleanly namespace the Google Cloud Monitoring (GCM) exporter configuration under theexportpackage.3. Exporter Disable No-Op Guard (
export.go)Exporter.ApplyConfig(cfg *config.Config)(if e.opts.Disable { return nil }), preventing unnecessary lock acquisition and runtime re-initialization when GCM export is disabled via flags.4. Histogram Distribution Processing Refactor (
transform.go)buildDistributionsintobuildDistribution. Instead of accumulating all completed histogram series (histResultsBuf) across an entire batch insidetransform.go,buildDistributionnow returns upon completing the first full histogram distribution along with itsresetTimestamp.hash,proto, andlsetfields from the internaldistributioncache object, reducing heap allocations during high-throughput histogram scraping.next()Loop: ThesampleBuilder.next()method now directly checks ifbuildDistributionproduced a distribution value and wraps it into amonitoring_pb.TypedValue_DistributionValueon demand.5. Kubernetes Secret Watcher Safety (
watch.go)secretWatcher.updateandsecret()methods ingoogle/secrets/watch.goto explicitly verifysecret.Name == config.Name. This ensures updates or errors from unrelated secrets in the same namespace do not corrupt or overwrite the cached secret value.6. Test Suite & E2E Reorganization (
google/export/gcm/)export_bench_test.goand 618 lines oftransform_test.go) were removed and replaced with a dedicated, robust end-to-end testing framework undergoogle/export/gcm/(export_gcm_test.goandpromtest/*).TAG=agy
CONV=2002b781-f79e-47f1-823f-25677b6b5cb8