diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index c6dde002..eb487a8f 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" @@ -61,6 +62,10 @@ type MonitoringCollector struct { descriptorCache DescriptorCache enableSystemLabels bool userLabelsOverride bool + timeSeriesListConcurrency int + timeSeriesListRate *rate.Limiter + timeSeriesListRateAllowedMetric prometheus.Counter + timeSeriesListRateWaitedMetric prometheus.Counter deduplicator *MetricDeduplicator // Metrics for tracking dropped data @@ -96,6 +101,17 @@ 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 + // 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 { @@ -146,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, @@ -207,6 +243,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{} @@ -240,6 +282,10 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv descriptorCache: descriptorCache, enableSystemLabels: opts.EnableSystemLabels, userLabelsOverride: opts.UserLabelsOverride, + timeSeriesListConcurrency: opts.TimeSeriesListConcurrency, + timeSeriesListRate: timeSeriesListRate, + timeSeriesListRateAllowedMetric: timeSeriesListRateAllowedMetric, + timeSeriesListRateWaitedMetric: timeSeriesListRateWaitedMetric, deduplicator: NewMetricDeduplicator(logger, projectID), droppedMetricsTotal: droppedMetricsTotal, } @@ -249,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) @@ -270,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) @@ -288,7 +338,15 @@ func (c *MonitoringCollector) Collect(ch chan<- prometheus.Metric) { } func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metric, begun time.Time) error { - metricDescriptorsFunction := func(descriptors []*monitoring.MetricDescriptor) 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(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 @@ -314,8 +372,12 @@ 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{}{} + 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 { @@ -354,6 +416,18 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri IntervalEndTime(endTime.Format(time.RFC3339Nano)) for { + if c.timeSeriesListRate != nil { + 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() page, err := timeSeriesListCall.Do() if err != nil { @@ -374,7 +448,7 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri } timeSeriesListCall.PageToken(page.NextPageToken) } - }(metricDescriptor, ch, startTime, endTime) + }(ctx, metricDescriptor, ch, startTime, endTime) } wg.Wait() @@ -402,7 +476,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 { @@ -411,7 +485,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 new file mode 100644 index 00000000..f7e0d718 --- /dev/null +++ b/collectors/monitoring_collector_concurrency_test.go @@ -0,0 +1,214 @@ +// 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/prometheus/client_golang/prometheus/testutil" + "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) { + // 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 + + 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 < 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) + 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: prefixes, + 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(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") + + // 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") +} 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=