feat: add partition ring to dataobj consumer

Author: grobinson-grafana

docs/sources/shared/configuration.md

@@ -1252,6 +1252,66 @@ dataobj:
       # CLI flag: -dataobj-consumer.lifecycler.ID
       [id: <string> | default = "<hostname>"]
 
+    partition_ring:
+      # The key-value store used to share the hash ring across multiple
+      # instances. This option needs be set on ingesters, distributors,
+      # queriers, and rulers when running in microservices mode.
+      kvstore:
+        # Backend storage to use for the ring. Supported values are: consul,
+        # etcd, inmemory, memberlist, multi.
+        # CLI flag: -dataobj-consumer.partition-ring.store
+        [store: <string> | default = "memberlist"]
+
+        # The prefix for the keys in the store. Should end with a /.
+        # CLI flag: -dataobj-consumer.partition-ring.prefix
+        [prefix: <string> | default = "collectors/"]
+
+        # Configuration for a Consul client. Only applies if the selected
+        # kvstore is consul.
+        # The CLI flags prefix for this block configuration is:
+        # dataobj-consumer.partition-ring
+        [consul: <consul>]
+
+        # Configuration for an ETCD v3 client. Only applies if the selected
+        # kvstore is etcd.
+        # The CLI flags prefix for this block configuration is:
+        # dataobj-consumer.partition-ring
+        [etcd: <etcd>]
+
+        multi:
+          # Primary backend storage used by multi-client.
+          # CLI flag: -dataobj-consumer.partition-ring.multi.primary
+          [primary: <string> | default = ""]
+
+          # Secondary backend storage used by multi-client.
+          # CLI flag: -dataobj-consumer.partition-ring.multi.secondary
+          [secondary: <string> | default = ""]
+
+          # Mirror writes to secondary store.
+          # CLI flag: -dataobj-consumer.partition-ring.multi.mirror-enabled
+          [mirror_enabled: <boolean> | default = false]
+
+          # Timeout for storing value to secondary store.
+          # CLI flag: -dataobj-consumer.partition-ring.multi.mirror-timeout
+          [mirror_timeout: <duration> | default = 2s]
+
+      # Minimum number of owners to wait before a PENDING partition gets
+      # switched to ACTIVE.
+      # CLI flag: -dataobj-consumer.partition-ring.min-partition-owners-count
+      [min_partition_owners_count: <int> | default = 1]
+
+      # How long the minimum number of owners are enforced before a PENDING
+      # partition gets switched to ACTIVE.
+      # CLI flag: -dataobj-consumer.partition-ring.min-partition-owners-duration
+      [min_partition_owners_duration: <duration> | default = 10s]
+
+      # How long to wait before an INACTIVE partition is eligible for deletion.
+      # The partition is deleted only if it has been in INACTIVE state for at
+      # least the configured duration and it has no owners registered. A value
+      # of 0 disables partitions deletion.
+      # CLI flag: -dataobj-consumer.partition-ring.delete-inactive-partition-after
+      [delete_inactive_partition_after: <duration> | default = 13h]
+
     uploader:
       # The size of the SHA prefix to use for generating object storage keys for
       # data objects.
@@ -2824,6 +2884,7 @@ Configuration for a Consul client. Only applies if the selected kvstore is `cons
 - `common.storage.ring`
 - `compactor.ring`
 - `dataobj-consumer`
+- `dataobj-consumer.partition-ring`
 - `distributor.ring`
 - `index-gateway.ring`
 - `ingest-limits`
@@ -3075,6 +3136,7 @@ Configuration for an ETCD v3 client. Only applies if the selected kvstore is `et
 - `common.storage.ring`
 - `compactor.ring`
 - `dataobj-consumer`
+- `dataobj-consumer.partition-ring`
 - `distributor.ring`
 - `index-gateway.ring`
 - `ingest-limits`
@@ -7498,6 +7560,7 @@ The TLS configuration. The supported CLI flags `<prefix>` used to reference this
 - `compactor.grpc-client`
 - `compactor.ring.etcd`
 - `dataobj-consumer.etcd`
