feat: tenant settings ruler (#3945)

* feat: tenant settings ruler

* doc

* use rlock for cache

* tenant settings client usage

* config opt

* flags refactor

* intermediate config object rules_source

Author: alsoba13

pkg/experiment/compactor/compaction_worker.go

@@ -57,12 +57,12 @@ type Worker struct {
 }
 
 type Config struct {
-	JobConcurrency         int           `yaml:"job_capacity"`
-	JobPollInterval        time.Duration `yaml:"job_poll_interval"`
-	SmallObjectSize        int           `yaml:"small_object_size_bytes"`
-	TempDir                string        `yaml:"temp_dir"`
-	RequestTimeout         time.Duration `yaml:"request_timeout"`
-	MetricsExporterEnabled bool          `yaml:"metrics_exporter_enabled"`
+	JobConcurrency        int            `yaml:"job_capacity"`
+	JobPollInterval       time.Duration  `yaml:"job_poll_interval"`
+	SmallObjectSize       int            `yaml:"small_object_size_bytes"`
+	TempDir               string         `yaml:"temp_dir"`
+	RequestTimeout        time.Duration  `yaml:"request_timeout"`
+	MetricsExporterConfig metrics.Config `yaml:"metrics_exporter_config"`
 }
 
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
@@ -72,7 +72,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.RequestTimeout, prefix+"request-timeout", 5*time.Second, "Job request timeout.")
 	f.IntVar(&cfg.SmallObjectSize, prefix+"small-object-size-bytes", 8<<20, "Size of the object that can be loaded in memory.")
 	f.StringVar(&cfg.TempDir, prefix+"temp-dir", os.TempDir(), "Temporary directory for compaction jobs.")
