From 42a95b930ac4aa7a4da4638c7a4aecb256e1a363 Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Tue, 30 Jun 2026 11:50:51 +0000 Subject: [PATCH 1/5] Add bounded concurrency for timeSeries.list fan-out GCP metrics scrapes fanned out one unbounded goroutine per metric descriptor, each issuing a paginated projects.timeSeries.list call. On high-cardinality projects this bursts dozens of simultaneous calls, tripping GCP soft-throttling (slow empty 200s) and stalling scrapes to the 240s context deadline. Add TimeSeriesListConcurrency to MonitoringCollectorOptions and bound the inner descriptor fan-out with a semaphore. A value <= 0 means unbounded, preserving previous behavior. Pagination, ingest-delay, filtering, and error accumulation are unchanged. Co-Authored-By: Claude Opus 4.8 --- collectors/monitoring_collector.go | 17 +++ .../monitoring_collector_concurrency_test.go | 113 ++++++++++++++++++ 2 files changed, 130 insertions(+) create mode 100644 collectors/monitoring_collector_concurrency_test.go diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index c6dde002..60dbe7f3 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -61,6 +61,7 @@ type MonitoringCollector struct { descriptorCache DescriptorCache enableSystemLabels bool userLabelsOverride bool + timeSeriesListConcurrency int deduplicator *MetricDeduplicator // Metrics for tracking dropped data @@ -96,6 +97,10 @@ type MonitoringCollectorOptions struct { EnableSystemLabels bool // UserLabelsOverride decides if user labels should override any conflicting labels UserLabelsOverride bool + // TimeSeriesListConcurrency bounds the number of concurrent projects.timeSeries.list calls + // (one per metric descriptor) issued within a single scrape. A value <= 0 means unbounded, + // preserving the previous behavior. + TimeSeriesListConcurrency int } func isGoogleMetric(name string) bool { @@ -240,6 +245,7 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv descriptorCache: descriptorCache, enableSystemLabels: opts.EnableSystemLabels, userLabelsOverride: opts.UserLabelsOverride, + timeSeriesListConcurrency: opts.TimeSeriesListConcurrency, deduplicator: NewMetricDeduplicator(logger, projectID), droppedMetricsTotal: droppedMetricsTotal, } @@ -312,10 +318,21 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri endTime := time.Now().UTC().Add(c.metricsOffset * -1) startTime := endTime.Add(c.metricsInterval * -1) + // sem bounds the number of concurrent timeSeries.list fan-out goroutines. A nil channel + // (concurrency <= 0) disables the limit, preserving the previous unbounded behavior. + var sem chan struct{} + if c.timeSeriesListConcurrency > 0 { + sem = make(chan struct{}, c.timeSeriesListConcurrency) + } + for _, metricDescriptor := range uniqueDescriptors { wg.Add(1) go func(metricDescriptor *monitoring.MetricDescriptor, ch chan<- prometheus.Metric, startTime, endTime time.Time) { defer wg.Done() + if sem != nil { + sem <- struct{}{} + defer func() { <-sem }() + } c.logger.Debug("retrieving Google Stackdriver Monitoring metrics for descriptor", "descriptor", metricDescriptor.Type) filter := fmt.Sprintf("metric.type=\"%s\"", metricDescriptor.Type) if c.monitoringDropDelegatedProjects { diff --git a/collectors/monitoring_collector_concurrency_test.go b/collectors/monitoring_collector_concurrency_test.go new file mode 100644 index 00000000..21c86174 --- /dev/null +++ b/collectors/monitoring_collector_concurrency_test.go @@ -0,0 +1,113 @@ +// Copyright 2020 The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package collectors + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/require" + "google.golang.org/api/monitoring/v3" + "google.golang.org/api/option" +) + +type noopCounterStore struct{} + +func (noopCounterStore) Increment(*monitoring.MetricDescriptor, *ConstMetric) {} +func (noopCounterStore) ListMetrics(string) []*ConstMetric { return nil } + +type noopHistogramStore struct{} + +func (noopHistogramStore) Increment(*monitoring.MetricDescriptor, *HistogramMetric) {} +func (noopHistogramStore) ListMetrics(string) []*HistogramMetric { return nil } + +// TestReportMonitoringMetrics_TimeSeriesListConcurrency verifies that the inner +// projects.timeSeries.list fan-out is bounded by TimeSeriesListConcurrency. +func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { + const numDescriptors = 8 + const limit = 2 + + var inFlight, maxInFlight, total int64 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/metricDescriptors"): + var sb strings.Builder + sb.WriteString(`{"metricDescriptors":[`) + for i := 0; i < numDescriptors; i++ { + if i > 0 { + sb.WriteString(",") + } + fmt.Fprintf(&sb, `{"type":"compute.googleapis.com/m%d","metricKind":"GAUGE","valueType":"DOUBLE"}`, i) + } + sb.WriteString(`]}`) + w.Write([]byte(sb.String())) + case strings.HasSuffix(r.URL.Path, "/timeSeries"): + atomic.AddInt64(&total, 1) + cur := atomic.AddInt64(&inFlight, 1) + for { + old := atomic.LoadInt64(&maxInFlight) + if cur <= old || atomic.CompareAndSwapInt64(&maxInFlight, old, cur) { + break + } + } + time.Sleep(20 * time.Millisecond) + atomic.AddInt64(&inFlight, -1) + w.Write([]byte(`{}`)) + default: + w.Write([]byte(`{}`)) + } + })) + defer server.Close() + + svc, err := monitoring.NewService(context.Background(), + option.WithEndpoint(server.URL), + option.WithoutAuthentication(), + option.WithHTTPClient(server.Client())) + require.NoError(t, err) + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) + collector, err := NewMonitoringCollector("test-project", svc, MonitoringCollectorOptions{ + MetricTypePrefixes: []string{"compute.googleapis.com"}, + RequestInterval: time.Minute, + TimeSeriesListConcurrency: limit, + }, logger, noopCounterStore{}, noopHistogramStore{}) + require.NoError(t, err) + + ch := make(chan prometheus.Metric, 1024) + var drain sync.WaitGroup + drain.Add(1) + go func() { + defer drain.Done() + for range ch { + } + }() + + require.NoError(t, collector.reportMonitoringMetrics(ch, time.Now())) + close(ch) + drain.Wait() + + require.Equal(t, int64(numDescriptors), atomic.LoadInt64(&total), "every descriptor should be queried") + require.LessOrEqual(t, atomic.LoadInt64(&maxInFlight), int64(limit), "concurrent timeSeries.list calls must not exceed the limit") +} From 193d01b2ffc0dbcb05fe4a8f5aafe5372b294a40 Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Mon, 6 Jul 2026 06:57:48 +0000 Subject: [PATCH 2/5] Make timeSeries.list concurrency cap scrape-wide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The semaphore was created inside metricDescriptorsFunction, which runs once per metric-type prefix, so each prefix got its own limiter of size N. With P prefixes fanning out concurrently the effective cap was N*P, not the intended scrape-wide N — undercutting the fix for the GCP soft-throttle stalls. Create the semaphore once in reportMonitoringMetrics and share it across all prefix goroutines and descriptor batches. Update the concurrency test to use multiple prefixes; it now fails on the old per-prefix limiter and passes on the shared one. Co-Authored-By: Claude Opus 4.8 (1M context) --- collectors/monitoring_collector.go | 15 ++++++------ .../monitoring_collector_concurrency_test.go | 24 +++++++++++++++---- 2 files changed, 27 insertions(+), 12 deletions(-) diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index 60dbe7f3..f52857a0 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -294,6 +294,14 @@ func (c *MonitoringCollector) Collect(ch chan<- prometheus.Metric) { } func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metric, begun time.Time) error { + // sem bounds the number of concurrent timeSeries.list fan-out goroutines across the whole + // scrape (all metric-type prefixes and their descriptor batches share it). A nil channel + // (concurrency <= 0) disables the limit, preserving the previous unbounded behavior. + var sem chan struct{} + if c.timeSeriesListConcurrency > 0 { + sem = make(chan struct{}, c.timeSeriesListConcurrency) + } + metricDescriptorsFunction := func(descriptors []*monitoring.MetricDescriptor) error { var wg = &sync.WaitGroup{} @@ -318,13 +326,6 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri endTime := time.Now().UTC().Add(c.metricsOffset * -1) startTime := endTime.Add(c.metricsInterval * -1) - // sem bounds the number of concurrent timeSeries.list fan-out goroutines. A nil channel - // (concurrency <= 0) disables the limit, preserving the previous unbounded behavior. - var sem chan struct{} - if c.timeSeriesListConcurrency > 0 { - sem = make(chan struct{}, c.timeSeriesListConcurrency) - } - for _, metricDescriptor := range uniqueDescriptors { wg.Add(1) go func(metricDescriptor *monitoring.MetricDescriptor, ch chan<- prometheus.Metric, startTime, endTime time.Time) { diff --git a/collectors/monitoring_collector_concurrency_test.go b/collectors/monitoring_collector_concurrency_test.go index 21c86174..307ed272 100644 --- a/collectors/monitoring_collector_concurrency_test.go +++ b/collectors/monitoring_collector_concurrency_test.go @@ -45,7 +45,11 @@ func (noopHistogramStore) ListMetrics(string) []*HistogramMetric // TestReportMonitoringMetrics_TimeSeriesListConcurrency verifies that the inner // projects.timeSeries.list fan-out is bounded by TimeSeriesListConcurrency. func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { - const numDescriptors = 8 + // Two prefixes fan out concurrently. With a scrape-wide limiter, total in-flight + // timeSeries.list calls must stay <= limit; a per-prefix limiter would allow up to + // limit*len(prefixes), so this guards against the semaphore leaking per prefix. + prefixes := []string{"compute.googleapis.com", "storage.googleapis.com"} + const descriptorsPerPrefix = 8 const limit = 2 var inFlight, maxInFlight, total int64 @@ -53,13 +57,23 @@ func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case strings.HasSuffix(r.URL.Path, "/metricDescriptors"): + // Namespace descriptor types by the requested prefix so each prefix batch is + // distinct (dedup happens per type) and produces its own timeSeries.list calls. + filter := r.URL.Query().Get("filter") + prefix := filter + if i := strings.Index(filter, `starts_with("`); i >= 0 { + rest := filter[i+len(`starts_with("`):] + if j := strings.Index(rest, `"`); j >= 0 { + prefix = rest[:j] + } + } var sb strings.Builder sb.WriteString(`{"metricDescriptors":[`) - for i := 0; i < numDescriptors; i++ { + for i := 0; i < descriptorsPerPrefix; i++ { if i > 0 { sb.WriteString(",") } - fmt.Fprintf(&sb, `{"type":"compute.googleapis.com/m%d","metricKind":"GAUGE","valueType":"DOUBLE"}`, i) + fmt.Fprintf(&sb, `{"type":"%s/m%d","metricKind":"GAUGE","valueType":"DOUBLE"}`, prefix, i) } sb.WriteString(`]}`) w.Write([]byte(sb.String())) @@ -89,7 +103,7 @@ func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) collector, err := NewMonitoringCollector("test-project", svc, MonitoringCollectorOptions{ - MetricTypePrefixes: []string{"compute.googleapis.com"}, + MetricTypePrefixes: prefixes, RequestInterval: time.Minute, TimeSeriesListConcurrency: limit, }, logger, noopCounterStore{}, noopHistogramStore{}) @@ -108,6 +122,6 @@ func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { close(ch) drain.Wait() - require.Equal(t, int64(numDescriptors), atomic.LoadInt64(&total), "every descriptor should be queried") + require.Equal(t, int64(descriptorsPerPrefix*len(prefixes)), atomic.LoadInt64(&total), "every descriptor should be queried") require.LessOrEqual(t, atomic.LoadInt64(&maxInFlight), int64(limit), "concurrent timeSeries.list calls must not exceed the limit") } From 227482a52fd09746a0d000c7b8e9a5e9912cf314 Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Mon, 6 Jul 2026 10:05:13 +0000 Subject: [PATCH 3/5] Add rate limiter for timeSeries.list calls A concurrency semaphore caps simultaneous calls but not the rate over a window. GCP soft-throttles on per-second admission (54 calls in ~284ms), so capping concurrency alone leaves the burst intact once calls return quickly. Add a token-bucket rate limiter (golang.org/x/time/rate) over the timeSeries.list fan-out, mirroring the CloudWatch/YACE GlobalRateLimiter pattern. Config is Count-per-Duration (burst = Count); both <= 0 disables it, preserving current behavior. The limiter is per-collector, i.e. per-project, and each timeSeries.list page call waits on it before firing. Concurrency and rate limits are independent knobs. Add a test asserting the scrape cannot complete faster than the token-bucket pacing allows. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 + collectors/monitoring_collector.go | 33 ++++++-- .../monitoring_collector_concurrency_test.go | 82 +++++++++++++++++++ go.mod | 1 + go.sum | 2 + 5 files changed, 114 insertions(+), 5 deletions(-) diff --git a/.gitignore b/.gitignore index ff751a42..35f2e8c4 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ *.test *-stamp /vendor +.omc/ diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index f52857a0..f47e7231 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -25,6 +25,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/tidwall/gjson" "golang.org/x/net/context" + "golang.org/x/time/rate" "google.golang.org/api/googleapi" "google.golang.org/api/monitoring/v3" @@ -62,6 +63,7 @@ type MonitoringCollector struct { enableSystemLabels bool userLabelsOverride bool timeSeriesListConcurrency int + timeSeriesListRate *rate.Limiter deduplicator *MetricDeduplicator // Metrics for tracking dropped data @@ -101,6 +103,13 @@ type MonitoringCollectorOptions struct { // (one per metric descriptor) issued within a single scrape. A value <= 0 means unbounded, // preserving the previous behavior. TimeSeriesListConcurrency int + // TimeSeriesListRateCount / TimeSeriesListRateDuration bound the *rate* of + // projects.timeSeries.list calls to Count calls per Duration (a token bucket with burst = + // Count). Both must be > 0 to enable; otherwise no rate limiting is applied. This smooths the + // fan-out burst that trips GCP per-second soft-throttling, complementing (and independent of) + // TimeSeriesListConcurrency. + TimeSeriesListRateCount int + TimeSeriesListRateDuration time.Duration } func isGoogleMetric(name string) bool { @@ -212,6 +221,12 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv []string{"reason", "metric_type", "resource_type", "metric_kind", "value_type"}, ) + var timeSeriesListRate *rate.Limiter + if opts.TimeSeriesListRateCount > 0 && opts.TimeSeriesListRateDuration > 0 { + ratePerSecond := float64(opts.TimeSeriesListRateCount) / opts.TimeSeriesListRateDuration.Seconds() + timeSeriesListRate = rate.NewLimiter(rate.Limit(ratePerSecond), opts.TimeSeriesListRateCount) + } + var descriptorCache DescriptorCache if opts.DescriptorCacheTTL == 0 { descriptorCache = &noopDescriptorCache{} @@ -246,6 +261,7 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv enableSystemLabels: opts.EnableSystemLabels, userLabelsOverride: opts.UserLabelsOverride, timeSeriesListConcurrency: opts.TimeSeriesListConcurrency, + timeSeriesListRate: timeSeriesListRate, deduplicator: NewMetricDeduplicator(logger, projectID), droppedMetricsTotal: droppedMetricsTotal, } @@ -302,7 +318,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri sem = make(chan struct{}, c.timeSeriesListConcurrency) } - metricDescriptorsFunction := func(descriptors []*monitoring.MetricDescriptor) error { + metricDescriptorsFunction := func(ctx context.Context, descriptors []*monitoring.MetricDescriptor) error { var wg = &sync.WaitGroup{} // It has been noticed that the same metric descriptor can be obtained from different GCP @@ -328,7 +344,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri for _, metricDescriptor := range uniqueDescriptors { wg.Add(1) - go func(metricDescriptor *monitoring.MetricDescriptor, ch chan<- prometheus.Metric, startTime, endTime time.Time) { + go func(ctx context.Context, metricDescriptor *monitoring.MetricDescriptor, ch chan<- prometheus.Metric, startTime, endTime time.Time) { defer wg.Done() if sem != nil { sem <- struct{}{} @@ -372,6 +388,13 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri IntervalEndTime(endTime.Format(time.RFC3339Nano)) for { + if c.timeSeriesListRate != nil { + if err := c.timeSeriesListRate.Wait(ctx); err != nil { + c.logger.Error("rate limiter wait failed for descriptor", "descriptor", metricDescriptor.Type, "err", err) + errChannel <- err + break + } + } c.apiCallsTotalMetric.Inc() page, err := timeSeriesListCall.Do() if err != nil { @@ -392,7 +415,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri } timeSeriesListCall.PageToken(page.NextPageToken) } - }(metricDescriptor, ch, startTime, endTime) + }(ctx, metricDescriptor, ch, startTime, endTime) } wg.Wait() @@ -420,7 +443,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri if cached := c.descriptorCache.Lookup(metricsTypePrefix); cached != nil { c.logger.Debug("using cached Google Stackdriver Monitoring metric descriptors starting with", "prefix", metricsTypePrefix) - if err := metricDescriptorsFunction(cached); err != nil { + if err := metricDescriptorsFunction(ctx, cached); err != nil { errChannel <- err } } else { @@ -429,7 +452,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri callback := func(r *monitoring.ListMetricDescriptorsResponse) error { c.apiCallsTotalMetric.Inc() cache = append(cache, r.MetricDescriptors...) - return metricDescriptorsFunction(r.MetricDescriptors) + return metricDescriptorsFunction(ctx, r.MetricDescriptors) } c.logger.Debug("listing Google Stackdriver Monitoring metric descriptors starting with", "prefix", metricsTypePrefix) diff --git a/collectors/monitoring_collector_concurrency_test.go b/collectors/monitoring_collector_concurrency_test.go index 307ed272..0928f6c5 100644 --- a/collectors/monitoring_collector_concurrency_test.go +++ b/collectors/monitoring_collector_concurrency_test.go @@ -125,3 +125,85 @@ func TestReportMonitoringMetrics_TimeSeriesListConcurrency(t *testing.T) { require.Equal(t, int64(descriptorsPerPrefix*len(prefixes)), atomic.LoadInt64(&total), "every descriptor should be queried") require.LessOrEqual(t, atomic.LoadInt64(&maxInFlight), int64(limit), "concurrent timeSeries.list calls must not exceed the limit") } + +// TestReportMonitoringMetrics_TimeSeriesListRate verifies that the token-bucket rate limiter +// spreads projects.timeSeries.list calls over time scrape-wide. With burst=Count the first Count +// calls fire immediately and the rest are paced at Count/Duration, so the scrape cannot complete +// faster than that pacing allows — proving rate (not just concurrency) is bounded. +func TestReportMonitoringMetrics_TimeSeriesListRate(t *testing.T) { + prefixes := []string{"compute.googleapis.com", "storage.googleapis.com"} + const descriptorsPerPrefix = 8 + const rateCount = 2 + const rateDuration = 100 * time.Millisecond // 20 calls/sec, burst 2 + + var total int64 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/metricDescriptors"): + filter := r.URL.Query().Get("filter") + prefix := filter + if i := strings.Index(filter, `starts_with("`); i >= 0 { + rest := filter[i+len(`starts_with("`):] + if j := strings.Index(rest, `"`); j >= 0 { + prefix = rest[:j] + } + } + var sb strings.Builder + sb.WriteString(`{"metricDescriptors":[`) + for i := 0; i < descriptorsPerPrefix; i++ { + if i > 0 { + sb.WriteString(",") + } + fmt.Fprintf(&sb, `{"type":"%s/m%d","metricKind":"GAUGE","valueType":"DOUBLE"}`, prefix, i) + } + sb.WriteString(`]}`) + w.Write([]byte(sb.String())) + case strings.HasSuffix(r.URL.Path, "/timeSeries"): + atomic.AddInt64(&total, 1) + w.Write([]byte(`{}`)) + default: + w.Write([]byte(`{}`)) + } + })) + defer server.Close() + + svc, err := monitoring.NewService(context.Background(), + option.WithEndpoint(server.URL), + option.WithoutAuthentication(), + option.WithHTTPClient(server.Client())) + require.NoError(t, err) + + logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) + collector, err := NewMonitoringCollector("test-project", svc, MonitoringCollectorOptions{ + MetricTypePrefixes: prefixes, + RequestInterval: time.Minute, + TimeSeriesListRateCount: rateCount, + TimeSeriesListRateDuration: rateDuration, + }, logger, noopCounterStore{}, noopHistogramStore{}) + require.NoError(t, err) + + ch := make(chan prometheus.Metric, 1024) + var drain sync.WaitGroup + drain.Add(1) + go func() { + defer drain.Done() + for range ch { + } + }() + + start := time.Now() + require.NoError(t, collector.reportMonitoringMetrics(ch, time.Now())) + elapsed := time.Since(start) + close(ch) + drain.Wait() + + numCalls := descriptorsPerPrefix * len(prefixes) + // After the initial burst of rateCount, the remaining calls are paced at rateCount/rateDuration. + ratePerSecond := float64(rateCount) / rateDuration.Seconds() + minExpected := time.Duration(float64(numCalls-rateCount) / ratePerSecond * float64(time.Second)) + + require.Equal(t, int64(numCalls), atomic.LoadInt64(&total), "every descriptor should be queried") + // Allow slack below the theoretical minimum for scheduling jitter. + require.GreaterOrEqual(t, elapsed, minExpected*8/10, "rate limiter must spread calls over time") +} diff --git a/go.mod b/go.mod index 1b22ce22..35f9ee3e 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ require ( github.com/tidwall/gjson v1.18.0 golang.org/x/net v0.37.0 golang.org/x/oauth2 v0.28.0 + golang.org/x/time v0.10.0 google.golang.org/api v0.224.0 ) diff --git a/go.sum b/go.sum index 3d6e9c82..3dfbc89e 100644 --- a/go.sum +++ b/go.sum @@ -180,6 +180,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= +golang.org/x/time v0.10.0 h1:3usCWA8tQn0L8+hFJQNgzpWbd89begxN66o1Ojdn5L4= +golang.org/x/time v0.10.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= From 04f9f48a78a0d9c34a2de629fc9c70c3b17e539f Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Mon, 6 Jul 2026 10:09:33 +0000 Subject: [PATCH 4/5] Add allow/waited counters for timeSeries.list rate limiter The agent runs the GCP client with telemetry disabled, so rate-limit behavior was only observable via the eBPF sensor. Expose it directly: split the gate into Allow() (admitted immediately) vs Wait() (throttled) and count each with stackdriver_monitoring_time_series_list_rate_{allowed,waited}_total, mirroring the CloudWatch/YACE RateLimitAllowed/Wait counters. Extend the rate test to assert the burst is counted as allowed and the remainder as waited. Co-Authored-By: Claude Opus 4.8 (1M context) --- collectors/monitoring_collector.go | 41 +++++++++++++++++-- .../monitoring_collector_concurrency_test.go | 5 +++ 2 files changed, 42 insertions(+), 4 deletions(-) diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index f47e7231..eb487a8f 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -64,6 +64,8 @@ type MonitoringCollector struct { userLabelsOverride bool timeSeriesListConcurrency int timeSeriesListRate *rate.Limiter + timeSeriesListRateAllowedMetric prometheus.Counter + timeSeriesListRateWaitedMetric prometheus.Counter deduplicator *MetricDeduplicator // Metrics for tracking dropped data @@ -160,6 +162,26 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv }, ) + timeSeriesListRateAllowedMetric := prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "time_series_list_rate_allowed_total", + Help: "Total number of projects.timeSeries.list calls admitted immediately by the rate limiter.", + ConstLabels: prometheus.Labels{"project_id": projectID}, + }, + ) + + timeSeriesListRateWaitedMetric := prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: namespace, + Subsystem: subsystem, + Name: "time_series_list_rate_waited_total", + Help: "Total number of projects.timeSeries.list calls delayed by the rate limiter (throttled).", + ConstLabels: prometheus.Labels{"project_id": projectID}, + }, + ) + scrapesTotalMetric := prometheus.NewCounter( prometheus.CounterOpts{ Namespace: namespace, @@ -262,6 +284,8 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv userLabelsOverride: opts.UserLabelsOverride, timeSeriesListConcurrency: opts.TimeSeriesListConcurrency, timeSeriesListRate: timeSeriesListRate, + timeSeriesListRateAllowedMetric: timeSeriesListRateAllowedMetric, + timeSeriesListRateWaitedMetric: timeSeriesListRateWaitedMetric, deduplicator: NewMetricDeduplicator(logger, projectID), droppedMetricsTotal: droppedMetricsTotal, } @@ -271,6 +295,8 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv func (c *MonitoringCollector) Describe(ch chan<- *prometheus.Desc) { c.apiCallsTotalMetric.Describe(ch) + c.timeSeriesListRateAllowedMetric.Describe(ch) + c.timeSeriesListRateWaitedMetric.Describe(ch) c.scrapesTotalMetric.Describe(ch) c.scrapeErrorsTotalMetric.Describe(ch) c.lastScrapeErrorMetric.Describe(ch) @@ -292,6 +318,8 @@ func (c *MonitoringCollector) Collect(ch chan<- prometheus.Metric) { c.scrapeErrorsTotalMetric.Collect(ch) c.apiCallsTotalMetric.Collect(ch) + c.timeSeriesListRateAllowedMetric.Collect(ch) + c.timeSeriesListRateWaitedMetric.Collect(ch) c.scrapesTotalMetric.Inc() c.scrapesTotalMetric.Collect(ch) @@ -389,10 +417,15 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri for { if c.timeSeriesListRate != nil { - if err := c.timeSeriesListRate.Wait(ctx); err != nil { - c.logger.Error("rate limiter wait failed for descriptor", "descriptor", metricDescriptor.Type, "err", err) - errChannel <- err - break + if c.timeSeriesListRate.Allow() { + c.timeSeriesListRateAllowedMetric.Inc() + } else { + c.timeSeriesListRateWaitedMetric.Inc() + if err := c.timeSeriesListRate.Wait(ctx); err != nil { + c.logger.Error("rate limiter wait failed for descriptor", "descriptor", metricDescriptor.Type, "err", err) + errChannel <- err + break + } } } c.apiCallsTotalMetric.Inc() diff --git a/collectors/monitoring_collector_concurrency_test.go b/collectors/monitoring_collector_concurrency_test.go index 0928f6c5..f7e0d718 100644 --- a/collectors/monitoring_collector_concurrency_test.go +++ b/collectors/monitoring_collector_concurrency_test.go @@ -27,6 +27,7 @@ import ( "time" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" "google.golang.org/api/monitoring/v3" "google.golang.org/api/option" @@ -206,4 +207,8 @@ func TestReportMonitoringMetrics_TimeSeriesListRate(t *testing.T) { require.Equal(t, int64(numCalls), atomic.LoadInt64(&total), "every descriptor should be queried") // Allow slack below the theoretical minimum for scheduling jitter. require.GreaterOrEqual(t, elapsed, minExpected*8/10, "rate limiter must spread calls over time") + + // The initial burst (rateCount) is admitted immediately; everything after is throttled. + require.Equal(t, float64(rateCount), testutil.ToFloat64(collector.timeSeriesListRateAllowedMetric), "burst calls should be counted as allowed") + require.Equal(t, float64(numCalls-rateCount), testutil.ToFloat64(collector.timeSeriesListRateWaitedMetric), "throttled calls should be counted as waited") } From f95c4fd782a61ede1a61fdb6bd1d10bead698d61 Mon Sep 17 00:00:00 2001 From: nir-sagi Date: Mon, 6 Jul 2026 10:12:49 +0000 Subject: [PATCH 5/5] Remove .omc/ from .gitignore Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 1 - 1 file changed, 1 deletion(-) diff --git a/.gitignore b/.gitignore index 35f2e8c4..ff751a42 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,3 @@ *.test *-stamp /vendor -.omc/