From a61d4cd073716c977ce479e61598c4b4edc6d82d Mon Sep 17 00:00:00 2001 From: Pavan Nalam Date: Mon, 24 Aug 2026 16:19:28 -0700 Subject: [PATCH 1/5] fix(collectors): periodically refresh google.projects.filter to pick up new/removed projects without a restart google.projects.filter was resolved to a concrete project list once at startup, so projects added to or removed from the matching org/folder were never reflected until the exporter restarted. Adds an opt-in google.projects.filter-refresh-interval flag (default 0s, disabled) that re-evaluates the filter on a background ticker and atomically swaps the in-memory project list, and updates the HTTP handler to rebuild its collector registry per scrape so the refreshed list actually reaches /metrics. Fixes #538 Signed-off-by: Pavan Nalam --- README.md | 1 + collectors/runtime.go | 87 ++++++++++++- collectors/runtime_test.go | 237 +++++++++++++++++++++++++++++++++++ config/config.go | 31 ++--- config/config_test.go | 3 + stackdriver_exporter.go | 61 ++++----- stackdriver_exporter_test.go | 48 +++++++ 7 files changed, 419 insertions(+), 49 deletions(-) create mode 100644 stackdriver_exporter_test.go diff --git a/README.md b/README.md index 7623a0b1..b5346c48 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ If you are still using the legacy [Access scopes][access-scopes], the `https://w | ----------------------------------- | -------- |---------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `google.project-ids` | No | GCloud SDK auto-discovery | Repeatable flag of Google Project IDs | | `google.projects.filter` | No | | GCloud projects filter expression. See more [here](https://cloud.google.com/sdk/gcloud/reference/projects/list). | +| `google.projects.filter-refresh-interval` | No | `0s` (disabled) | How often to re-evaluate `google.projects.filter` to pick up projects added to or removed from the matching org/folder without restarting. `0` keeps the pre-existing behavior of resolving the filter once at startup only. | | `google.universe-domain` | No | `googleapis.com` | Target specific Google Cloud environments, such as public cloud, or specific sovereign clouds | | `monitoring.metrics-ingest-delay` | No | | Offsets metric collection by a delay appropriate for each metric type, e.g. because bigquery metrics are slow to appear | | `monitoring.drop-delegated-projects` | No | No | Drop metrics from attached projects and fetch `project_id` only. | diff --git a/collectors/runtime.go b/collectors/runtime.go index f07c0e9d..5edb14cf 100644 --- a/collectors/runtime.go +++ b/collectors/runtime.go @@ -19,6 +19,7 @@ import ( "log/slog" "slices" "strings" + "sync/atomic" "time" "golang.org/x/oauth2/google" @@ -37,13 +38,22 @@ type HistogramStoreFactory func(logger *slog.Logger, ttl time.Duration) DeltaHis // Runtime holds the resolved state produced by NewRuntime. type Runtime struct { - cfg *config.Config - projectIDs []string + cfg *config.Config + // projectIDs is a *atomic.Pointer[[]string] (rather than a plain []string) + // so that WithCache's shallow struct copy shares live state with the + // original: a background refresh (see StartProjectDiscoveryRefresh) must + // be visible through every Runtime value derived from the same NewRuntime + // call, and scrapes must be able to read it without lock contention. + projectIDs *atomic.Pointer[[]string] service *monitoring.Service logger *slog.Logger counterStoreFactory CounterStoreFactory histogramStoreFactory HistogramStoreFactory cache *collectorCache + // discoverProjectIDs resolves cfg.ProjectsFilter to project IDs. It + // defaults to getProjectIDsFromFilter; tests override it to avoid + // depending on Application Default Credentials. + discoverProjectIDs func(ctx context.Context, filter string) ([]string, error) } // NewRuntime resolves project IDs and creates the monitoring service. The @@ -84,16 +94,73 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co return nil, err } + projectIDsPtr := &atomic.Pointer[[]string]{} + projectIDsPtr.Store(&projectIDs) + return &Runtime{ cfg: cfg, - projectIDs: projectIDs, + projectIDs: projectIDsPtr, service: service, logger: logger, counterStoreFactory: counterFactory, histogramStoreFactory: histogramFactory, + discoverProjectIDs: getProjectIDsFromFilter, }, nil } +// StartProjectDiscoveryRefresh periodically re-resolves cfg.ProjectsFilter in +// the background so that projects added to or removed from the matching +// org/folder are picked up without restarting the exporter. It is a no-op +// unless both cfg.ProjectsFilter and cfg.ProjectsRefreshInterval are set, so +// existing deployments see no behavior change unless they opt in. +// +// The refresh runs until ctx is canceled; callers must cancel ctx on shutdown +// to avoid leaking the goroutine. +func (r *Runtime) StartProjectDiscoveryRefresh(ctx context.Context) { + if r.cfg.ProjectsFilter == "" || r.cfg.ProjectsRefreshInterval <= 0 { + return + } + go r.refreshProjectIDsLoop(ctx) +} + +func (r *Runtime) refreshProjectIDsLoop(ctx context.Context) { + ticker := time.NewTicker(r.cfg.ProjectsRefreshInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.refreshProjectIDs(ctx) + } + } +} + +// refreshProjectIDs re-resolves cfg.ProjectsFilter and, on a successful +// non-empty result, atomically replaces the project list used by future +// scrapes. A transient API error or an unexpectedly empty result (e.g. IAM +// eventual consistency, a momentarily over-narrow filter) is logged and the +// previous known-good list is kept, so a background refresh hiccup never +// blanks out or fails a scrape. +func (r *Runtime) refreshProjectIDs(ctx context.Context) { + ids, err := r.discoverProjectIDs(ctx, r.cfg.ProjectsFilter) + if err != nil { + r.logger.Warn("failed to refresh project list from google.projects.filter; keeping previous list", "err", err) + return + } + + ids = append(ids, r.cfg.ProjectIDs...) + ids = deduplicateProjectIDs(ids) + + if len(ids) == 0 { + r.logger.Warn("google.projects.filter refresh returned zero projects; keeping previous list") + return + } + + r.projectIDs.Store(&ids) + r.logger.Info("refreshed project list from google.projects.filter", "count", len(ids)) +} + // WithCache returns a Runtime configured to cache its collectors per // (project, prefix-filter). Subsequent calls to Collectors or // CollectorsForPrefixes reuse cached entries until they expire, which lets @@ -125,8 +192,9 @@ func (r *Runtime) CollectorsForPrefixes(prefixFilter []string) ([]*MonitoringCol } func (r *Runtime) buildCollectors(prefixFilter []string) ([]*MonitoringCollector, error) { - result := make([]*MonitoringCollector, 0, len(r.projectIDs)) - for _, projectID := range r.projectIDs { + projectIDs := *r.projectIDs.Load() + result := make([]*MonitoringCollector, 0, len(projectIDs)) + for _, projectID := range projectIDs { c, err := r.collectorFor(projectID, prefixFilter) if err != nil { return nil, fmt.Errorf("collector for %q: %w", projectID, err) @@ -222,9 +290,16 @@ func getProjectIDsFromFilter(ctx context.Context, filter string) ([]string, erro if err != nil { return nil, err } + return listProjectIDs(ctx, service, filter) +} +// listProjectIDs returns the list of project IDs that match filter, using an +// already-constructed cloudresourcemanager service. Split out from +// getProjectIDsFromFilter so tests can inject a fake service pointed at an +// httptest.Server instead of relying on Application Default Credentials. +func listProjectIDs(ctx context.Context, service *cloudresourcemanager.Service, filter string) ([]string, error) { var projectIDs []string - err = service.Projects.List().Filter(filter).Pages(ctx, func(page *cloudresourcemanager.ListProjectsResponse) error { + err := service.Projects.List().Filter(filter).Pages(ctx, func(page *cloudresourcemanager.ListProjectsResponse) error { for _, project := range page.Projects { projectIDs = append(projectIDs, project.ProjectId) } diff --git a/collectors/runtime_test.go b/collectors/runtime_test.go index cdf2327b..5ae29ed9 100644 --- a/collectors/runtime_test.go +++ b/collectors/runtime_test.go @@ -15,12 +15,21 @@ package collectors import ( "context" + "encoding/json" + "errors" + "fmt" "log/slog" + "net/http" + "net/http/httptest" "reflect" "strings" + "sync/atomic" "testing" "time" + "google.golang.org/api/cloudresourcemanager/v1" + "google.golang.org/api/option" + "github.com/prometheus-community/stackdriver_exporter/config" ) @@ -189,3 +198,231 @@ func TestRuntimeFilterMetricTypePrefixes(t *testing.T) { }) } } + +// newTestRuntime builds a minimal Runtime for unit testing the project +// discovery refresh path, bypassing NewRuntime's ADC/GCP service setup. +func newTestRuntime(t *testing.T, cfg *config.Config, initialIDs []string, discover func(ctx context.Context, filter string) ([]string, error)) *Runtime { + t.Helper() + ptr := &atomic.Pointer[[]string]{} + ptr.Store(&initialIDs) + return &Runtime{ + cfg: cfg, + projectIDs: ptr, + logger: slog.Default(), + discoverProjectIDs: discover, + } +} + +// fakeCloudResourceManagerServer serves a canned ListProjectsResponse (or an +// error status) so listProjectIDs can be tested against a real HTTP client +// without depending on Application Default Credentials. +func fakeCloudResourceManagerServer(t *testing.T, statusCode int, resp *cloudresourcemanager.ListProjectsResponse) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(statusCode) + if resp != nil { + _ = json.NewEncoder(w).Encode(resp) + } + })) + t.Cleanup(server.Close) + return server +} + +func fakeCloudResourceManagerService(t *testing.T, server *httptest.Server) *cloudresourcemanager.Service { + t.Helper() + service, err := cloudresourcemanager.NewService(context.Background(), + option.WithEndpoint(server.URL), + option.WithHTTPClient(server.Client()), + option.WithoutAuthentication(), + ) + if err != nil { + t.Fatalf("cloudresourcemanager.NewService() error = %v", err) + } + return service +} + +func TestListProjectIDsReturnsMatchingProjects(t *testing.T) { + t.Parallel() + + server := fakeCloudResourceManagerServer(t, http.StatusOK, &cloudresourcemanager.ListProjectsResponse{ + Projects: []*cloudresourcemanager.Project{ + {ProjectId: "project-a"}, + {ProjectId: "project-b"}, + }, + }) + service := fakeCloudResourceManagerService(t, server) + + got, err := listProjectIDs(context.Background(), service, "parent.id:12345") + if err != nil { + t.Fatalf("listProjectIDs() error = %v", err) + } + want := []string{"project-a", "project-b"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("listProjectIDs() = %#v, want %#v", got, want) + } +} + +func TestListProjectIDsPropagatesAPIError(t *testing.T) { + t.Parallel() + + server := fakeCloudResourceManagerServer(t, http.StatusInternalServerError, nil) + service := fakeCloudResourceManagerService(t, server) + + _, err := listProjectIDs(context.Background(), service, "parent.id:12345") + if err == nil { + t.Fatal("listProjectIDs() expected error for 500 response, got nil") + } +} + +func TestRefreshProjectIDsUpdatesListOnSuccess(t *testing.T) { + t.Parallel() + + cfg := &config.Config{ProjectsFilter: "parent.id:12345"} + r := newTestRuntime(t, cfg, []string{"old-project"}, func(_ context.Context, _ string) ([]string, error) { + return []string{"new-project-b", "new-project-a"}, nil + }) + + r.refreshProjectIDs(context.Background()) + + got := *r.projectIDs.Load() + want := []string{"new-project-a", "new-project-b"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("projectIDs after refresh = %#v, want %#v", got, want) + } +} + +func TestRefreshProjectIDsMergesStaticProjectIDs(t *testing.T) { + t.Parallel() + + cfg := &config.Config{ProjectsFilter: "parent.id:12345", ProjectIDs: []string{"static-project", "new-project-a"}} + r := newTestRuntime(t, cfg, []string{"old-project"}, func(_ context.Context, _ string) ([]string, error) { + return []string{"new-project-a"}, nil + }) + + r.refreshProjectIDs(context.Background()) + + got := *r.projectIDs.Load() + want := []string{"new-project-a", "static-project"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("projectIDs after refresh = %#v, want %#v", got, want) + } +} + +func TestRefreshProjectIDsKeepsPreviousListOnAPIError(t *testing.T) { + t.Parallel() + + cfg := &config.Config{ProjectsFilter: "parent.id:12345"} + r := newTestRuntime(t, cfg, []string{"old-project"}, func(_ context.Context, _ string) ([]string, error) { + return nil, errors.New("transient GCP error") + }) + + r.refreshProjectIDs(context.Background()) + + got := *r.projectIDs.Load() + want := []string{"old-project"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("projectIDs after failed refresh = %#v, want unchanged %#v", got, want) + } +} + +func TestRefreshProjectIDsKeepsPreviousListWhenFilterReturnsZero(t *testing.T) { + t.Parallel() + + cfg := &config.Config{ProjectsFilter: "parent.id:12345"} + r := newTestRuntime(t, cfg, []string{"old-project"}, func(_ context.Context, _ string) ([]string, error) { + return nil, nil + }) + + r.refreshProjectIDs(context.Background()) + + got := *r.projectIDs.Load() + want := []string{"old-project"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("projectIDs after empty-result refresh = %#v, want unchanged %#v", got, want) + } +} + +func TestStartProjectDiscoveryRefreshRunsOnlyWhenFilterAndIntervalSet(t *testing.T) { + tests := []struct { + name string + filter string + interval time.Duration + expectTriggered bool + }{ + {name: "neither filter nor interval set", filter: "", interval: 0, expectTriggered: false}, + {name: "filter set, interval zero", filter: "parent.id:1", interval: 0, expectTriggered: false}, + {name: "interval set, filter empty", filter: "", interval: 10 * time.Millisecond, expectTriggered: false}, + {name: "filter and interval both set", filter: "parent.id:1", interval: 10 * time.Millisecond, expectTriggered: true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var calls int32 + cfg := &config.Config{ProjectsFilter: tt.filter, ProjectsRefreshInterval: tt.interval} + r := newTestRuntime(t, cfg, []string{"p"}, func(_ context.Context, _ string) ([]string, error) { + atomic.AddInt32(&calls, 1) + return []string{"p2"}, nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + r.StartProjectDiscoveryRefresh(ctx) + + time.Sleep(50 * time.Millisecond) + cancel() + time.Sleep(20 * time.Millisecond) + + triggered := atomic.LoadInt32(&calls) > 0 + if triggered != tt.expectTriggered { + t.Fatalf("refresh triggered = %v, want %v (calls=%d)", triggered, tt.expectTriggered, calls) + } + }) + } +} + +func TestBuildCollectorsSafeUnderConcurrentRefresh(t *testing.T) { + cfg := &config.Config{MetricsPrefixes: []string{"compute.googleapis.com/"}} + r := newTestRuntime(t, cfg, []string{"project-a"}, nil) + r.counterStoreFactory = func(_ *slog.Logger, _ time.Duration) DeltaCounterStore { return nil } + r.histogramStoreFactory = func(_ *slog.Logger, _ time.Duration) DeltaHistogramStore { return nil } + + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 200; i++ { + ids := []string{fmt.Sprintf("project-%d", i)} + r.projectIDs.Store(&ids) + } + }() + + for i := 0; i < 200; i++ { + if _, err := r.buildCollectors(nil); err != nil { + t.Fatalf("buildCollectors() error = %v", err) + } + } + <-done +} + +// TestWithCacheSharesProjectIDsPointer guards against a regression where +// WithCache's shallow struct copy (sibling := *r) would give the cached +// sibling a disconnected snapshot of the project list instead of sharing +// live state with the original Runtime that a background refresh updates. +func TestWithCacheSharesProjectIDsPointer(t *testing.T) { + cfg := &config.Config{MetricsPrefixes: []string{"compute.googleapis.com/"}} + r := newTestRuntime(t, cfg, []string{"project-a"}, nil) + r.counterStoreFactory = func(_ *slog.Logger, _ time.Duration) DeltaCounterStore { return nil } + r.histogramStoreFactory = func(_ *slog.Logger, _ time.Duration) DeltaHistogramStore { return nil } + + sibling := r.WithCache() + + updated := []string{"project-a", "project-b"} + r.projectIDs.Store(&updated) + + cs, err := sibling.Collectors() + if err != nil { + t.Fatalf("sibling.Collectors() error = %v", err) + } + if len(cs) != 2 { + t.Fatalf("sibling.Collectors() returned %d collectors after original's projectIDs updated, want 2", len(cs)) + } +} diff --git a/config/config.go b/config/config.go index e6301474..95217484 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 + DefaultProjectsRefreshInterval = 0 * time.Second ) // DefaultRetryStatuses must be treated as immutable after declaration. @@ -45,6 +46,7 @@ var DefaultRetryStatuses = []int{http.StatusServiceUnavailable} type Config struct { ProjectIDs []string ProjectsFilter string + ProjectsRefreshInterval time.Duration UniverseDomain string MaxRetries int HTTPTimeout time.Duration @@ -87,6 +89,7 @@ func NewConfigWithDefaults() *Config { AggregateDeltasTTL: DefaultDeltasTTL, DescriptorCacheTTL: DefaultDescriptorTTL, DescriptorCacheOnlyGoogle: DefaultDescriptorGoogleOnly, + ProjectsRefreshInterval: DefaultProjectsRefreshInterval, } } diff --git a/config/config_test.go b/config/config_test.go index 07b44e34..71c7ebd9 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -69,6 +69,9 @@ func TestNewConfigWithDefaults(t *testing.T) { if c.HTTPTimeout != DefaultHTTPTimeout { t.Errorf("HTTPTimeout = %v, want %v", c.HTTPTimeout, DefaultHTTPTimeout) } + if c.ProjectsRefreshInterval != DefaultProjectsRefreshInterval { + t.Errorf("ProjectsRefreshInterval = %v, want %v", c.ProjectsRefreshInterval, DefaultProjectsRefreshInterval) + } if len(c.RetryStatuses) != len(DefaultRetryStatuses) || c.RetryStatuses[0] != DefaultRetryStatuses[0] { t.Errorf("RetryStatuses = %v, want %v", c.RetryStatuses, DefaultRetryStatuses) } diff --git a/stackdriver_exporter.go b/stackdriver_exporter.go index fa7cda26..b5c688a1 100644 --- a/stackdriver_exporter.go +++ b/stackdriver_exporter.go @@ -63,6 +63,10 @@ var ( "google.projects.filter", "Google projects search filter.", ).String() + projectsRefreshInterval = kingpin.Flag( + "google.projects.filter-refresh-interval", "How often to re-evaluate google.projects.filter to pick up new/removed projects. 0 disables periodic refresh.", + ).Default(config.DefaultProjectsRefreshInterval.String()).Duration() + googleUniverseDomain = kingpin.Flag( "google.universe-domain", "The Cloud universe to use.", ).Default(config.DefaultUniverseDomain).String() @@ -143,31 +147,37 @@ func init() { } type handler struct { - handler http.Handler logger *slog.Logger runtime *collectors.Runtime additionalGatherer prometheus.Gatherer } +// ServeHTTP rebuilds its collector registry from the runtime's current +// project list on every request (as the ?collect= filtered path always has), +// rather than serving a registry frozen at startup. This is what lets a +// background google.projects.filter refresh (see Runtime.StartProjectDiscoveryRefresh) +// actually reach scrapes: Runtime.WithCache() keeps the underlying +// MonitoringCollectors (and their delta-counter state) cached across calls, +// so rebuilding the registry wrapper per request is cheap. func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { collectParams := r.URL.Query()["collect"] filters := make(map[string]bool) for _, param := range collectParams { filters[param] = true } + prefixFilter := make([]string, 0, len(filters)) + for f := range filters { + prefixFilter = append(prefixFilter, f) + } + slices.Sort(prefixFilter) - if len(filters) > 0 { - handler, err := h.filteredHandler(filters) - if err != nil { - h.logger.Error("error creating monitoring collector", "err", err) - http.Error(w, "internal error", http.StatusInternalServerError) - return - } - handler.ServeHTTP(w, r) + httpHandler, err := h.handlerForPrefixes(prefixFilter) + if err != nil { + h.logger.Error("error creating monitoring collector", "err", err) + http.Error(w, "internal error", http.StatusInternalServerError) return } - - h.handler.ServeHTTP(w, r) + httpHandler.ServeHTTP(w, r) } func newHandler(runtime *collectors.Runtime, logger *slog.Logger, additionalGatherer prometheus.Gatherer) (*handler, error) { @@ -177,34 +187,24 @@ func newHandler(runtime *collectors.Runtime, logger *slog.Logger, additionalGath additionalGatherer: additionalGatherer, } - cs, err := runtime.Collectors() - if err != nil { + // Fail fast on startup config errors (e.g. bad metric prefixes); the + // resulting collectors are discarded since each scrape builds its own. + if _, err := runtime.Collectors(); err != nil { return nil, fmt.Errorf("build collectors: %w", err) } - registry := prometheus.NewRegistry() - for _, c := range cs { - if err := registry.Register(c); err != nil { - return nil, fmt.Errorf("register collector: %w", err) - } - } - h.handler = h.handlerFor(registry) return h, nil } -func (h *handler) filteredHandler(filters map[string]bool) (http.Handler, error) { - prefixFilter := make([]string, 0, len(filters)) - for f := range filters { - prefixFilter = append(prefixFilter, f) - } - slices.Sort(prefixFilter) - +func (h *handler) handlerForPrefixes(prefixFilter []string) (http.Handler, error) { cs, err := h.runtime.CollectorsForPrefixes(prefixFilter) if err != nil { return nil, err } registry := prometheus.NewRegistry() for _, c := range cs { - registry.MustRegister(c) + if err := registry.Register(c); err != nil { + return nil, fmt.Errorf("register collector: %w", err) + } } return h.handlerFor(registry), nil } @@ -236,7 +236,8 @@ func main() { if *monitoringMetricsTypePrefixes != "" { logger.Warn("The monitoring.metrics-type-prefixes flag is deprecated and will be replaced by monitoring.metrics-prefix.") } - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() cfg := collectorConfigFromFlags() @@ -268,6 +269,7 @@ func main() { os.Exit(1) } runtime = runtime.WithCache() + runtime.StartProjectDiscoveryRefresh(ctx) if *metricsPath == *stackdriverMetricsPath { h, err := newHandler(runtime, logger, prometheus.DefaultGatherer) @@ -326,6 +328,7 @@ func collectorConfigFromFlags() *config.Config { return &config.Config{ ProjectIDs: slices.Clone(*projectIDs), ProjectsFilter: *projectsFilter, + ProjectsRefreshInterval: *projectsRefreshInterval, UniverseDomain: *googleUniverseDomain, MaxRetries: *stackdriverMaxRetries, HTTPTimeout: *stackdriverHttpTimeout, diff --git a/stackdriver_exporter_test.go b/stackdriver_exporter_test.go new file mode 100644 index 00000000..3eacdeb2 --- /dev/null +++ b/stackdriver_exporter_test.go @@ -0,0 +1,48 @@ +// Copyright 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 main + +import ( + "testing" + "time" + + "github.com/alecthomas/kingpin/v2" + + "github.com/prometheus-community/stackdriver_exporter/config" +) + +func TestCollectorConfigFromFlagsIncludesProjectsRefreshInterval(t *testing.T) { + if _, err := kingpin.CommandLine.Parse([]string{ + "--google.projects.filter-refresh-interval=2m", + }); err != nil { + t.Fatalf("kingpin.CommandLine.Parse() error = %v", err) + } + + cfg := collectorConfigFromFlags() + want := 2 * time.Minute + if cfg.ProjectsRefreshInterval != want { + t.Fatalf("ProjectsRefreshInterval = %v, want %v", cfg.ProjectsRefreshInterval, want) + } +} + +func TestCollectorConfigFromFlagsDefaultsProjectsRefreshIntervalToDisabled(t *testing.T) { + if _, err := kingpin.CommandLine.Parse([]string{}); err != nil { + t.Fatalf("kingpin.CommandLine.Parse() error = %v", err) + } + + cfg := collectorConfigFromFlags() + if cfg.ProjectsRefreshInterval != config.DefaultProjectsRefreshInterval { + t.Fatalf("ProjectsRefreshInterval = %v, want default %v", cfg.ProjectsRefreshInterval, config.DefaultProjectsRefreshInterval) + } +} From 2d40e3532e95bcfeca7d13ec8a46f389298abf54 Mon Sep 17 00:00:00 2001 From: Pavan Nalam Date: Fri, 18 Sep 2026 09:35:33 -0700 Subject: [PATCH 2/5] remove comments Signed-off-by: Pavan Nalam --- collectors/runtime.go | 30 ++---------------------------- collectors/runtime_test.go | 5 ----- stackdriver_exporter.go | 2 -- 3 files changed, 2 insertions(+), 35 deletions(-) diff --git a/collectors/runtime.go b/collectors/runtime.go index 5edb14cf..0bfac249 100644 --- a/collectors/runtime.go +++ b/collectors/runtime.go @@ -38,22 +38,14 @@ type HistogramStoreFactory func(logger *slog.Logger, ttl time.Duration) DeltaHis // Runtime holds the resolved state produced by NewRuntime. type Runtime struct { - cfg *config.Config - // projectIDs is a *atomic.Pointer[[]string] (rather than a plain []string) - // so that WithCache's shallow struct copy shares live state with the - // original: a background refresh (see StartProjectDiscoveryRefresh) must - // be visible through every Runtime value derived from the same NewRuntime - // call, and scrapes must be able to read it without lock contention. + cfg *config.Config projectIDs *atomic.Pointer[[]string] service *monitoring.Service logger *slog.Logger counterStoreFactory CounterStoreFactory histogramStoreFactory HistogramStoreFactory cache *collectorCache - // discoverProjectIDs resolves cfg.ProjectsFilter to project IDs. It - // defaults to getProjectIDsFromFilter; tests override it to avoid - // depending on Application Default Credentials. - discoverProjectIDs func(ctx context.Context, filter string) ([]string, error) + discoverProjectIDs func(ctx context.Context, filter string) ([]string, error) } // NewRuntime resolves project IDs and creates the monitoring service. The @@ -108,14 +100,6 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co }, nil } -// StartProjectDiscoveryRefresh periodically re-resolves cfg.ProjectsFilter in -// the background so that projects added to or removed from the matching -// org/folder are picked up without restarting the exporter. It is a no-op -// unless both cfg.ProjectsFilter and cfg.ProjectsRefreshInterval are set, so -// existing deployments see no behavior change unless they opt in. -// -// The refresh runs until ctx is canceled; callers must cancel ctx on shutdown -// to avoid leaking the goroutine. func (r *Runtime) StartProjectDiscoveryRefresh(ctx context.Context) { if r.cfg.ProjectsFilter == "" || r.cfg.ProjectsRefreshInterval <= 0 { return @@ -136,12 +120,6 @@ func (r *Runtime) refreshProjectIDsLoop(ctx context.Context) { } } -// refreshProjectIDs re-resolves cfg.ProjectsFilter and, on a successful -// non-empty result, atomically replaces the project list used by future -// scrapes. A transient API error or an unexpectedly empty result (e.g. IAM -// eventual consistency, a momentarily over-narrow filter) is logged and the -// previous known-good list is kept, so a background refresh hiccup never -// blanks out or fails a scrape. func (r *Runtime) refreshProjectIDs(ctx context.Context) { ids, err := r.discoverProjectIDs(ctx, r.cfg.ProjectsFilter) if err != nil { @@ -293,10 +271,6 @@ func getProjectIDsFromFilter(ctx context.Context, filter string) ([]string, erro return listProjectIDs(ctx, service, filter) } -// listProjectIDs returns the list of project IDs that match filter, using an -// already-constructed cloudresourcemanager service. Split out from -// getProjectIDsFromFilter so tests can inject a fake service pointed at an -// httptest.Server instead of relying on Application Default Credentials. func listProjectIDs(ctx context.Context, service *cloudresourcemanager.Service, filter string) ([]string, error) { var projectIDs []string err := service.Projects.List().Filter(filter).Pages(ctx, func(page *cloudresourcemanager.ListProjectsResponse) error { diff --git a/collectors/runtime_test.go b/collectors/runtime_test.go index 5ae29ed9..bbc31c51 100644 --- a/collectors/runtime_test.go +++ b/collectors/runtime_test.go @@ -199,8 +199,6 @@ func TestRuntimeFilterMetricTypePrefixes(t *testing.T) { } } -// newTestRuntime builds a minimal Runtime for unit testing the project -// discovery refresh path, bypassing NewRuntime's ADC/GCP service setup. func newTestRuntime(t *testing.T, cfg *config.Config, initialIDs []string, discover func(ctx context.Context, filter string) ([]string, error)) *Runtime { t.Helper() ptr := &atomic.Pointer[[]string]{} @@ -213,9 +211,6 @@ func newTestRuntime(t *testing.T, cfg *config.Config, initialIDs []string, disco } } -// fakeCloudResourceManagerServer serves a canned ListProjectsResponse (or an -// error status) so listProjectIDs can be tested against a real HTTP client -// without depending on Application Default Credentials. func fakeCloudResourceManagerServer(t *testing.T, statusCode int, resp *cloudresourcemanager.ListProjectsResponse) *httptest.Server { t.Helper() server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/stackdriver_exporter.go b/stackdriver_exporter.go index b5c688a1..85a4091d 100644 --- a/stackdriver_exporter.go +++ b/stackdriver_exporter.go @@ -187,8 +187,6 @@ func newHandler(runtime *collectors.Runtime, logger *slog.Logger, additionalGath additionalGatherer: additionalGatherer, } - // Fail fast on startup config errors (e.g. bad metric prefixes); the - // resulting collectors are discarded since each scrape builds its own. if _, err := runtime.Collectors(); err != nil { return nil, fmt.Errorf("build collectors: %w", err) } From 4fa4a409b8de65ce86f858525d35ea3db2384a30 Mon Sep 17 00:00:00 2001 From: Pavan Nalam Date: Fri, 18 Sep 2026 09:53:20 -0700 Subject: [PATCH 3/5] fix(collectors): fix data race in collectorCache.Get holding only RLock while mutating Get took an RLock (shared among concurrent callers) but still mutated shared state on every call: deleting the map entry on expiry, or writing entry.expiry on a hit. Two goroutines calling Get on the same key concurrently could race on entry.expiry, or both call delete on the same map concurrently, which Go maps do not support and can panic the process with "fatal error: concurrent map writes". This was pre-existing but rarely exercised, since only the ?collect= filtered scrape path went through the cache per request. The prior commit in this branch made the default /metrics path do the same on every scrape, so any overlapping scrapes (HA Prometheus replicas, manual curl during a scheduled scrape, etc.) now hit this far more easily. Every path in Get mutates state, so there is no actual read-only case for RLock to protect; switched to a plain Mutex, matching what Store already used. Added a concurrent-access regression test that reproduces the data race under -race against the old RLock code and passes cleanly with the fix. Signed-off-by: Pavan Nalam --- collectors/cache.go | 6 +++--- collectors/cache_test.go | 22 ++++++++++++++++++++++ 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/collectors/cache.go b/collectors/cache.go index c60fde9e..b26bcaf3 100644 --- a/collectors/cache.go +++ b/collectors/cache.go @@ -78,7 +78,7 @@ func (d *descriptorCache) Store(prefix string, data []*monitoring.MetricDescript // rebuild per request can preserve delta-counter state across calls. type collectorCache struct { cache map[string]*collectorCacheEntry - lock sync.RWMutex + lock sync.Mutex ttl time.Duration } @@ -99,8 +99,8 @@ func newCollectorCache(ttl time.Duration) *collectorCache { // Get returns the cached collector for key, refreshing its TTL on hit. func (c *collectorCache) Get(key string) (*MonitoringCollector, bool) { - c.lock.RLock() - defer c.lock.RUnlock() + c.lock.Lock() + defer c.lock.Unlock() entry, ok := c.cache[key] diff --git a/collectors/cache_test.go b/collectors/cache_test.go index d8960b5b..ffaad38d 100644 --- a/collectors/cache_test.go +++ b/collectors/cache_test.go @@ -15,6 +15,7 @@ package collectors import ( "fmt" + "sync" "testing" "time" @@ -127,3 +128,24 @@ func TestCollectorCache(t *testing.T) { } }) } + +func TestCollectorCacheConcurrentGetIsRaceFree(t *testing.T) { + cache := newCollectorCache(time.Millisecond) + key := "test-key" + cache.Store(key, &MonitoringCollector{projectID: "test-project"}) + + var wg sync.WaitGroup + for i := 0; i < 50; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for j := 0; j < 200; j++ { + cache.Get(key) + if j%10 == 0 { + cache.Store(key, &MonitoringCollector{projectID: "test-project"}) + } + } + }() + } + wg.Wait() +} From 2a16c2986bcb5fdf6b2cb1c41b215e3313566f4d Mon Sep 17 00:00:00 2001 From: Pavan Nalam Date: Mon, 21 Sep 2026 07:15:39 -0700 Subject: [PATCH 4/5] fix(collectors): hold a single Resource Manager service with timeout and retries getProjectIDsFromFilter constructed a bare cloudresourcemanager.Service with cloudresourcemanager.NewService(ctx) on every call: no timeout, no retry handling. It ran once at startup and again on every google.projects.filter-refresh-interval tick, so a stall on Google's side during a refresh would block with no indication why, and the service was needlessly rebuilt from scratch each refresh. Adds createResourceManagerService in service.go, mirroring createMonitoringService: same cfg.HTTPTimeout, MaxRetries, RetryStatuses, BackoffJitter, and MaxBackoff wiring via rehttp, using CloudPlatformReadOnlyScope (least-privilege) instead of the broader default scope the bare constructor fell back to, and skipping UniverseDomain. NewRuntime now constructs it once, only when ProjectsFilter is set, and closes over it for both the initial resolution and every later refresh via the existing discoverProjectIDs seam. Signed-off-by: Pavan Nalam --- collectors/runtime.go | 23 +++++++++++------------ collectors/service.go | 24 ++++++++++++++++++++++++ 2 files changed, 35 insertions(+), 12 deletions(-) diff --git a/collectors/runtime.go b/collectors/runtime.go index 0bfac249..e7c6f92e 100644 --- a/collectors/runtime.go +++ b/collectors/runtime.go @@ -60,9 +60,18 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co } var projectIDs []string + var discoverProjectIDs func(ctx context.Context, filter string) ([]string, error) if cfg.ProjectsFilter != "" { - ids, err := getProjectIDsFromFilter(ctx, cfg.ProjectsFilter) + resourceManagerService, err := createResourceManagerService(ctx, cfg) + if err != nil { + return nil, fmt.Errorf("failed to create Google Cloud Resource Manager service: %w", err) + } + discoverProjectIDs = func(ctx context.Context, filter string) ([]string, error) { + return listProjectIDs(ctx, resourceManagerService, filter) + } + + ids, err := discoverProjectIDs(ctx, cfg.ProjectsFilter) if err != nil { return nil, fmt.Errorf("failed to resolve project IDs from projects_filter: %w", err) } @@ -96,7 +105,7 @@ func NewRuntime(ctx context.Context, logger *slog.Logger, cfg *config.Config, co logger: logger, counterStoreFactory: counterFactory, histogramStoreFactory: histogramFactory, - discoverProjectIDs: getProjectIDsFromFilter, + discoverProjectIDs: discoverProjectIDs, }, nil } @@ -261,16 +270,6 @@ func discoverDefaultProjectID(ctx context.Context) (string, error) { return credentials.ProjectID, nil } -// getProjectIDsFromFilter returns the list of project IDs that match a Google -// Cloud organization-scoped projects filter. -func getProjectIDsFromFilter(ctx context.Context, filter string) ([]string, error) { - service, err := cloudresourcemanager.NewService(ctx) - if err != nil { - return nil, err - } - return listProjectIDs(ctx, service, filter) -} - func listProjectIDs(ctx context.Context, service *cloudresourcemanager.Service, filter string) ([]string, error) { var projectIDs []string err := service.Projects.List().Filter(filter).Pages(ctx, func(page *cloudresourcemanager.ListProjectsResponse) error { diff --git a/collectors/service.go b/collectors/service.go index 4ec9450a..622670e7 100644 --- a/collectors/service.go +++ b/collectors/service.go @@ -19,6 +19,7 @@ import ( "github.com/PuerkitoBio/rehttp" "golang.org/x/oauth2/google" + "google.golang.org/api/cloudresourcemanager/v1" "google.golang.org/api/monitoring/v3" "google.golang.org/api/option" @@ -62,3 +63,26 @@ func createMonitoringService(ctx context.Context, cfg *config.Config) (*monitori } return service, nil } + +func createResourceManagerService(ctx context.Context, cfg *config.Config) (*cloudresourcemanager.Service, error) { + googleClient, err := google.DefaultClient(ctx, cloudresourcemanager.CloudPlatformReadOnlyScope) + if err != nil { + return nil, fmt.Errorf("error creating Google client: %w", err) + } + + googleClient.Timeout = cfg.HTTPTimeout + googleClient.Transport = rehttp.NewTransport( + googleClient.Transport, + rehttp.RetryAll( + rehttp.RetryMaxRetries(cfg.MaxRetries), + rehttp.RetryStatuses(cfg.RetryStatuses...), + ), + rehttp.ExpJitterDelay(cfg.BackoffJitter, cfg.MaxBackoff), + ) + + service, err := cloudresourcemanager.NewService(ctx, option.WithHTTPClient(googleClient)) + if err != nil { + return nil, fmt.Errorf("error creating Google Cloud Resource Manager service: %w", err) + } + return service, nil +} From 44eedd8c4062ffab7edc804b23e7b9490c983404 Mon Sep 17 00:00:00 2001 From: Pavan Nalam Date: Mon, 21 Sep 2026 07:32:26 -0700 Subject: [PATCH 5/5] docs: note that google.projects.filter resolution reuses stackdriver.* timeout/retry settings The stackdriver.http-timeout, stackdriver.max-retries, and stackdriver.retry-statuses descriptions read as Monitoring-API-only, but createResourceManagerService now applies the same settings to resolving and refreshing google.projects.filter. Signed-off-by: Pavan Nalam --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index b5346c48..17273837 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ If you are still using the legacy [Access scopes][access-scopes], the `https://w | Flag | Required | Default | Description | | ----------------------------------- | -------- |---------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `google.project-ids` | No | GCloud SDK auto-discovery | Repeatable flag of Google Project IDs | -| `google.projects.filter` | No | | GCloud projects filter expression. See more [here](https://cloud.google.com/sdk/gcloud/reference/projects/list). | +| `google.projects.filter` | No | | GCloud projects filter expression. See more [here](https://cloud.google.com/sdk/gcloud/reference/projects/list). Resolving and refreshing this filter uses the same `stackdriver.http-timeout`, `stackdriver.max-retries`, and `stackdriver.retry-statuses` settings as the Monitoring API calls. | | `google.projects.filter-refresh-interval` | No | `0s` (disabled) | How often to re-evaluate `google.projects.filter` to pick up projects added to or removed from the matching org/folder without restarting. `0` keeps the pre-existing behavior of resolving the filter once at startup only. | | `google.universe-domain` | No | `googleapis.com` | Target specific Google Cloud environments, such as public cloud, or specific sovereign clouds | | `monitoring.metrics-ingest-delay` | No | | Offsets metric collection by a delay appropriate for each metric type, e.g. because bigquery metrics are slow to appear |