+- `dataobj-consumer.partition-ring.etcd`
 - `distributor.ring.etcd`
 - `etcd`
 - `frontend.grpc-client-config`

pkg/dataobj/consumer/config.go

@@ -8,14 +8,16 @@ import (
 
 	"github.com/grafana/loki/v3/pkg/dataobj/consumer/logsobj"
 	"github.com/grafana/loki/v3/pkg/dataobj/uploader"
+	"github.com/grafana/loki/v3/pkg/kafka/partitionring"
 	util_log "github.com/grafana/loki/v3/pkg/util/log"
 )
 
 type Config struct {
-	BuilderConfig    logsobj.BuilderConfig `yaml:"builder,omitempty"`
-	LifecyclerConfig ring.LifecyclerConfig `yaml:"lifecycler,omitempty"`
-	UploaderConfig   uploader.Config       `yaml:"uploader"`
-	IdleFlushTimeout time.Duration         `yaml:"idle_flush_timeout"`
+	BuilderConfig       logsobj.BuilderConfig `yaml:"builder,omitempty"`
+	LifecyclerConfig    ring.LifecyclerConfig `yaml:"lifecycler,omitempty"`
+	PartitionRingConfig partitionring.Config  `yaml:"partition_ring" category:"experimental"`
+	UploaderConfig      uploader.Config       `yaml:"uploader"`
+	IdleFlushTimeout    time.Duration         `yaml:"idle_flush_timeout"`
 }
 
 func (cfg *Config) Validate() error {
@@ -38,6 +40,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	cfg.BuilderConfig.RegisterFlagsWithPrefix(prefix, f)
 	cfg.LifecyclerConfig.RegisterFlagsWithPrefix(prefix, f, util_log.Logger)
+	cfg.PartitionRingConfig.RegisterFlagsWithPrefix(prefix, f)
 	cfg.UploaderConfig.RegisterFlagsWithPrefix(prefix, f)
 
 	f.DurationVar(&cfg.IdleFlushTimeout, prefix+"idle-flush-timeout", 60*60*time.Second, "The maximum amount of time to wait in seconds before flushing an object that is no longer receiving new writes")

pkg/loki/config_wrapper.go

@@ -379,6 +379,7 @@ func applyConfigToRings(r, defaults *ConfigWrapper, rc lokiring.RingConfig, merg
 		r.DataObj.Consumer.LifecyclerConfig.ListenPort = rc.ListenPort
 		r.DataObj.Consumer.LifecyclerConfig.ObservePeriod = rc.ObservePeriod
 		r.DataObj.Consumer.LifecyclerConfig.EnableInet6 = rc.EnableIPv6
+		r.DataObj.Consumer.PartitionRingConfig.KVStore = rc.KVStore
 	}
 }
 

pkg/loki/loki.go

