[chore][tracker]: save most recent (archive) write index to disk #36799

VihasMakwana · 2024-12-12T11:36:36Z

This PR stores the most recent index to disk. Much similar to what happens for persistent queue. It also adds Batch methods to operator.Persister, as saving the metadata and saving the index should be a transaction and it can only be achieved via Batch.

For eg. if user has configured archiving to store 100 poll cycles, let's assume:

For first collector run, it stores 10 cycles and archiveIndex is 11 (pointing to the next index).
When the collector is restarted, we will restore the archiveIndex from disk and continue from index 11

Link to tracking issue

Related #32727

Testing

Added UT for checking index

VihasMakwana · 2024-12-12T11:37:11Z

pkg/stanza/fileconsumer/internal/checkpoint/checkpoint.go

-
-	if err := persister.Set(ctx, key, buf.Bytes()); err != nil {
+	ops = append(ops, storage.SetOperation(key, buf.Bytes()))
+	if err := persister.Batch(ctx, ops...); err != nil {


For existing usage, this will be a no-op.

pkg/stanza/fileconsumer/internal/tracker/tracker.go

djaglowski · 2024-12-13T17:37:58Z

pkg/stanza/fileconsumer/internal/tracker/tracker.go

+		// It's best if we reset the index or else we might end up writing invalid keys
+		t.set.Logger.Warn("the read index was found, but it exceeds the bounds. Starting from 0")
+		t.archiveIndex = 0
+	}


Good idea to check for this case.

However, I wonder if we can handle it better than restarting from zero. What would it take to search the archive for the most recently updated?

I think we could maintain some kind of data structure which notes the time each archive was written. Maybe just map[index]time.Time. Then when we first create the tracker, we can load this up and find the most recent timestamp. We can also check for the case where pollsToArchive has changed and then rewrite the storage to align with the new value.

For example, if we previously saved 10 archives and find that pollsToArchive is now 5, we can find the 5 most recent indices based on the timestamp structure, then rewrite the archive files so that these are 0-4. We should probably even delete the extras from storage as well.

@djaglowski This solution does makes sense to me, but it becomes tricky when we eventually overwrite old archive data, as it is a ring buffer.
We might need to load the filesets in memory.
I'll find a few ways.

it becomes tricky when we eventually overwrite old archive data, as it is a ring buffer.

Can you elaborate?

We might need to load the filesets in memory.

If it's more than one at a time then it defeats the point of the archive.

Can you elaborate?

Consider this archive,

We've rolled over once and the latest data is at index 4 and archiveIndex (i.e. where the next data will be written) is at index 5.

Let's suppose that new polls_to_archive is 7.
We now need to construct a new, smaller archive with 7 most recent elements.
These elements are (from most recent to least recent):

14, 13, 12, 11, 10, 9, 8

We cannot simply rewrite archive in-place without caching values.

It would be much simpler to convert archive like following image,

and we would delete excess data.

Wdyt?

What would it take to search the archive for the most recently updated?

It would always be data stored at archiveIndex-1 index. We will store archiveIndex on disk, so in next collector run, we would load that value and we can find most recent data.

archiveIndex points at the next location where data will be written.
This can point to either of following:

Least recent data

Pointing to an empty slot (archive is partially filled)

pkg/stanza/fileconsumer/internal/tracker/tracker.go

Co-authored-by: Daniel Jaglowski <[email protected]>

VihasMakwana · 2024-12-23T14:59:59Z

@djaglowski I've added documentation and implemented the archive restoration.

I think adding a new data structure like map[index]time.Time will add to the complexity if we've rolled over and overwriting older data.

We can accomplish archive restoration without any new data structure and I've added a document to highlight this. Please take a look and let me know your thoughts.

djaglowski

Can we move the documentation changes to another PR?

djaglowski · 2025-01-06T16:51:25Z

pkg/stanza/fileconsumer/design/design.md

@@ -0,0 +1,279 @@
+# File Matching


How much of this is duplicated from https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/stanza/operator/input/file/design.md?

@djaglowski all of it. I've renamed the file and placed it in a new directory.

Anyway, I'll separate it out into a new PR.

djaglowski · 2025-01-06T16:52:14Z

pkg/stanza/fileconsumer/design/archive.md

@@ -0,0 +1,173 @@
+# File archiving
+
+The file consumer now supports archiving. Previously, file offsets older than three poll cycles were discarded, and if such files reappeared (which could happen if they were temporarily removed or if `exclude_older_than` was enabled), the entire file contents would be read again.


I don't think we need to describe previous functionality in this document

djaglowski · 2025-01-06T16:53:57Z

pkg/stanza/fileconsumer/design/archive.md

+
+## How does archiving work?
+
+- We stores the offsets older than three poll cycles on disk. If we use `polls_to_archive: 10`, the on-disk structure looks like following:


Do we have 3 in memory and 7 on disk? This seems worth calling out explicitly in the example.

Do we have 3 in memory and 7 on disk? This seems worth calling out explicitly in the example.

No. We have 3 in memory and 10 on disk.
I will explicitly mention this to clear the difference.

djaglowski · 2025-01-06T16:54:43Z

pkg/stanza/fileconsumer/design/archive.md

+### How does reading from archiving work?
+
+During reader creation, we group all the new (or unmatched) files and try to find a match in archive. From high level, it consists of following steps:
+1. We start from most recently writen index on archive and load the data from it.


We should mention in-memory first, then archive is used as a fallback.

VihasMakwana · 2025-01-07T15:00:41Z

@djaglowski new PR for docs #37067

chore: save most recent written index

25105a4

VihasMakwana requested review from djaglowski and a team as code owners December 12, 2024 11:36

github-actions bot assigned mx-psi Dec 12, 2024

github-actions bot added the pkg/stanza label Dec 12, 2024

VihasMakwana commented Dec 12, 2024

View reviewed changes

VihasMakwana added 5 commits December 12, 2024 17:12

chore: readability

b586d84

retunr

7cdba33

test

43c4298

lint

a7d6903

lint and test

c8a4c51

VihasMakwana force-pushed the archive-minor-fix branch from 262c3e3 to eb13fdb Compare December 12, 2024 13:24

gci

ab6bdd1

VihasMakwana force-pushed the archive-minor-fix branch from eb13fdb to ab6bdd1 Compare December 12, 2024 13:34

djaglowski reviewed Dec 13, 2024

View reviewed changes

Update pkg/stanza/fileconsumer/internal/tracker/tracker.go

c8171a4

Co-authored-by: Daniel Jaglowski <[email protected]>

mx-psi removed their assignment Dec 13, 2024

VihasMakwana added 2 commits December 20, 2024 17:19

chore: archive restoration

5d5f12d

chore: add docs

1a66069

VihasMakwana requested a review from djaglowski December 23, 2024 15:06

VihasMakwana and others added 2 commits December 23, 2024 20:46

linting

f6f6815

Merge branch 'main' into archive-minor-fix

2c578f5

djaglowski reviewed Jan 6, 2025

View reviewed changes

doc update

b4adbf5

VihasMakwana mentioned this pull request Jan 7, 2025

DO NOT MERGE [pkg/stanza] - Documentation for archive #37067

Open

VihasMakwana added 2 commits January 7, 2025 20:27

remove doc

b9e55ba

remove image

a843a3b

VihasMakwana requested a review from djaglowski January 7, 2025 15:00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[chore][tracker]: save most recent (archive) write index to disk #36799

[chore][tracker]: save most recent (archive) write index to disk #36799

VihasMakwana commented Dec 12, 2024

VihasMakwana Dec 12, 2024

djaglowski Dec 13, 2024

VihasMakwana Dec 13, 2024 •

edited

Loading

djaglowski Dec 18, 2024

VihasMakwana Dec 23, 2024

VihasMakwana Dec 23, 2024

VihasMakwana commented Dec 23, 2024

djaglowski left a comment

djaglowski Jan 6, 2025

VihasMakwana Jan 7, 2025

djaglowski Jan 6, 2025

djaglowski Jan 6, 2025

VihasMakwana Jan 6, 2025

djaglowski Jan 6, 2025

VihasMakwana Jan 6, 2025

VihasMakwana commented Jan 7, 2025

		@@ -0,0 +1,173 @@
		# File archiving

		The file consumer now supports archiving. Previously, file offsets older than three poll cycles were discarded, and if such files reappeared (which could happen if they were temporarily removed or if `exclude_older_than` was enabled), the entire file contents would be read again.


		## How does archiving work?

		- We stores the offsets older than three poll cycles on disk. If we use `polls_to_archive: 10`, the on-disk structure looks like following:

[chore][tracker]: save most recent (archive) write index to disk #36799

Are you sure you want to change the base?

[chore][tracker]: save most recent (archive) write index to disk #36799

Conversation

VihasMakwana commented Dec 12, 2024

Link to tracking issue

Testing

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VihasMakwana Dec 13, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VihasMakwana commented Dec 23, 2024

djaglowski left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

VihasMakwana commented Jan 7, 2025

VihasMakwana Dec 13, 2024 •

edited

Loading