feat: Initial index-builder implementation #18297

benclive · 2025-07-01T14:13:13Z

What this PR does / why we need it:
Adds an initial implementation of an index-builder for logs stored in dataobjects.

It reads the loki.metastore-events kafka topic and when it gets a big enough group of events (configurable with a flag), it'll process them in a batch to generate an index object: an aggregated stream section + a bloom filter for every column in every log object.
Reading the log objects & building the index objects is extremely parallelisble and CPU bound. When I tested it in a dev environment it maxed out 10 vCPUs for the whole period and processed 32 objects in just over 4 minute. This should be fast enough for now.
I added a flag to only enable this for a specific set of tenants, because it doesn't (currently) handle multi-tenancy properly because it doesn't commit back to kafka in a multi-tenant aware way.
I chose to add this as part of the dataobj-consumer component, with flags to enable and disable it. I'm not sure I like this though so it might make more sense to create a new Loki component for it. That said, it's conceptually quite similar to the existing dataobj-consumer's partition-processor code.

Which issue(s) this PR fixes:
Part of https://github.com/grafana/loki-private/issues/1725

github-actions · 2025-07-01T14:50:25Z

💻 Deploy preview available:

https://deploy-preview-loki-18297-zb444pucvq-vp.a.run.app/docs/loki/latest/

chaudum

I only skimmed pkg/dataobj/consumer/indexing/indexobj/builder.go because it is dictated by the pointer section of the data object and contains a lot of (copied?) boilderplate code.

chaudum · 2025-07-02T09:31:18Z

pkg/dataobj/consumer/config.go

@@ -31,4 +41,9 @@ func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	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")
+	f.BoolVar(&cfg.PartitionProcessingEnabled, prefix+"partition-processing-enabled", true, "If true, partition processing will be enabled")


From the help text it is not clear to me what this setting (or rather "partition processing") is supposed to do. Do we need two separate config options to enable index building?

chaudum · 2025-07-02T09:33:34Z

pkg/dataobj/consumer/config.go

+	f.BoolVar(&cfg.PartitionProcessingEnabled, prefix+"partition-processing-enabled", true, "If true, partition processing will be enabled")
+	f.BoolVar(&cfg.IndexBuildingEnabled, prefix+"index-building-enabled", false, "If true, index building will be enabled")
+	f.IntVar(&cfg.IndexBuildingEventsPerIndex, prefix+"index-building-events-per-index", 32, "Experimental: The number of events to batch before building an index")
+	f.StringVar(&cfg.IndexStoragePrefix, prefix+"index-storage-prefix", "indexing-v0/", "Experimental: A prefix to use for storing indexes in object storage. Used to separate the metastore & index files during initial testing.")


"indexing-v0/" is a strange prefix IMO. Maybe call it dataobjindex-v0/ or (dataobjindex/v0/)?

chaudum · 2025-07-02T09:35:22Z

pkg/dataobj/consumer/config.go

+	f.BoolVar(&cfg.IndexBuildingEnabled, prefix+"index-building-enabled", false, "If true, index building will be enabled")
+	f.IntVar(&cfg.IndexBuildingEventsPerIndex, prefix+"index-building-events-per-index", 32, "Experimental: The number of events to batch before building an index")
+	f.StringVar(&cfg.IndexStoragePrefix, prefix+"index-storage-prefix", "indexing-v0/", "Experimental: A prefix to use for storing indexes in object storage. Used to separate the metastore & index files during initial testing.")
+	f.Var(&cfg.EnabledTenantIDs, prefix+"enabled-tenant-ids", "Experimental: A list of tenant IDs to enable index building for. If empty, all tenants will be enabled.")


This really should be a per-tenant setting.

chaudum · 2025-07-02T10:30:02Z

pkg/dataobj/consumer/indexing/index_builder.go

+			if err != nil {
+				level.Error(objLogger).Log("msg", "failed to read object", "err", err)
+				continue
+			}


I think there is a danger here: Failed downloads will get unnoticed, because only successfully downloaded objects are consumed when building the index.

chaudum · 2025-07-02T11:49:25Z

