diff --git a/README.md b/README.md index 7623a0b1..bf6afdd4 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,7 @@ If you are still using the legacy [Access scopes][access-scopes], the `https://w | `monitoring.aggregate-deltas` | No | | If enabled will treat all DELTA metrics as an in-memory counter instead of a gauge. Be sure to read [what to know about aggregating DELTA metrics](#what-to-know-about-aggregating-delta-metrics) | | `monitoring.aggregate-deltas-ttl` | No | `30m` | How long should a delta metric continue to be exported and stored after GCP stops producing it. Read [slow moving metrics](#slow-moving-metrics) to understand the problem this attempts to solve | | `monitoring.descriptor-cache-ttl` | No | `0s` | How long should the metric descriptors for a prefixed be cached for | +| `monitoring.max-concurrency` | No | `0` | Maximum number of concurrent Monitoring API time series requests across all projects. `0` means unbounded. Set this when `google.projects.filter` or a long `google.project-ids` list resolves to many projects, to bound memory usage. | | `stackdriver.max-retries` | No | `0` | Max number of retries that should be attempted on 503 errors from stackdriver. | | `stackdriver.http-timeout` | No | `10s` | How long should stackdriver_exporter wait for a result from the Stackdriver API. | | `stackdriver.max-backoff=` | No | | Max time between each request in an exp backoff scenario. | @@ -188,6 +189,29 @@ stackdriver_exporter \ --google.projects.filter='labels.monitoring="true"' ``` +### Limiting concurrency for many-project setups + +When `google.projects.filter` (or a long, repeated `google.project-ids`) resolves to many +projects, each scrape fetches metrics for every metric descriptor of every project +concurrently, with no limit by default. In memory-constrained environments (e.g. a GKE pod +with a small CPU/memory limit), this can spawn far more concurrent Monitoring API requests +than the container can actually service at once and lead to OOM kills. + +Use `monitoring.max-concurrency` to cap the number of concurrent Monitoring API time series +requests across **all** projects in a single scrape: + +``` +stackdriver_exporter \ + --google.projects.filter='labels.monitoring="true"' \ + --monitoring.metrics-prefixes='compute.googleapis.com/instance/cpu' \ + --monitoring.max-concurrency=20 +``` + +This is a single, process-wide limit shared across every resolved project, so it bounds +memory regardless of how many projects the filter matches. It defaults to `0` (unbounded, +matching prior behavior); start with a value in the `10`-`30` range and adjust based on your +pod's CPU/memory limits and how many projects/metric prefixes you're scraping. + ### Filtering enabled collectors The `stackdriver_exporter` collects all metrics type prefixes by default. diff --git a/collectors/monitoring_collector.go b/collectors/monitoring_collector.go index 7d5b138d..e2e83d13 100644 --- a/collectors/monitoring_collector.go +++ b/collectors/monitoring_collector.go @@ -82,6 +82,22 @@ type MonitoringCollector struct { histogramStore DeltaHistogramStore aggregateDeltas bool descriptorCache DescriptorCache + requestLimiter chan struct{} +} + +// acquireRequestLimiter blocks until a slot is available in sem. A nil sem +// means unlimited concurrency and never blocks. +func acquireRequestLimiter(sem chan struct{}) { + if sem != nil { + sem <- struct{}{} + } +} + +// releaseRequestLimiter releases a slot acquired via acquireRequestLimiter. +func releaseRequestLimiter(sem chan struct{}) { + if sem != nil { + <-sem + } } type MonitoringCollectorOptions struct { @@ -144,7 +160,7 @@ type DeltaHistogramStore interface { ListMetrics(metricDescriptorName string) []*HistogramMetric } -func NewMonitoringCollector(projectID string, monitoringService *monitoring.Service, opts MonitoringCollectorOptions, logger *slog.Logger, counterStore DeltaCounterStore, histogramStore DeltaHistogramStore) (*MonitoringCollector, error) { +func NewMonitoringCollector(projectID string, monitoringService *monitoring.Service, opts MonitoringCollectorOptions, logger *slog.Logger, counterStore DeltaCounterStore, histogramStore DeltaHistogramStore, requestLimiter chan struct{}) (*MonitoringCollector, error) { const subsystem = "monitoring" logger = logger.With("project_id", projectID) @@ -240,6 +256,7 @@ func NewMonitoringCollector(projectID string, monitoringService *monitoring.Serv histogramStore: histogramStore, aggregateDeltas: opts.AggregateDeltas, descriptorCache: descriptorCache, + requestLimiter: requestLimiter, } return monitoringCollector, nil @@ -339,6 +356,9 @@ func (c *MonitoringCollector) reportMonitoringMetrics(ch chan<- prometheus.Metri c.logger.Debug("retrieving Google Stackdriver Monitoring metrics with filter", "filter", filter) + acquireRequestLimiter(c.requestLimiter) + defer releaseRequestLimiter(c.requestLimiter) + timeSeriesListCall := c.monitoringService.Projects.TimeSeries.List(projectResource(c.projectID)). Filter(filter). IntervalStartTime(startTime.Format(time.RFC3339Nano)). diff --git a/collectors/monitoring_collector_test.go b/collectors/monitoring_collector_test.go index 9e391ef2..fc8e3265 100644 --- a/collectors/monitoring_collector_test.go +++ b/collectors/monitoring_collector_test.go @@ -14,8 +14,21 @@ package collectors import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" "reflect" + "strings" + "sync" "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "google.golang.org/api/monitoring/v3" + "google.golang.org/api/option" ) func TestIsGoogleMetric(t *testing.T) { @@ -113,3 +126,174 @@ func TestProjectResource(t *testing.T) { t.Fatalf("projectResource() = %q, want %q", got, "projects/fake-project-1") } } + +func TestAcquireReleaseRequestLimiterNilIsUnbounded(t *testing.T) { + t.Parallel() + + done := make(chan struct{}) + go func() { + defer close(done) + acquireRequestLimiter(nil) + releaseRequestLimiter(nil) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("acquire/releaseRequestLimiter blocked on a nil limiter") + } +} + +func TestAcquireRequestLimiterBlocksWhenFull(t *testing.T) { + t.Parallel() + + sem := make(chan struct{}, 1) + acquireRequestLimiter(sem) + + acquired := make(chan struct{}) + go func() { + acquireRequestLimiter(sem) + close(acquired) + }() + + select { + case <-acquired: + t.Fatal("second acquireRequestLimiter succeeded while limiter was full") + case <-time.After(50 * time.Millisecond): + } + + releaseRequestLimiter(sem) + + select { + case <-acquired: + case <-time.After(time.Second): + t.Fatal("acquireRequestLimiter did not unblock after releaseRequestLimiter") + } +} + +// TestRequestLimiterBoundsConcurrency verifies that a non-nil requestLimiter +// caps the number of concurrent TimeSeries.List requests a single Collect +// call can have in flight, regardless of how many metric descriptors are +// being fetched. This guards against the unbounded per-descriptor goroutine +// fan-out (one HTTP request + JSON decode per descriptor) that can spike +// memory enough to OOM the process when a project has many metric +// descriptors. +func TestRequestLimiterBoundsConcurrency(t *testing.T) { + const ( + numDescriptors = 6 + limit = 2 + ) + + var ( + mu sync.Mutex + inFlight int + maxInFlight int + ) + + mux := http.NewServeMux() + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.Contains(r.URL.Path, "metricDescriptors"): + descriptors := make([]*monitoring.MetricDescriptor, 0, numDescriptors) + for i := 0; i < numDescriptors; i++ { + descriptors = append(descriptors, &monitoring.MetricDescriptor{ + Type: "custom.googleapis.com/metric_" + string(rune('a'+i)), + }) + } + writeJSONResponse(w, &monitoring.ListMetricDescriptorsResponse{MetricDescriptors: descriptors}) + + case strings.Contains(r.URL.Path, "timeSeries"): + mu.Lock() + inFlight++ + if inFlight > maxInFlight { + maxInFlight = inFlight + } + mu.Unlock() + + // Hold the request open briefly so concurrent requests overlap + // long enough for the test to observe them. + time.Sleep(50 * time.Millisecond) + + mu.Lock() + inFlight-- + mu.Unlock() + + writeJSONResponse(w, &monitoring.ListTimeSeriesResponse{}) + + default: + http.NotFound(w, r) + } + }) + + server := httptest.NewServer(mux) + defer server.Close() + + ctx := context.Background() + service, err := monitoring.NewService(ctx, + option.WithHTTPClient(server.Client()), + option.WithEndpoint(server.URL), + option.WithoutAuthentication(), + ) + if err != nil { + t.Fatalf("failed to create monitoring service: %v", err) + } + + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + limiter := make(chan struct{}, limit) + + collector, err := NewMonitoringCollector( + "test-project", + service, + MonitoringCollectorOptions{ + MetricTypePrefixes: []string{"custom.googleapis.com"}, + RequestInterval: 5 * time.Minute, + DescriptorCacheTTL: 0, + }, + logger, + noopCounterStore{}, + noopHistogramStore{}, + limiter, + ) + if err != nil { + t.Fatalf("failed to create collector: %v", err) + } + + ch := make(chan prometheus.Metric, 100) + done := make(chan struct{}) + go func() { + defer close(done) + for range ch { + } + }() + + collector.Collect(ch) + close(ch) + <-done + + if maxInFlight > limit { + t.Fatalf("observed %d concurrent TimeSeries.List requests, want <= %d", maxInFlight, limit) + } + if maxInFlight < limit { + t.Fatalf("expected concurrency to reach the configured limit %d, got max observed %d; test may not be exercising real contention", limit, maxInFlight) + } +} + +// noopCounterStore and noopHistogramStore are minimal stand-ins for +// DeltaCounterStore/DeltaHistogramStore. The real implementations live in +// the delta package, which imports collectors, so they can't be used here +// without a circular import; a real implementation isn't needed since these +// tests leave AggregateDeltas disabled. +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 } + +func writeJSONResponse(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(v) +} diff --git a/collectors/runtime.go b/collectors/runtime.go index f07c0e9d..1dd8dd60 100644 --- a/collectors/runtime.go +++ b/collectors/runtime.go @@ -44,6 +44,7 @@ type Runtime struct { counterStoreFactory CounterStoreFactory histogramStoreFactory HistogramStoreFactory cache *collectorCache + requestLimiter chan struct{} } // NewRuntime resolves project IDs and creates the monitoring service. The @@ -84,6 +85,8 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co return nil, err } + requestLimiter := newRequestLimiter(cfg.MaxConcurrentRequests) + return &Runtime{ cfg: cfg, projectIDs: projectIDs, @@ -91,6 +94,7 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co logger: logger, counterStoreFactory: counterFactory, histogramStoreFactory: histogramFactory, + requestLimiter: requestLimiter, }, nil } @@ -161,6 +165,7 @@ func (r *Runtime) newCollector(projectID string, prefixFilter []string) (*Monito r.logger, r.counterStoreFactory(r.logger, r.cfg.AggregateDeltasTTL), r.histogramStoreFactory(r.logger, r.cfg.AggregateDeltasTTL), + r.requestLimiter, ) } @@ -198,6 +203,16 @@ func collectorCacheTTL(cfg *config.Config) time.Duration { return 2 * time.Hour } +// newRequestLimiter returns a semaphore channel with capacity limit, or nil +// if limit is not positive. A nil channel means unbounded concurrency to +// acquireRequestLimiter/releaseRequestLimiter. +func newRequestLimiter(limit int) chan struct{} { + if limit <= 0 { + return nil + } + return make(chan struct{}, limit) +} + func deduplicateProjectIDs(projectIDs []string) []string { normalized := slices.Clone(projectIDs) slices.Sort(normalized) diff --git a/collectors/runtime_test.go b/collectors/runtime_test.go index cdf2327b..a67993dc 100644 --- a/collectors/runtime_test.go +++ b/collectors/runtime_test.go @@ -144,6 +144,38 @@ func TestParseMetricTypePrefixes(t *testing.T) { } } +func TestNewRequestLimiter(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + limit int + wantCap int + wantNil bool + }{ + {name: "zero means unbounded", limit: 0, wantNil: true}, + {name: "negative means unbounded", limit: -1, wantNil: true}, + {name: "positive limit sizes the channel", limit: 5, wantCap: 5}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + got := newRequestLimiter(tt.limit) + if tt.wantNil { + if got != nil { + t.Fatalf("newRequestLimiter(%d) = %v, want nil", tt.limit, got) + } + return + } + if cap(got) != tt.wantCap { + t.Fatalf("newRequestLimiter(%d) cap = %d, want %d", tt.limit, cap(got), tt.wantCap) + } + }) + } +} + func TestRuntimeFilterMetricTypePrefixes(t *testing.T) { t.Parallel() diff --git a/config/config.go b/config/config.go index e6301474..8a47c9c6 100644 --- a/config/config.go +++ b/config/config.go @@ -23,20 +23,21 @@ import ( ) const ( - DefaultUniverseDomain = "googleapis.com" - DefaultMaxRetries = 0 - DefaultHTTPTimeout = 10 * time.Second - DefaultMaxBackoff = 5 * time.Second - DefaultBackoffJitter = 1 * time.Second - DefaultMetricsInterval = 5 * time.Minute - DefaultMetricsOffset = 0 * time.Second - DefaultMetricsIngest = false - DefaultFillMissing = true - DefaultDropDelegated = false - DefaultAggregateDeltas = false - DefaultDeltasTTL = 30 * time.Minute - DefaultDescriptorTTL = 0 * time.Second - DefaultDescriptorGoogleOnly = true + DefaultUniverseDomain = "googleapis.com" + DefaultMaxRetries = 0 + DefaultHTTPTimeout = 10 * time.Second + DefaultMaxBackoff = 5 * time.Second + DefaultBackoffJitter = 1 * time.Second + DefaultMetricsInterval = 5 * time.Minute + DefaultMetricsOffset = 0 * time.Second + DefaultMetricsIngest = false + DefaultFillMissing = true + DefaultDropDelegated = false + DefaultAggregateDeltas = false + DefaultDeltasTTL = 30 * time.Minute + DefaultDescriptorTTL = 0 * time.Second + DefaultDescriptorGoogleOnly = true + DefaultMaxConcurrentRequests = 0 ) // DefaultRetryStatuses must be treated as immutable after declaration. @@ -62,6 +63,7 @@ type Config struct { AggregateDeltasTTL time.Duration DescriptorCacheTTL time.Duration DescriptorCacheOnlyGoogle bool + MaxConcurrentRequests int // validated is set by Validate on success. validated bool @@ -87,6 +89,7 @@ func NewConfigWithDefaults() *Config { AggregateDeltasTTL: DefaultDeltasTTL, DescriptorCacheTTL: DefaultDescriptorTTL, DescriptorCacheOnlyGoogle: DefaultDescriptorGoogleOnly, + MaxConcurrentRequests: DefaultMaxConcurrentRequests, } } diff --git a/config/config_test.go b/config/config_test.go index 07b44e34..1331b0f3 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -72,6 +72,9 @@ func TestNewConfigWithDefaults(t *testing.T) { if len(c.RetryStatuses) != len(DefaultRetryStatuses) || c.RetryStatuses[0] != DefaultRetryStatuses[0] { t.Errorf("RetryStatuses = %v, want %v", c.RetryStatuses, DefaultRetryStatuses) } + if c.MaxConcurrentRequests != DefaultMaxConcurrentRequests { + t.Errorf("MaxConcurrentRequests = %v, want %v", c.MaxConcurrentRequests, DefaultMaxConcurrentRequests) + } c.RetryStatuses[0] = 999 if DefaultRetryStatuses[0] == 999 { t.Fatal("NewConfigWithDefaults did not copy RetryStatuses; default mutated") diff --git a/stackdriver_exporter.go b/stackdriver_exporter.go index fa7cda26..0fa78632 100644 --- a/stackdriver_exporter.go +++ b/stackdriver_exporter.go @@ -136,6 +136,10 @@ var ( monitoringDescriptorCacheOnlyGoogle = kingpin.Flag( "monitoring.descriptor-cache-only-google", "Only cache descriptors for *.googleapis.com metrics", ).Default(strconv.FormatBool(config.DefaultDescriptorGoogleOnly)).Bool() + + monitoringMaxConcurrency = kingpin.Flag( + "monitoring.max-concurrency", "Maximum number of concurrent Monitoring API time series requests across all projects. 0 means unbounded.", + ).Default(strconv.Itoa(config.DefaultMaxConcurrentRequests)).Int() ) func init() { @@ -343,6 +347,7 @@ func collectorConfigFromFlags() *config.Config { AggregateDeltasTTL: *monitoringMetricsDeltasTTL, DescriptorCacheTTL: *monitoringDescriptorCacheTTL, DescriptorCacheOnlyGoogle: *monitoringDescriptorCacheOnlyGoogle, + MaxConcurrentRequests: *monitoringMaxConcurrency, } }