refactor: create a new prometheus registry for every server

Author: miparnisari

CHANGELOG.md

@@ -22,7 +22,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Tracing: When server is shutting down, flush traces. Also, elide the need for setting `OTEL_EXPORTER_OTLP_ENDPOINT`. (https://github.com/authzed/spicedb/pull/3108)
 - Fixed a LookupSubjects issue in the query planner around the handling of wildcards in compound permissions (https://github.com/authzed/spicedb/pull/3140)
 - MySQL: identifiers (object/subject IDs and relationship counter names) are now stored with a case-sensitive (binary) collation, matching the Postgres, CockroachDB, and Spanner datastores. Previously, identifiers differing only in letter case (e.g. `Foo` and `foo`) incorrectly collided in unique indexes and lookups. ⚠️ The migration rebuilds the `relation_tuple` table in place via `ALTER TABLE`, which can hold a metadata/table lock for a long time on large datasets — run the upgrade in a low-traffic window, or apply it with an online schema-change tool (e.g. gh-ost). (https://github.com/authzed/spicedb/pull/3161)
-- `server.NewConfigWithOptionsAndDefaults` now populates `Config` and its embedded structs with the same defaults as the CLI flags, fixing zero-value behavior when embedding SpiceDB as a library. (https://github.com/authzed/spicedb/pull/3156)
+- `server.NewConfigWithOptionsAndDefaults` now populates `Config` and its embedded structs with the same defaults as the CLI flags, fixing zero-value behavior when embedding SpiceDB as a library. (https://github.com/authzed/spicedb/pull/3156, https://github.com/authzed/spicedb/pull/3170)
 
 ## [1.53.0] - 2026-05-13
 ### Added

internal/datastore/proxy/schemacaching/watchingcache_test.go

@@ -40,6 +40,9 @@ func TestWatchingCachingProxyUnwrap(t *testing.T) {
 	wcache := NewWatchingCacheProxy(fakeDS, NewDefinitionCachingProxy(fakeDS, nil), 1*time.Hour, 100*time.Millisecond)
 	unwrapped := wcache.Unwrap()
 	require.Equal(t, fakeDS, unwrapped)
+	t.Cleanup(func() {
+		wcache.Close()
+	})
 }
 
 func TestOldWatchingCacheBasicOperation(t *testing.T) {

internal/dispatch/caching/caching.go

@@ -2,6 +2,7 @@ package caching
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"maps"
 	"sync"
@@ -55,93 +56,73 @@ func DispatchTestCache(t testing.TB) cache.Cache[keys.DispatchCacheKey, any] {
 
 // NewCachingDispatcher creates a new dispatch.Dispatcher which delegates
 // dispatch requests and caches the responses when possible and desirable.
-func NewCachingDispatcher(cacheInst cache.Cache[keys.DispatchCacheKey, any], metricsEnabled bool, prometheusSubsystem string, keyHandler keys.Handler) (*Dispatcher, error) {
+func NewCachingDispatcher(cacheInst cache.Cache[keys.DispatchCacheKey, any], prometheusmetrics dispatch.MetricsOptions, keyHandler keys.Handler) (*Dispatcher, error) {
 	if cacheInst == nil {
 		cacheInst = cache.NoopCache[keys.DispatchCacheKey, any]()
 	}
 
 	checkTotalCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "check_total",
 		Help:      "Total number of CheckPermission dispatch requests processed.",
 	})
 	checkFromCacheCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "check_from_cache_total",
 		Help:      "Total number of CheckPermission dispatch requests served directly from the dispatch cache, avoiding re-computation.",
 	})
 
 	lookupResourcesTotalCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "lookup_resources_total",
 		Help:      "Total number of LookupResources dispatch requests processed.",
 	})
 	lookupResourcesFromCacheCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "lookup_resources_from_cache_total",
 		Help:      "Total number of LookupResources dispatch requests served directly from the dispatch cache.",
 	})
 
 	lookupSubjectsTotalCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "lookup_subjects_total",
 		Help:      "Total number of LookupSubjects dispatch requests processed.",
 	})
 	lookupSubjectsFromCacheCounter := prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "lookup_subjects_from_cache_total",
 		Help:      "Total number of LookupSubjects dispatch requests served directly from the dispatch cache.",
 	})
 
 	queryPlanTotalCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "query_plan_total",
 		Help:      "Total number of DispatchQueryPlan requests processed, labelled by plan operation.",
 	}, []string{"operation"})
 	queryPlanFromCacheCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: prometheusNamespace,
