feat: wire up ingest-limits service (#16660)

Author: grobinson-grafana

docs/sources/shared/configuration.md

@@ -5,6 +5,7 @@ package client
 
 import (
 	"flag"
+	"fmt"
 	"io"
 	"time"
 
@@ -62,6 +63,13 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	cfg.PoolConfig.RegisterFlagsWithPrefix(prefix, f)
 }
 
+func (cfg *Config) Validate() error {
+	if err := cfg.GRPCClientConfig.Validate(); err != nil {
+		return fmt.Errorf("invalid gRPC client config: %w", err)
+	}
+	return nil
+}
+
 // PoolConfig contains the config for a pool of ingest-limits-frontend clients.
 type PoolConfig struct {
 	ClientCleanupPeriod     time.Duration `yaml:"client_cleanup_period"`

pkg/limits/frontend/client/client.go

@@ -151,6 +151,20 @@ func (f *Frontend) ExceedsLimits(ctx context.Context, r *logproto.ExceedsLimitsR
 	return f.limits.ExceedsLimits(ctx, r)
 }
 
+func (f *Frontend) CheckReady(ctx context.Context) error {
+	if f.State() != services.Running && f.State() != services.Stopping {
+		return fmt.Errorf("ingest limits frontend not ready: %v", f.State())
+	}
+
+	err := f.lifecycler.CheckReady(ctx)
+	if err != nil {
+		level.Error(f.logger).Log("msg", "ingest limits frontend not ready", "err", err)
+		return err
+	}
+
+	return nil
+}
+
 type exceedsLimitsRequest struct {
 	TenantID     string   `json:"tenantID"`
 	StreamHashes []uint64 `json:"streamHashes"`

pkg/limits/frontend/frontend.go

@@ -85,6 +85,10 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.NumPartitions, "ingest-limits.num-partitions", 64, "The number of partitions for the Kafka topic used to read and write stream metadata. It is fixed, not a maximum.")
 }
 
+func (cfg *Config) Validate() error {
+	return nil
+}
+
 type metrics struct {
 	tenantStreamEvictionsTotal *prometheus.CounterVec
 

pkg/limits/ingest_limits.go

@@ -262,6 +262,38 @@ func applyConfigToRings(r, defaults *ConfigWrapper, rc lokiring.RingConfig, merg
 		r.Pattern.LifecyclerConfig.ObservePeriod = rc.ObservePeriod
 	}
 
+	// IngestLimits
+	if mergeWithExisting || reflect.DeepEqual(r.IngestLimits.LifecyclerConfig.RingConfig, defaults.IngestLimits.LifecyclerConfig.RingConfig) {
+		r.IngestLimits.LifecyclerConfig.RingConfig.KVStore = rc.KVStore
+		r.IngestLimits.LifecyclerConfig.HeartbeatPeriod = rc.HeartbeatPeriod
+		r.IngestLimits.LifecyclerConfig.RingConfig.HeartbeatTimeout = rc.HeartbeatTimeout
+		r.IngestLimits.LifecyclerConfig.TokensFilePath = rc.TokensFilePath
+		r.IngestLimits.LifecyclerConfig.RingConfig.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled
+		r.IngestLimits.LifecyclerConfig.ID = rc.InstanceID
+		r.IngestLimits.LifecyclerConfig.InfNames = rc.InstanceInterfaceNames
+		r.IngestLimits.LifecyclerConfig.Port = rc.InstancePort
+		r.IngestLimits.LifecyclerConfig.Addr = rc.InstanceAddr
+		r.IngestLimits.LifecyclerConfig.Zone = rc.InstanceZone
+		r.IngestLimits.LifecyclerConfig.ListenPort = rc.ListenPort
+		r.IngestLimits.LifecyclerConfig.ObservePeriod = rc.ObservePeriod
+	}
+
+	// IngestLimitsFrontend
+	if mergeWithExisting || reflect.DeepEqual(r.IngestLimitsFrontend.LifecyclerConfig.RingConfig, defaults.IngestLimitsFrontend.LifecyclerConfig.RingConfig) {
+		r.IngestLimitsFrontend.LifecyclerConfig.RingConfig.KVStore = rc.KVStore
+		r.IngestLimitsFrontend.LifecyclerConfig.HeartbeatPeriod = rc.HeartbeatPeriod
+		r.IngestLimitsFrontend.LifecyclerConfig.RingConfig.HeartbeatTimeout = rc.HeartbeatTimeout
+		r.IngestLimitsFrontend.LifecyclerConfig.TokensFilePath = rc.TokensFilePath
+		r.IngestLimitsFrontend.LifecyclerConfig.RingConfig.ZoneAwarenessEnabled = rc.ZoneAwarenessEnabled
+		r.IngestLimitsFrontend.LifecyclerConfig.ID = rc.InstanceID
+		r.IngestLimitsFrontend.LifecyclerConfig.InfNames = rc.InstanceInterfaceNames
+		r.IngestLimitsFrontend.LifecyclerConfig.Port = rc.InstancePort
+		r.IngestLimitsFrontend.LifecyclerConfig.Addr = rc.InstanceAddr
+		r.IngestLimitsFrontend.LifecyclerConfig.Zone = rc.InstanceZone
+		r.IngestLimitsFrontend.LifecyclerConfig.ListenPort = rc.ListenPort
+		r.IngestLimitsFrontend.LifecyclerConfig.ObservePeriod = rc.ObservePeriod
+	}
+
 	// Distributor
 	if mergeWithExisting || reflect.DeepEqual(r.Distributor.DistributorRing, defaults.Distributor.DistributorRing) {
 		r.Distributor.DistributorRing.HeartbeatTimeout = rc.HeartbeatTimeout
@@ -332,6 +364,20 @@ func applyTokensFilePath(cfg *ConfigWrapper) error {
 	}
 	cfg.Ingester.LifecyclerConfig.TokensFilePath = f
 
+	// IngestLimits
+	f, err = tokensFile(cfg, "ingestlimits.tokens")
+	if err != nil {
+		return err
+	}
+	cfg.IngestLimits.LifecyclerConfig.TokensFilePath = f
+
+	// IngestLimitsFrontend
+	f, err = tokensFile(cfg, "ingestlimitsfrontend.tokens")
+	if err != nil {
+		return err
+	}
+	cfg.IngestLimitsFrontend.LifecyclerConfig.TokensFilePath = f
+
 	// Compactor
 	f, err = tokensFile(cfg, "compactor.tokens")
 	if err != nil {
@@ -414,6 +460,15 @@ func appendLoopbackInterface(cfg, defaults *ConfigWrapper) {
 	if reflect.DeepEqual(cfg.Ingester.LifecyclerConfig.InfNames, defaults.Ingester.LifecyclerConfig.InfNames) {
 		cfg.Ingester.LifecyclerConfig.InfNames = append(cfg.Ingester.LifecyclerConfig.InfNames, loopbackIface)
 	}
+
+	if reflect.DeepEqual(cfg.IngestLimits.LifecyclerConfig.InfNames, defaults.IngestLimits.LifecyclerConfig.InfNames) {
+		cfg.IngestLimits.LifecyclerConfig.InfNames = append(cfg.IngestLimits.LifecyclerConfig.InfNames, loopbackIface)
+	}
+
+	if reflect.DeepEqual(cfg.IngestLimitsFrontend.LifecyclerConfig.InfNames, defaults.IngestLimitsFrontend.LifecyclerConfig.InfNames) {
+		cfg.IngestLimitsFrontend.LifecyclerConfig.InfNames = append(cfg.IngestLimitsFrontend.LifecyclerConfig.InfNames, loopbackIface)
+	}
+
 	if reflect.DeepEqual(cfg.Pattern.LifecyclerConfig.InfNames, defaults.Pattern.LifecyclerConfig.InfNames) {
 		cfg.Pattern.LifecyclerConfig.InfNames = append(cfg.Pattern.LifecyclerConfig.InfNames, loopbackIface)
 	}
@@ -453,6 +508,8 @@ func appendLoopbackInterface(cfg, defaults *ConfigWrapper) {
 // (for example, use consul for the distributor), it seems harmless to take a guess at better defaults here.
 func applyMemberlistConfig(r *ConfigWrapper) {
 	r.Ingester.LifecyclerConfig.RingConfig.KVStore.Store = memberlistStr
+	r.IngestLimits.LifecyclerConfig.RingConfig.KVStore.Store = memberlistStr
+	r.IngestLimitsFrontend.LifecyclerConfig.RingConfig.KVStore.Store = memberlistStr
 	r.Pattern.LifecyclerConfig.RingConfig.KVStore.Store = memberlistStr
 	r.Distributor.DistributorRing.KVStore.Store = memberlistStr
 	r.Ruler.Ring.KVStore.Store = memberlistStr