-	f.BoolVar(&cfg.MetricsExporterEnabled, prefix+"metrics-exporter.enabled", false, "This parameter specifies whether the metrics exporter is enabled.")
+	cfg.MetricsExporterConfig.RegisterFlags(f)
 }
 
 type compactionJob struct {
@@ -486,7 +486,7 @@ func (w *Worker) runCompaction(job *compactionJob) {
 }
 
 func (w *Worker) buildSampleObserver(md *metastorev1.BlockMeta) *metrics.SampleObserver {
-	if !w.config.MetricsExporterEnabled || md.CompactionLevel > 0 {
+	if !w.config.MetricsExporterConfig.Enabled || md.CompactionLevel > 0 {
 		return nil
 	}
 	recordingTime := int64(ulid.MustParse(md.Id).Time())

pkg/experiment/metrics/config.go

@@ -0,0 +1,23 @@
+package metrics
+
+import (
+	"flag"
+)
+
+type Config struct {
+	Enabled     bool `yaml:"enabled"`
+	RulesSource struct {
+		ClientAddress string `yaml:"client_address"`
+	} `yaml:"rules_source"`
+}
+
+func (c *Config) Validate() error {
+	return nil
+}
+
+func (c *Config) RegisterFlags(f *flag.FlagSet) {
+	const prefix = "compaction-worker.metrics-exporter."
+
+	f.BoolVar(&c.Enabled, prefix+"enabled", false, "This parameter specifies whether the metrics exporter is enabled.")
+	f.StringVar(&c.RulesSource.ClientAddress, prefix+"rules-source.client-address", "", "The address to use for the recording rules client connection.")
+}

pkg/experiment/metrics/ruler.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sync"
+	"time"
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
@@ -14,6 +16,7 @@ import (
 
 const (
 	envVarRecordingRules = "PYROSCOPE_RECORDING_RULES"
+	rulesExpiryTime      = time.Minute
 )
 
 type StaticRuler struct {
@@ -51,3 +54,92 @@ func NewStaticRulerFromEnvVars(logger log.Logger) (Ruler, error) {
 func (r StaticRuler) RecordingRules(tenant string) []*model.RecordingRule {
 	return r.rules[tenant]
 }
+
+// CachedRemoteRuler is a thread-safe ruler that retrieves rules from an external service.
+// It has a per-tenant cache: rulesPerTenant
+type CachedRemoteRuler struct {
+	rulesPerTenant map[string]*tenantCache
+	mu             sync.RWMutex
+
+	client RecordingRulesClient
+
+	logger log.Logger
+}
+
+type RecordingRulesClient interface {
+	RecordingRules(tenant string) ([]*model.RecordingRule, error)
+}
+
+func NewCachedRemoteRuler(client RecordingRulesClient, logger log.Logger) (Ruler, error) {
+	return &CachedRemoteRuler{
+		rulesPerTenant: make(map[string]*tenantCache),
+		client:         client,
+		logger:         logger,
+	}, nil
+}
+
+func (r *CachedRemoteRuler) RecordingRules(tenant string) []*model.RecordingRule {
+	// get the per-tenant cache
+	r.mu.RLock()
+	cache, ok := r.rulesPerTenant[tenant]
+	r.mu.RUnlock()
+
+	// There's no cache for given tenant: init it
+	if !ok {
+		r.mu.Lock()
+		defer r.mu.Unlock()
+
+		// only race-winner will initialize the per-tenant cache
+		cache, ok = r.rulesPerTenant[tenant]
+		if !ok {
+			cache = &tenantCache{
+				initFunc: func() ([]*model.RecordingRule, error) {
+					return r.client.RecordingRules(tenant)
+				},
+				logger: r.logger,
+			}
+			r.rulesPerTenant[tenant] = cache
+		}
+	}
+
+	// get data from cache:
+	return cache.get()
+}
+
+// tenantCache is a thread-safe cache that holds an expirable array of rules.
+type tenantCache struct {
+	value    []*model.RecordingRule
+	ttl      time.Time
+	initFunc func() ([]*model.RecordingRule, error)
+	mu       sync.RWMutex
+	logger   log.Logger
+}
+
+// get returns the stored value if present and not expired.
+// Otherwise, a single call to initFunc will be performed to retrieve the value and hold it for future calls within
+// the ttl.
+func (c *tenantCache) get() []*model.RecordingRule {
+	c.mu.RLock()
+	if c.value != nil && time.Now().Before(c.ttl) {
+		defer c.mu.RUnlock()
+		// value exists and didn't expired
+		return c.value
+	}
+	c.mu.RUnlock()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	// only race-winner will fetch the data
+	if c.value == nil || time.Now().After(c.ttl) {
+		value, err := c.initFunc()
+		if err != nil {
+			// keep old value and ttl, just log an error
+			level.Error(c.logger).Log("msg", "failed to fetch recording rules", "err", err)
+		} else {
+			c.value = value
+			c.ttl = time.Now().Add(rulesExpiryTime)
+		}
+	}
+	return c.value
+}

pkg/phlare/modules.go

@@ -92,18 +92,19 @@ const (
 
 	// Experimental modules
 
-	Metastore           string = "metastore"
-	MetastoreClient     string = "metastore-client"
-	MetastoreAdmin      string = "metastore-admin"
-	SegmentWriter       string = "segment-writer"
-	SegmentWriterRing   string = "segment-writer-ring"
-	SegmentWriterClient string = "segment-writer-client"
-	QueryBackend        string = "query-backend"
-	QueryBackendClient  string = "query-backend-client"
-	CompactionWorker    string = "compaction-worker"
-	PlacementAgent      string = "placement-agent"
-	PlacementManager    string = "placement-manager"
-	HealthServer        string = "health-server"
+	Metastore            string = "metastore"
+	MetastoreClient      string = "metastore-client"
+	MetastoreAdmin       string = "metastore-admin"
+	SegmentWriter        string = "segment-writer"
+	SegmentWriterRing    string = "segment-writer-ring"
+	SegmentWriterClient  string = "segment-writer-client"
+	QueryBackend         string = "query-backend"
+	QueryBackendClient   string = "query-backend-client"
+	CompactionWorker     string = "compaction-worker"
+	PlacementAgent       string = "placement-agent"
+	PlacementManager     string = "placement-manager"
+	HealthServer         string = "health-server"
+	RecordingRulesClient string = "recording-rules-client"
 )
 
 var objectStoreTypeStats = usagestats.NewString("store_object_type")

pkg/phlare/modules_experimental.go

@@ -27,6 +27,7 @@ import (
 	"github.com/grafana/pyroscope/pkg/experiment/metrics"
 	querybackend "github.com/grafana/pyroscope/pkg/experiment/query_backend"
 	querybackendclient "github.com/grafana/pyroscope/pkg/experiment/query_backend/client"
+	recordingrulesclient "github.com/grafana/pyroscope/pkg/settings/recording/client"
 	"github.com/grafana/pyroscope/pkg/util"
 	"github.com/grafana/pyroscope/pkg/util/health"
 )
@@ -101,11 +102,16 @@ func (f *Phlare) initCompactionWorker() (svc services.Service, err error) {
 
 	var ruler metrics.Ruler
 	var exporter metrics.Exporter
-	if f.Cfg.CompactionWorker.MetricsExporterEnabled {
-		ruler, err = metrics.NewStaticRulerFromEnvVars(f.logger)
+	if f.Cfg.CompactionWorker.MetricsExporterConfig.Enabled {
+		if f.recordingRulesClient != nil {
+			ruler, err = metrics.NewCachedRemoteRuler(f.recordingRulesClient, f.logger)
+		} else {
+			ruler, err = metrics.NewStaticRulerFromEnvVars(f.logger)
+		}
 		if err != nil {
 			return nil, err
 		}
+
 		exporter, err = metrics.NewStaticExporterFromEnvVars(f.logger, f.reg)
 		if err != nil {
 			return nil, err
@@ -224,6 +230,22 @@ func (f *Phlare) initQueryBackendClient() (services.Service, error) {
 	return c.Service(), nil
 }
 
+func (f *Phlare) initRecordingRulesClient() (services.Service, error) {
+	if err := f.Cfg.CompactionWorker.MetricsExporterConfig.Validate(); err != nil {
+		return nil, err
+	}
+	if !f.Cfg.CompactionWorker.MetricsExporterConfig.Enabled ||
+		f.Cfg.CompactionWorker.MetricsExporterConfig.RulesSource.ClientAddress == "" {
+		return nil, nil
+	}
+	c, err := recordingrulesclient.NewClient(f.Cfg.CompactionWorker.MetricsExporterConfig.RulesSource.ClientAddress, f.logger, f.auth)
+	if err != nil {
+		return nil, err
+	}
+	f.recordingRulesClient = c
+	return c.Service(), nil
+}
+
 func (f *Phlare) initPlacementAgent() (services.Service, error) {
 	f.placementAgent = adaptiveplacement.NewAgent(
 		f.logger,

pkg/phlare/phlare.go

@@ -68,6 +68,7 @@ import (
 	"github.com/grafana/pyroscope/pkg/scheduler"
 	"github.com/grafana/pyroscope/pkg/scheduler/schedulerdiscovery"
 	"github.com/grafana/pyroscope/pkg/settings"
+	recordingrulesclient "github.com/grafana/pyroscope/pkg/settings/recording/client"
 	"github.com/grafana/pyroscope/pkg/storegateway"
 	"github.com/grafana/pyroscope/pkg/tenant"
 	"github.com/grafana/pyroscope/pkg/tracing"
@@ -359,17 +360,18 @@ type Phlare struct {
 	frontend *frontend.Frontend
 
 	// Experimental modules.
-	segmentWriter       *segmentwriter.SegmentWriterService
-	segmentWriterClient *segmentwriterclient.Client
-	segmentWriterRing   *ring.Ring
-	placementAgent      *adaptiveplacement.Agent
-	placementManager    *adaptiveplacement.Manager
-	metastore           *metastore.Metastore
-	metastoreClient     *metastoreclient.Client
-	metastoreAdmin      *metastoreadmin.Admin
-	queryBackendClient  *querybackendclient.Client
-	compactionWorker    *compactionworker.Worker
-	healthServer        *health.Server
+	segmentWriter        *segmentwriter.SegmentWriterService
+	segmentWriterClient  *segmentwriterclient.Client
+	segmentWriterRing    *ring.Ring
+	placementAgent       *adaptiveplacement.Agent
+	placementManager     *adaptiveplacement.Manager
+	metastore            *metastore.Metastore
+	metastoreClient      *metastoreclient.Client
+	metastoreAdmin       *metastoreadmin.Admin
+	queryBackendClient   *querybackendclient.Client
+	compactionWorker     *compactionworker.Worker
+	healthServer         *health.Server
+	recordingRulesClient *recordingrulesclient.Client
 }
 
 func New(cfg Config) (*Phlare, error) {
@@ -485,7 +487,7 @@ func (f *Phlare) setupModuleManager() error {
 			SegmentWriter:       {Overrides, API, MemberlistKV, Storage, UsageReport, MetastoreClient},
 			Metastore:           {Overrides, API, MetastoreClient, Storage, PlacementManager},
 			MetastoreAdmin:      {API, MetastoreClient},
-			CompactionWorker:    {Overrides, API, Storage, MetastoreClient},
+			CompactionWorker:    {Overrides, API, Storage, MetastoreClient, RecordingRulesClient},
 			QueryBackend:        {Overrides, API, Storage, QueryBackendClient},
 			SegmentWriterRing:   {Overrides, API, MemberlistKV},
 			SegmentWriterClient: {Overrides, API, SegmentWriterRing, PlacementAgent},
@@ -515,6 +517,7 @@ func (f *Phlare) setupModuleManager() error {
 		mm.RegisterModule(PlacementAgent, f.initPlacementAgent, modules.UserInvisibleModule)
 		mm.RegisterModule(PlacementManager, f.initPlacementManager, modules.UserInvisibleModule)
 		mm.RegisterModule(HealthServer, f.initHealthServer, modules.UserInvisibleModule)
+		mm.RegisterModule(RecordingRulesClient, f.initRecordingRulesClient, modules.UserInvisibleModule)
 	}
 
 	for mod, targets := range deps {

pkg/settings/recording/client/client.go

@@ -0,0 +1,57 @@
+package recording
+
+import (
+	"context"
+
+	"connectrpc.com/connect"
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/grafana/dskit/services"
+
+	settingsv1 "github.com/grafana/pyroscope/api/gen/proto/go/settings/v1"
+	"github.com/grafana/pyroscope/api/gen/proto/go/settings/v1/settingsv1connect"
+	connectapi "github.com/grafana/pyroscope/pkg/api/connect"
+	phlaremodel "github.com/grafana/pyroscope/pkg/model"
+	"github.com/grafana/pyroscope/pkg/tenant"
+	"github.com/grafana/pyroscope/pkg/util"
+)
+
+type Client struct {
+	service services.Service
+	client  settingsv1connect.RecordingRulesServiceClient
+	logger  log.Logger
+}
+
+func NewClient(address string, logger log.Logger, auth connect.Option) (*Client, error) {
+	httpClient := util.InstrumentedDefaultHTTPClient()
+	opts := connectapi.DefaultClientOptions()
+	opts = append(opts, auth)
+	c := Client{
+		client: settingsv1connect.NewRecordingRulesServiceClient(httpClient, "http://"+address, opts...),
+		logger: logger,
+	}
+	c.service = services.NewIdleService(c.starting, c.stopping)
+	return &c, nil
+}
+
+func (b *Client) RecordingRules(tenantId string) ([]*phlaremodel.RecordingRule, error) {
+	ctx := tenant.InjectTenantID(context.Background(), tenantId)
+	resp, err := b.client.ListRecordingRules(ctx, connect.NewRequest(&settingsv1.ListRecordingRulesRequest{}))
+	if err != nil {
+		return nil, err
+	}
+	rules := make([]*phlaremodel.RecordingRule, 0, len(resp.Msg.Rules))
+	for _, rule := range resp.Msg.Rules {
+		r, err := phlaremodel.NewRecordingRule(rule)
+		if err == nil {
+			rules = append(rules, r)
+		} else {
+			level.Error(b.logger).Log("msg", "failed to parse recording rule", "rule", rule, "err", err)
+		}
+	}
+	return rules, nil
+}
+
+func (b *Client) Service() services.Service      { return b.service }
+func (b *Client) starting(context.Context) error { return nil }
+func (b *Client) stopping(error) error           { return nil }

pkg/settings/recording/config.go

@@ -26,6 +26,5 @@ func (cfg *Config) Validate() error {
 	if !cfg.Enabled {
 		return nil
 	}
-
 	return nil
 }