-		Subsystem: prometheusSubsystem,
+		Subsystem: prometheusmetrics.PrometheusSubsystem,
 		Name:      "query_plan_from_cache_total",
 		Help:      "Total number of DispatchQueryPlan requests served directly from the dispatch cache, labelled by plan operation.",
 	}, []string{"operation"})
 
-	if metricsEnabled && prometheusSubsystem != "" {
-		err := prometheus.Register(checkTotalCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(checkFromCacheCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(lookupResourcesTotalCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(lookupResourcesFromCacheCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(lookupSubjectsTotalCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(lookupSubjectsFromCacheCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(queryPlanTotalCounter)
-		if err != nil {
-			return nil, fmt.Errorf(errCachingInitialization, err)
-		}
-		err = prometheus.Register(queryPlanFromCacheCounter)
+	if prometheusmetrics.Enabled() {
+		err1 := prometheusmetrics.PrometheusRegistry.Register(checkTotalCounter)
+		err2 := prometheusmetrics.PrometheusRegistry.Register(checkFromCacheCounter)
+		err3 := prometheusmetrics.PrometheusRegistry.Register(lookupResourcesTotalCounter)
+		err4 := prometheusmetrics.PrometheusRegistry.Register(lookupResourcesFromCacheCounter)
+		err5 := prometheusmetrics.PrometheusRegistry.Register(lookupSubjectsTotalCounter)
+		err6 := prometheusmetrics.PrometheusRegistry.Register(lookupSubjectsFromCacheCounter)
+		err7 := prometheusmetrics.PrometheusRegistry.Register(queryPlanTotalCounter)
+		err8 := prometheusmetrics.PrometheusRegistry.Register(queryPlanFromCacheCounter)
+		err := errors.Join(err1, err2, err3, err4, err5, err6, err7, err8)
 		if err != nil {
 			return nil, fmt.Errorf(errCachingInitialization, err)
 		}

internal/dispatch/caching/cachingdispatch_test.go

@@ -125,7 +125,7 @@ func TestMaxDepthCaching(t *testing.T) {
 				}
 			}
 
-			dispatch, err := NewCachingDispatcher(DispatchTestCache(t), false, "", nil)
+			dispatch, err := NewCachingDispatcher(DispatchTestCache(t), dispatch.MetricsOptions{}, nil)
 			dispatch.SetDelegate(delegate)
 			require.NoError(err)
 			defer dispatch.Close()
@@ -177,7 +177,7 @@ func TestConcurrentDebugInfoAccess(t *testing.T) {
 			},
 		}, nil)
 
-	dispatcher, err := NewCachingDispatcher(DispatchTestCache(t), false, "", nil)
+	dispatcher, err := NewCachingDispatcher(DispatchTestCache(t), dispatch.MetricsOptions{}, nil)
 	require.NoError(err)
 	dispatcher.SetDelegate(delegate)
 	t.Cleanup(func() {
@@ -294,8 +294,9 @@ func TestDispatchQueryPlanRecordsCachingMetrics(t *testing.T) {
 	// Use a unique subsystem so we don't collide with other tests in the global
 	// default Prometheus registry.
 	subsystem := fmt.Sprintf("test_caching_%d", time.Now().UnixNano())
+	reg := prometheus.NewRegistry()
 
-	dispatcher, err := NewCachingDispatcher(DispatchTestCache(t), true, subsystem, nil)
+	dispatcher, err := NewCachingDispatcher(DispatchTestCache(t), dispatch.MetricsOptions{PrometheusSubsystem: subsystem, PrometheusRegistry: reg}, nil)
 	require.NoError(t, err)
 	t.Cleanup(func() { _ = dispatcher.Close() })
 
@@ -326,16 +327,14 @@ func TestDispatchQueryPlanRecordsCachingMetrics(t *testing.T) {
 	require.NoError(t, dispatcher.DispatchQueryPlan(req, stream2))
 	require.Len(t, stream2.Results(), 1)
 
-	gatherer := prometheus.DefaultGatherer
-
-	totalCheck := sumOperationCounter(t, gatherer, "spicedb_"+subsystem+"_query_plan_total", "check")
+	totalCheck := sumOperationCounter(t, reg, "spicedb_"+subsystem+"_query_plan_total", "check")
 	require.InEpsilon(t, float64(2), totalCheck, 1e-9, "query_plan_total{operation=check} should bump for each plan dispatch")
 
-	fromCacheCheck := sumOperationCounter(t, gatherer, "spicedb_"+subsystem+"_query_plan_from_cache_total", "check")
+	fromCacheCheck := sumOperationCounter(t, reg, "spicedb_"+subsystem+"_query_plan_from_cache_total", "check")
 	require.InEpsilon(t, float64(1), fromCacheCheck, 1e-9, "query_plan_from_cache_total{operation=check} should bump only on cache hit")
 
 	// Ensure the plan path no longer inflates the classic DispatchCheck counter.
-	checkTotal := sumPlainCounter(t, gatherer, "spicedb_"+subsystem+"_check_total")
+	checkTotal := sumPlainCounter(t, reg, "spicedb_"+subsystem+"_check_total")
 	require.Zero(t, checkTotal, "check_total must not be incremented from the plan path")
 
 	// The delegate is only consulted on the cache miss.

internal/dispatch/cluster/cluster.go

@@ -18,8 +18,8 @@ import (
 type Option func(*optionState)
 
 type optionState struct {
-	metricsEnabled               bool
-	prometheusSubsystem          string
+	metrics dispatch.MetricsOptions
+
 	cache                        cache.Cache[keys.DispatchCacheKey, any]
 	concurrencyLimits            graph.ConcurrencyLimits
 	remoteDispatchTimeout        time.Duration
@@ -38,17 +38,10 @@ func QueryPlanMetadata(m *query.QueryPlanMetadata) Option {
 	}
 }
 
-// MetricsEnabled enables issuing prometheus metrics
-func MetricsEnabled(enabled bool) Option {
-	return func(state *optionState) {
-		state.metricsEnabled = enabled
-	}
-}
-
-// PrometheusSubsystem sets the subsystem name for the prometheus metrics
-func PrometheusSubsystem(name string) Option {
+// Metrics sets the prometheus metrics
+func Metrics(reg dispatch.MetricsOptions) Option {
 	return func(state *optionState) {
-		state.prometheusSubsystem = name
+		state.metrics = reg
 	}
 }
 
@@ -155,11 +148,11 @@ func NewClusterDispatcher(dispatch dispatch.Dispatcher, options ...Option) (disp
 		return nil, fmt.Errorf("failed to create cluster dispatcher: %w", err)
 	}
 
-	if opts.prometheusSubsystem == "" {
-		opts.prometheusSubsystem = "dispatch"
+	if opts.metrics.PrometheusSubsystem == "" {
+		opts.metrics.PrometheusSubsystem = "dispatch"
 	}
 
-	cachingClusterDispatch, err := caching.NewCachingDispatcher(opts.cache, opts.metricsEnabled, opts.prometheusSubsystem, &keys.CanonicalKeyHandler{})
+	cachingClusterDispatch, err := caching.NewCachingDispatcher(opts.cache, opts.metrics, &keys.CanonicalKeyHandler{})
 	if err != nil {
 		return nil, err
 	}

internal/dispatch/combined/combined.go

@@ -26,8 +26,7 @@ import (
 type Option func(*optionState)
 
 type optionState struct {
-	metricsEnabled                               bool
-	prometheusSubsystem                          string
+	metrics                                      dispatch.MetricsOptions
 	upstreamAddr                                 string
 	upstreamCAPath                               string
 	grpcPresharedKey                             string
@@ -55,17 +54,10 @@ func QueryPlanMetadata(m *query.QueryPlanMetadata) Option {
 	}
 }
 
-// MetricsEnabled enables issuing prometheus metrics
-func MetricsEnabled(enabled bool) Option {
+// Metrics sets the prometheus metrics
+func Metrics(reg dispatch.MetricsOptions) Option {
 	return func(state *optionState) {
-		state.metricsEnabled = enabled
-	}
-}
-
-// PrometheusSubsystem sets the subsystem name for the prometheus metrics
-func PrometheusSubsystem(name string) Option {
-	return func(state *optionState) {
-		state.prometheusSubsystem = name
+		state.metrics = reg
 	}
 }
 
@@ -202,11 +194,11 @@ func NewDispatcher(options ...Option) (dispatch.Dispatcher, error) {
 	opts := newOptions(options...)
 	log.Debug().Str("upstream", opts.upstreamAddr).Msg("configured combined dispatcher")
 
-	if opts.prometheusSubsystem == "" {
-		opts.prometheusSubsystem = "dispatch_client"
+	if opts.metrics.PrometheusSubsystem == "" {
+		opts.metrics.PrometheusSubsystem = "dispatch_client"
 	}
 
-	cachingRedispatch, err := caching.NewCachingDispatcher(opts.cache, opts.metricsEnabled, opts.prometheusSubsystem, &keys.CanonicalKeyHandler{})
+	cachingRedispatch, err := caching.NewCachingDispatcher(opts.cache, opts.metrics, &keys.CanonicalKeyHandler{})
 	if err != nil {
 		return nil, err
 	}

internal/dispatch/combined/combined_test.go

@@ -4,10 +4,12 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 
 	"github.com/authzed/spicedb/internal/datastore/dsfortesting"
 	"github.com/authzed/spicedb/internal/datastore/memdb"
+	"github.com/authzed/spicedb/internal/dispatch"
 	"github.com/authzed/spicedb/internal/dispatch/graph"
 	"github.com/authzed/spicedb/internal/dispatch/keys"
 	"github.com/authzed/spicedb/internal/testfixtures"
@@ -84,9 +86,9 @@ func TestNewDispatcher_AppliesAllOptions_NoUpstream(t *testing.T) {
 
 	concurrencyLimits := graph.ConcurrencyLimits{Check: 10, LookupResources: 5}
 
+	reg := prometheus.NewRegistry()
 	options := []Option{
-		MetricsEnabled(true),
-		PrometheusSubsystem("test_subsystem"),
+		Metrics(dispatch.MetricsOptions{PrometheusSubsystem: "test_subsystem", PrometheusRegistry: reg}),
 		DispatchChunkSize(50),
 		RelationshipChunkCacheConfig(cacheConfig),
 		CaveatTypeSet(caveattypes.Default.TypeSet),
@@ -96,8 +98,8 @@ func TestNewDispatcher_AppliesAllOptions_NoUpstream(t *testing.T) {
 
 	// Field-level assertions: each setter wrote to the right slot.
 	opts := newOptions(options...)
-	require.True(t, opts.metricsEnabled)
-	require.Equal(t, "test_subsystem", opts.prometheusSubsystem)
+	require.Equal(t, reg, opts.metrics.PrometheusRegistry)
+	require.Equal(t, "test_subsystem", opts.metrics.PrometheusSubsystem)
 	require.Equal(t, uint16(50), opts.dispatchChunkSize)
 	require.Same(t, cacheConfig, opts.relationshipChunkCacheConfig)
 	require.Same(t, caveattypes.Default.TypeSet, opts.caveatTypeSet)

internal/dispatch/graph/check_test.go

@@ -2068,15 +2068,15 @@ func newLocalDispatcher(t testing.TB) (context.Context, dispatch.Dispatcher, dat
 
 	ds, revision := testfixtures.StandardDatastoreWithData(t, rawDS)
 
-	dispatch, err := NewLocalOnlyDispatcher(MustNewDefaultDispatcherParametersForTesting())
+	dispatcher, err := NewLocalOnlyDispatcher(MustNewDefaultDispatcherParametersForTesting())
 	require.NoError(t, err)
 	t.Cleanup(func() {
-		_ = dispatch.Close()
+		_ = dispatcher.Close()
 	})
 
-	cachingDispatcher, err := caching.NewCachingDispatcher(caching.DispatchTestCache(t), false, "", &keys.CanonicalKeyHandler{})
+	cachingDispatcher, err := caching.NewCachingDispatcher(caching.DispatchTestCache(t), dispatch.MetricsOptions{}, &keys.CanonicalKeyHandler{})
 	require.NoError(t, err)
-	cachingDispatcher.SetDelegate(dispatch)
+	cachingDispatcher.SetDelegate(dispatcher)
 	t.Cleanup(func() {
 		cachingDispatcher.Close()
 	})
@@ -2093,15 +2093,15 @@ func newLocalDispatcherWithSchemaAndRels(t testing.TB, schema string, rels []tup
 
 	ds, revision := testfixtures.DatastoreFromSchemaAndTestRelationships(t, rawDS, schema, rels)
 
-	dispatch, err := NewLocalOnlyDispatcher(MustNewDefaultDispatcherParametersForTesting())
+	dispatcher, err := NewLocalOnlyDispatcher(MustNewDefaultDispatcherParametersForTesting())
 	require.NoError(t, err)
 	t.Cleanup(func() {
-		dispatch.Close()
+		dispatcher.Close()
 	})
 
-	cachingDispatcher, err := caching.NewCachingDispatcher(caching.DispatchTestCache(t), false, "", &keys.CanonicalKeyHandler{})
+	cachingDispatcher, err := caching.NewCachingDispatcher(caching.DispatchTestCache(t), dispatch.MetricsOptions{}, &keys.CanonicalKeyHandler{})
 	require.NoError(t, err)
-	cachingDispatcher.SetDelegate(dispatch)
+	cachingDispatcher.SetDelegate(dispatcher)
 	t.Cleanup(func() {
 		cachingDispatcher.Close()
 	})

internal/dispatch/metrics.go

@@ -0,0 +1,23 @@
+package dispatch
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// MetricsOptions configures how a dispatcher reports Prometheus metrics.
+//
+// Metrics are only emitted when both fields are set.
+// The zero value disables metrics entirely.
+type MetricsOptions struct {
+	// PrometheusSubsystem is the subsystem name prepended to every metric
+	// emitted by the dispatcher (e.g. "dispatch" or "dispatch_client"). It
+	// namespaces the dispatcher's metrics so they don't collide with those of
+	// other components registered to the same registry.
+	PrometheusSubsystem string
+
+	// PrometheusRegistry is the registry the dispatcher's metrics are registered with.
+	PrometheusRegistry prometheus.Registerer
+}
+
+// Enabled reports whether metrics should be emitted.
+func (m *MetricsOptions) Enabled() bool {
+	return m.PrometheusSubsystem != "" && m.PrometheusRegistry != nil
+}

internal/services/integrationtesting/benchmark_dispatchqueryplan_test.go

@@ -181,7 +181,7 @@ func (h *dispatchQueryPlanHandle) newCachingDispatcher(b *testing.B) *caching.Di
 	c, err := cache.NewStandardCache[keys.DispatchCacheKey, any](cacheConfig)
 	require.NoError(b, err)
 
-	cd, err := caching.NewCachingDispatcher(c, false, "bench", &keys.DirectKeyHandler{})
+	cd, err := caching.NewCachingDispatcher(c, dispatch.MetricsOptions{}, &keys.DirectKeyHandler{})
 	require.NoError(b, err)
 
 	lpd := &localQueryPlanDispatcher{handle: h}