@@ -413,51 +413,53 @@ type Loki struct {
 	deps          map[string][]string
 	SignalHandler *signals.Handler
 
-	Server                    *server.Server
-	InternalServer            *server.Server
-	UI                        *ui.Service
-	ring                      *ring.Ring
-	Overrides                 limiter.CombinedLimits
-	tenantConfigs             *runtime.TenantConfigs
-	TenantLimits              validation.TenantLimits
-	distributor               *distributor.Distributor
-	ingestLimits              *limits.Service
-	ingestLimitsRing          *ring.Ring
-	ingestLimitsFrontend      *limits_frontend.Frontend
-	ingestLimitsFrontendRing  *ring.Ring
-	Ingester                  ingester.Interface
-	PatternIngester           *pattern.Ingester
-	PatternRingClient         pattern.RingClient
-	Querier                   querier.Querier
-	cacheGenerationLoader     queryrangebase.CacheGenNumberLoader
-	querierAPI                *querier.QuerierAPI
-	ingesterQuerier           *querier.IngesterQuerier
-	Store                     storage.Store
-	BloomStore                bloomshipper.Store
-	bloomGatewayClient        bloomgateway.Client
-	tableManager              *index.TableManager
-	frontend                  Frontend
-	ruler                     *base_ruler.Ruler
-	ruleEvaluator             ruler.Evaluator
-	RulerStorage              rulestore.RuleStore
-	rulerAPI                  *base_ruler.API
-	stopper                   queryrange.Stopper
-	runtimeConfig             *runtimeconfig.Manager
-	MemberlistKV              *memberlist.KVInitService
-	compactor                 *compactor.Compactor
-	QueryFrontEndMiddleware   queryrangebase.Middleware
-	queryScheduler            *scheduler.Scheduler
-	querySchedulerRingManager *lokiring.RingManager
-	usageReport               *analytics.Reporter
-	indexGatewayRingManager   *lokiring.RingManager
-	PartitionRingWatcher      *ring.PartitionRingWatcher
-	partitionRing             *ring.PartitionInstanceRing
-	blockBuilder              *blockbuilder.BlockBuilder
-	blockScheduler            *blockscheduler.BlockScheduler
-	dataObjConsumer           *consumer.Service
-	dataObjConsumerRing       *ring.Ring
-	dataObjIndexBuilder       *dataobjindex.Builder
-	scratchStore              scratch.Store
+	Server                              *server.Server
+	InternalServer                      *server.Server
+	UI                                  *ui.Service
+	ring                                *ring.Ring
+	Overrides                           limiter.CombinedLimits
+	tenantConfigs                       *runtime.TenantConfigs
+	TenantLimits                        validation.TenantLimits
+	distributor                         *distributor.Distributor
+	ingestLimits                        *limits.Service
+	ingestLimitsRing                    *ring.Ring
+	ingestLimitsFrontend                *limits_frontend.Frontend
+	ingestLimitsFrontendRing            *ring.Ring
+	Ingester                            ingester.Interface
+	PatternIngester                     *pattern.Ingester
+	PatternRingClient                   pattern.RingClient
+	Querier                             querier.Querier
+	cacheGenerationLoader               queryrangebase.CacheGenNumberLoader
+	querierAPI                          *querier.QuerierAPI
+	ingesterQuerier                     *querier.IngesterQuerier
+	Store                               storage.Store
+	BloomStore                          bloomshipper.Store
+	bloomGatewayClient                  bloomgateway.Client
+	tableManager                        *index.TableManager
+	frontend                            Frontend
+	ruler                               *base_ruler.Ruler
+	ruleEvaluator                       ruler.Evaluator
+	RulerStorage                        rulestore.RuleStore
+	rulerAPI                            *base_ruler.API
+	stopper                             queryrange.Stopper
+	runtimeConfig                       *runtimeconfig.Manager
+	MemberlistKV                        *memberlist.KVInitService
+	compactor                           *compactor.Compactor
+	QueryFrontEndMiddleware             queryrangebase.Middleware
+	queryScheduler                      *scheduler.Scheduler
+	querySchedulerRingManager           *lokiring.RingManager
+	usageReport                         *analytics.Reporter
+	indexGatewayRingManager             *lokiring.RingManager
+	PartitionRingWatcher                *ring.PartitionRingWatcher
+	partitionRing                       *ring.PartitionInstanceRing
+	blockBuilder                        *blockbuilder.BlockBuilder
+	blockScheduler                      *blockscheduler.BlockScheduler
+	dataObjConsumer                     *consumer.Service
+	dataObjConsumerRing                 *ring.Ring
+	dataObjConsumerPartitionRing        *ring.PartitionInstanceRing
+	DataObjConsumerPartitionRingWatcher *ring.PartitionRingWatcher
+	dataObjIndexBuilder                 *dataobjindex.Builder
+	scratchStore                        scratch.Store
 
 	ClientMetrics       storage.ClientMetrics
 	deleteClientMetrics *deletion.DeleteRequestClientMetrics
@@ -800,6 +802,7 @@ func (t *Loki) setupModuleManager() error {
 	mm.RegisterModule(DataObjExplorer, t.initDataObjExplorer)
 	mm.RegisterModule(UI, t.initUI)
 	mm.RegisterModule(DataObjConsumerRing, t.initDataObjConsumerRing)
+	mm.RegisterModule(DataObjConsumerPartitionRing, t.initDataObjConsumerPartitionRing)
 	mm.RegisterModule(DataObjConsumer, t.initDataObjConsumer)
 	mm.RegisterModule(DataObjIndexBuilder, t.initDataObjIndexBuilder)
 	mm.RegisterModule(ScratchStore, t.initScratchStore)
@@ -811,47 +814,48 @@ func (t *Loki) setupModuleManager() error {
 
 	// Add dependencies
 	deps := map[string][]string{
-		Ring:                     {RuntimeConfig, Server, MemberlistKV},
-		Analytics:                {},
-		Overrides:                {RuntimeConfig},
-		OverridesExporter:        {Overrides, Server, UI},
-		TenantConfigs:            {RuntimeConfig},
-		UI:                       {Server},
-		Distributor:              {Ring, Server, Overrides, TenantConfigs, PatternRingClient, PatternIngesterTee, Analytics, PartitionRing, IngestLimitsFrontendRing, DataObjConsumerRing, UI},
-		IngestLimitsRing:         {RuntimeConfig, Server, MemberlistKV},
-		IngestLimits:             {MemberlistKV, Overrides, Server},
-		IngestLimitsFrontend:     {IngestLimitsRing, Overrides, Server, MemberlistKV},
-		IngestLimitsFrontendRing: {RuntimeConfig, Server, MemberlistKV},
-		Store:                    {Overrides, IndexGatewayRing},
-		Ingester:                 {Store, Server, MemberlistKV, TenantConfigs, Analytics, PartitionRing, UI},
-		Querier:                  {Store, Ring, Server, IngesterQuerier, PatternRingClient, Overrides, Analytics, CacheGenerationLoader, QuerySchedulerRing, UI},
-		QueryFrontendTripperware: {Server, Overrides, TenantConfigs},
-		QueryFrontend:            {QueryFrontendTripperware, Analytics, CacheGenerationLoader, QuerySchedulerRing, UI},
-		QueryScheduler:           {Server, Overrides, MemberlistKV, Analytics, QuerySchedulerRing, UI},
-		Ruler:                    {Ring, Server, RulerStorage, RuleEvaluator, Overrides, TenantConfigs, Analytics, UI},
-		RuleEvaluator:            {Ring, Server, Store, IngesterQuerier, Overrides, TenantConfigs, Analytics},
-		TableManager:             {Server, Analytics, UI},
-		Compactor:                {Server, Overrides, MemberlistKV, Analytics, UI},
-		IndexGateway:             {Server, Store, BloomStore, IndexGatewayRing, IndexGatewayInterceptors, Analytics, UI},
-		BloomGateway:             {Server, BloomStore, Analytics, UI},
-		BloomPlanner:             {Server, BloomStore, Analytics, Store, UI},
-		BloomBuilder:             {Server, BloomStore, Analytics, Store, UI},
-		BloomStore:               {IndexGatewayRing, BloomGatewayClient},
-		PatternRingClient:        {Server, MemberlistKV, Analytics},
-		PatternIngesterTee:       {Server, Overrides, MemberlistKV, Analytics, PatternRingClient},
-		PatternIngester:          {Server, MemberlistKV, Analytics, PatternRingClient, PatternIngesterTee, Overrides, UI},
-		IngesterQuerier:          {Ring, PartitionRing, Overrides},
-		QuerySchedulerRing:       {Overrides, MemberlistKV},
-		IndexGatewayRing:         {Overrides, MemberlistKV},
-		PartitionRing:            {MemberlistKV, Server, Ring},
-		MemberlistKV:             {Server},
-		BlockBuilder:             {PartitionRing, Store, Server, UI},
-		BlockScheduler:           {Server, UI},
-		DataObjExplorer:          {Server, UI},
-		DataObjConsumerRing:      {RuntimeConfig, Server, MemberlistKV},
-		DataObjConsumer:          {MemberlistKV, ScratchStore, PartitionRing, Server, UI},
-		DataObjIndexBuilder:      {ScratchStore, Server, UI},
-		ScratchStore:             {},
+		Ring:                         {RuntimeConfig, Server, MemberlistKV},
+		Analytics:                    {},
+		Overrides:                    {RuntimeConfig},
+		OverridesExporter:            {Overrides, Server, UI},
+		TenantConfigs:                {RuntimeConfig},
+		UI:                           {Server},
+		Distributor:                  {Ring, Server, Overrides, TenantConfigs, PatternRingClient, PatternIngesterTee, Analytics, PartitionRing, IngestLimitsFrontendRing, DataObjConsumerRing, UI},
+		IngestLimitsRing:             {RuntimeConfig, Server, MemberlistKV},
+		IngestLimits:                 {MemberlistKV, Overrides, Server},
+		IngestLimitsFrontend:         {IngestLimitsRing, Overrides, Server, MemberlistKV},
+		IngestLimitsFrontendRing:     {RuntimeConfig, Server, MemberlistKV},
+		Store:                        {Overrides, IndexGatewayRing},
+		Ingester:                     {Store, Server, MemberlistKV, TenantConfigs, Analytics, PartitionRing, UI},
+		Querier:                      {Store, Ring, Server, IngesterQuerier, PatternRingClient, Overrides, Analytics, CacheGenerationLoader, QuerySchedulerRing, UI},
+		QueryFrontendTripperware:     {Server, Overrides, TenantConfigs},
+		QueryFrontend:                {QueryFrontendTripperware, Analytics, CacheGenerationLoader, QuerySchedulerRing, UI},
+		QueryScheduler:               {Server, Overrides, MemberlistKV, Analytics, QuerySchedulerRing, UI},
+		Ruler:                        {Ring, Server, RulerStorage, RuleEvaluator, Overrides, TenantConfigs, Analytics, UI},
+		RuleEvaluator:                {Ring, Server, Store, IngesterQuerier, Overrides, TenantConfigs, Analytics},
+		TableManager:                 {Server, Analytics, UI},
+		Compactor:                    {Server, Overrides, MemberlistKV, Analytics, UI},
+		IndexGateway:                 {Server, Store, BloomStore, IndexGatewayRing, IndexGatewayInterceptors, Analytics, UI},
+		BloomGateway:                 {Server, BloomStore, Analytics, UI},
+		BloomPlanner:                 {Server, BloomStore, Analytics, Store, UI},
+		BloomBuilder:                 {Server, BloomStore, Analytics, Store, UI},
+		BloomStore:                   {IndexGatewayRing, BloomGatewayClient},
+		PatternRingClient:            {Server, MemberlistKV, Analytics},
+		PatternIngesterTee:           {Server, Overrides, MemberlistKV, Analytics, PatternRingClient},
+		PatternIngester:              {Server, MemberlistKV, Analytics, PatternRingClient, PatternIngesterTee, Overrides, UI},
+		IngesterQuerier:              {Ring, PartitionRing, Overrides},
+		QuerySchedulerRing:           {Overrides, MemberlistKV},
+		IndexGatewayRing:             {Overrides, MemberlistKV},
+		PartitionRing:                {MemberlistKV, Server, Ring},
+		MemberlistKV:                 {Server},
+		BlockBuilder:                 {PartitionRing, Store, Server, UI},
+		BlockScheduler:               {Server, UI},
+		DataObjExplorer:              {Server, UI},
+		DataObjConsumerRing:          {RuntimeConfig, Server, MemberlistKV},
+		DataObjConsumerPartitionRing: {MemberlistKV, Server, Ring},
+		DataObjConsumer:              {MemberlistKV, ScratchStore, DataObjConsumerPartitionRing, Server, UI},
+		DataObjIndexBuilder:          {ScratchStore, Server, UI},
+		ScratchStore:                 {},
 
 		Read:    {QueryFrontend, Querier},
 		Write:   {Ingester, Distributor, PatternIngester},