Unbounded concurrent Monitoring API requests exhaust memory and OOM-kill the container #537

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.

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)).

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)
+}

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,13 +85,16 @@ 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,
 		service:               service,
 		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)

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()