feat(collectors): periodically refresh google.projects.filter to pick up new/removed projects without a restart

README.md

@@ -80,7 +80,8 @@ 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                                                                           |
 | `monitoring.drop-delegated-projects` | No       | No                        | Drop metrics from attached projects and fetch `project_id` only.                                                                                                                                  |

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]
 

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

collectors/runtime.go

@@ -19,6 +19,7 @@ import (
 	"log/slog"
 	"slices"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"golang.org/x/oauth2/google"
@@ -38,12 +39,13 @@ 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
+	projectIDs            *atomic.Pointer[[]string]
 	service               *monitoring.Service
 	logger                *slog.Logger
 	counterStoreFactory   CounterStoreFactory
 	histogramStoreFactory HistogramStoreFactory
 	cache                 *collectorCache
+	discoverProjectIDs    func(ctx context.Context, filter string) ([]string, error)
 }
 
 // NewRuntime resolves project IDs and creates the monitoring service. The
@@ -58,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)
 		}
@@ -84,16 +95,59 @@ 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:    discoverProjectIDs,
 	}, nil
 }
 
+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)
+		}
+	}
+}
+
+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 +179,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)
@@ -215,16 +270,9 @@ 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
-	}
-
+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)
 		}