chore: Refactor dataobj index builder to handle buffered events from stale partitions #19189

periklis · 2025-09-12T15:43:47Z

What this PR does / why we need it:
This pull request is a medium sized refactoring of the dataobj index builder to support handling stale partitions in terms of buffered events per index but less than Config.EventsPerIndex. In particular:

All the build index code is moved to a separate abstraction in indexer.go that the builder is feeding via a golang channel.
The builder incorporates now a secondary async routine that flushes buffered event objects to the indexer according to configured flush timeout.

Which issue(s) this PR fixes:
Fixes grafana/loki-private#1967

Special notes for your reviewer:

Checklist

Reviewed the CONTRIBUTING.md guide (required)
Documentation added
Tests updated
Title matches the required conventional commits format, see here
- Note that Promtail is considered to be feature complete, and future development for logs collection will be in Grafana Alloy. As such, feat PRs are unlikely to be accepted unless a case can be made for the feature actually being a bug fix to existing behavior.
Changes that require user attention or interaction to upgrade are documented in docs/sources/setup/upgrade/_index.md
If the change is deprecating or removing a configuration option, update the deprecated-config.yaml and deleted-config.yaml files respectively in the tools/deprecated-config-checker directory. Example PR

github-actions · 2025-09-12T15:45:45Z

💻 Deploy preview available:

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

benclive

Thanks for this - it looks good apart from a couple of questions.

I had a thought about the approach: Would it be simpler to use a mutex or semaphore within buildIndex? That way you don't need to coordinate across goroutines and dispatch work, each entrypoint could just call buildIndex directly and wait for it complete. I may be missing some nuance, however!

benclive · 2025-09-16T09:23:20Z

pkg/dataobj/index/builder.go

+	switch tt {
+	case triggerTypeAppend:
+		return "append"
+	case triggerTypeFlush:


nit: maybe this should be "max-age" or something instead of flush?

Considering that the config is called max-idle-time we can say triggerTypeMaxIdle is the winner?

benclive · 2025-09-16T09:32:28Z

pkg/dataobj/index/builder.go

-			processingErrors.Add(fmt.Errorf("failed to download object: %w", obj.err))
-			continue
-		}
+	p.wg.Add(1)


Does p.wg.Wait() ever get called?

Yes p.wg.Wait() is called in stopping(). AFAICT my analysis is that this is ok:

Add(1) calls are called for the flush ticker routine and for each flush async partition routine

Done is called when the flush ticker routine and for each flush async partition routine exit

Wait is called in stopping waiting for all routines to end before closing the client.

Did I miss something here?

benclive · 2025-09-16T09:36:47Z

pkg/dataobj/index/builder.go

+			return nil, nil
+		}
+	case triggerTypeFlush:
+		if len(state.events) < p.cfg.MinFlushEvents {


I think this condition & flag can be removed. If something is older than the MaxIdleTime, we need to flush it anyway even if it means it'll be a small index.

benclive · 2025-09-16T09:57:35Z

pkg/dataobj/index/indexer.go

+	}
+
+	// Extract records for committing
+	records := make([]*kgo.Record, len(req.events))


very minor performance optimization, but records isn't used unless the build was successful so you could do this after the error check later

benclive · 2025-09-16T11:55:40Z

pkg/dataobj/index/indexer.go

+			// Successfully sent event for download
+		case <-ctx.Done():
+			return "", ctx.Err()
+		default:


is this default case needed?
If the channel is closed, the context should already have been cancelled and would be caught at the start of the loop

You are right for a moment a test was failing and that was a fix but never got back to review this properly.

benclive · 2025-09-16T11:59:44Z

pkg/dataobj/index/builder.go

 }

 func (p *Builder) cleanupPartition(partition int32) {
 	p.partitionsMutex.Lock()
 	defer p.partitionsMutex.Unlock()

-	p.cancelActiveCalculation(nil)
+	// Cancel active calculation for this partition
+	p.calculationsMutex.Lock()


I think calculations mutex is always acquired under the partitionsMutex - are they both needed?

You are right! The calculationMutex was in play because I use it also in stopping which does not rely on the partitionMutex. However now that the context propagation is refactor to pass through functions we can rely on the golang pattern let context cancelation drive cleanup.

periklis · 2025-09-16T13:16:54Z

I had a thought about the approach: Would it be simpler to use a mutex or semaphore within buildIndex? That way you don't need to coordinate across goroutines and dispatch work, each entrypoint could just call buildIndex directly and wait for it complete. I may be missing some nuance, however!

Practically yes you are right. However I decided to use channels for the buildWorker to stay similar to the downloadWorker. For now the limitation to keep only one buildWorker is CPU usage, but we may lift this and add more workers later. WDYT?

edit: one more thing that came into my mind when I built this is that downloading/processing are two independent queues so we could better observe where things go wrong/slow later.

periklis self-assigned this Sep 12, 2025

pull-request-size bot added the size/XXL label Sep 12, 2025

periklis marked this pull request as ready for review September 15, 2025 13:56

periklis requested a review from a team as a code owner September 15, 2025 13:56

periklis force-pushed the index-builder-correctness branch from 2bd1e56 to a5632c9 Compare September 16, 2025 07:46

benclive reviewed Sep 16, 2025

View reviewed changes

periklis added 11 commits September 16, 2025 15:17

chore: Add async flush for emitted index events

789e17e

Move to serial indexing

4933998

Fix metrics for new indexer

ae57abf

Fix docs and linter

5ee0b42

Fine tunings

b6d3975

Missing stringer impl

14cd015

Adapt default flush settings

ae6393a

Fix context.Context usage

aee0804

Apply code review suggestions

563ea08

Rename trigger flush to maxIdle

2812152

Fix linter

2a6babd

periklis force-pushed the index-builder-correctness branch from e9a7737 to 2a6babd Compare September 16, 2025 13:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

chore: Refactor dataobj index builder to handle buffered events from stale partitions #19189

chore: Refactor dataobj index builder to handle buffered events from stale partitions #19189

Uh oh!

periklis commented Sep 12, 2025

Uh oh!

github-actions bot commented Sep 12, 2025 •

edited

Loading

Uh oh!

benclive left a comment

Uh oh!

benclive Sep 16, 2025

Uh oh!

periklis Sep 16, 2025

Uh oh!

benclive Sep 16, 2025

Uh oh!

periklis Sep 16, 2025

Uh oh!

benclive Sep 16, 2025

Uh oh!

benclive Sep 16, 2025

Uh oh!

benclive Sep 16, 2025

Uh oh!

periklis Sep 16, 2025

Uh oh!

benclive Sep 16, 2025

Uh oh!

periklis Sep 16, 2025

Uh oh!

periklis commented Sep 16, 2025 •

edited

Loading

Uh oh!

Uh oh!

chore: Refactor dataobj index builder to handle buffered events from stale partitions #19189

Are you sure you want to change the base?

chore: Refactor dataobj index builder to handle buffered events from stale partitions #19189

Uh oh!

Conversation

periklis commented Sep 12, 2025

Uh oh!

github-actions bot commented Sep 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

benclive 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!

periklis commented Sep 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

github-actions bot commented Sep 12, 2025 •

edited

Loading

periklis commented Sep 16, 2025 •

edited

Loading