pkg/dataobj/consumer/service.go

+		indexBuilder := indexing.NewIndexBuilder(ctx, s.logger, s.eventConsumerClient, builderCfg, s.bucket, "loki.metastore-events", s.reg, s.cfg.IndexBuildingEventsPerIndex, s.cfg.IndexStoragePrefix, s.cfg.EnabledTenantIDs)
+		indexBuilder.Start()
+		defer indexBuilder.Stop()
+		for {


I would prefer if the logic for consuming and processing records would be encapsulated in the index builder.

chaudum · 2025-07-02T13:04:29Z

pkg/dataobj/consumer/indexing/index_builder.go

+	// Fetch the column statistics in order to init the bloom filters for each column
+	stats, err := logs.ReadStats(ctx, logsSection)
+	if err != nil {
+		level.Error(objLogger).Log("msg", "failed to read logs stats", "err", err)


I think you want to use the sectionLogger, do you?

Suggested change

level.Error(objLogger).Log("msg", "failed to read logs stats", "err", err)

level.Error(sectionLogger).Log("msg", "failed to read logs stats", "err", err)

Same for other Log() calls in this function

chaudum · 2025-07-02T13:10:43Z

pkg/dataobj/consumer/indexing/indexobj/builder_metrics.go

+			Namespace: "loki",
+			Subsystem: "indexobj",
+			Name:      "config_target_page_size_bytes",


Yesterday I learned that Namespace and Subsystem are going to be removed in future versions of prometheus/client_golang.

That said, that comment is from 2019 🤣

chaudum · 2025-07-02T13:13:30Z

pkg/dataobj/consumer/indexing/indexobj/builder_metrics.go

+		flushFailures: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: "loki",
+			Subsystem: "indexobj",
+			Name:      "flush_failures_total",
+
+			Help: "Total number of flush failures.",
+		}),


We should also have a counter metric which counts all flushes (loki_indexobj_flush_count_total or sth like that)

chaudum · 2025-07-02T13:16:07Z

pkg/dataobj/consumer/service.go

-	if err != nil {
-		level.Error(logger).Log("msg", "failed to create consumer", "err", err)
-		return nil
+	if cfg.PartitionProcessingEnabled {


Ah, now I understand the setting from the very first file of the diff. I think you need to improve the help text of the setting :)

chaudum · 2025-07-02T13:16:30Z

pkg/dataobj/consumer/service.go

+	if cfg.IndexBuildingEnabled {
+		consumerCfg := kafkaCfg
+		consumerCfg.AutoCreateTopicEnabled = true
+		consumerCfg.AutoCreateTopicDefaultPartitions = 1


benclive added 5 commits July 1, 2025 13:25

Add index-builder that processes metastore events

581c0c1

Undo unrelated changes

5d1d561

Move tenants to config

4a6e72d

Fixup tests

fe7c248

Remove todos

06ee080

benclive requested a review from a team as a code owner July 1, 2025 14:13

pull-request-size bot added the size/XXL label Jul 1, 2025

benclive changed the title ~~Benclive/prototype index builder~~ feat: Initial index-builder implementation Jul 1, 2025

benclive added 3 commits July 1, 2025 15:21

Linting

50a55fd

Format

b1e2cdb

docs

8575c99

chaudum reviewed Jul 2, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: Initial index-builder implementation #18297

feat: Initial index-builder implementation #18297

benclive commented Jul 1, 2025 •

edited

Loading

Uh oh!

github-actions bot commented Jul 1, 2025

Uh oh!

chaudum left a comment

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

chaudum Jul 2, 2025

Uh oh!

Uh oh!

	level.Error(objLogger).Log("msg", "failed to read logs stats", "err", err)
	level.Error(sectionLogger).Log("msg", "failed to read logs stats", "err", err)

feat: Initial index-builder implementation #18297

Are you sure you want to change the base?

feat: Initial index-builder implementation #18297

Conversation

benclive commented Jul 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions bot commented Jul 1, 2025

Uh oh!

chaudum left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

benclive commented Jul 1, 2025 •

edited

Loading