Persistent storage in queued_retry, backed by file storage extension

[pmm-sumo]

Description:

Persistent queue implementation within queued_retry, aimed at being compatible with Jager's BoundedQueue interface (providing a simple replacement) and backed by file storage extension for storing WAL.

Currently, to run the persistent queue, OpenTelemetry Collector Contrib with enable_unstable build tag is required.

Link to tracking Issue: #2285

Design doc

Testing: Unit Tests and manual testing, more to come

Documentation: README.md updated, including an example

[tigrannajaryan]

Reviewed partially.

[tigrannajaryan]

Thank you for the nice diagram!

[tigrannajaryan]

Can we change this behavior and remove only after the item is confirmed to be delivered? Otherwise we risk loosing data if we crash/restart which defeats the purpose of the persistent queue.

[pmm-sumo]

Yes, I believe we must do it. I didn't want to make this PR too complicated and taking consumers into play adds some level of complexity (actually, I am not sure what's the best implementation, so far, thought about separate set of keys for just the consumers to keep the currently processed items there)

[tigrannajaryan]

Is this the in-memory size when WAL is enabled? Do we need the second setting to limit the size of the persisted data?

[pmm-sumo]

I was thinking about another setting for persistent data (since the number would be probably much different) but then maybe we should just keep the same setting to keep it simple

[tigrannajaryan]

Should this be changed to be more type safe?

Suggested change

	func (wq *WALQueue) Produce(item interface{}) bool {
	func (wq *WALQueue) Produce(item request) bool {

[pmm-sumo]

Actually, I kept it like that to match the BoundedQueue interface

[github-actions]

This PR was marked stale due to lack of activity. It will be closed in 7 days.

[tigrannajaryan]

@pmm-sumo ping me when it is time to review again.

[pmm-sumo]

@pmm-sumo ping me when it is time to review again.

Got a bit delayed but I am getting back to this now and should have an update by the end of the week

[pmm-sumo]

@tigrannajaryan I have updated the codebase. It is now based on just the file storage interface (#3425). In effect, it requires OpenTelemetry Collector Contrib to run currently. Let me know if you would like to arrange it differently (and move most of the code here to Contrib - I wasn't sure how to do that without making it much more complex)

I renamed it to "Persistent Queue" and also addressed most of the comments. The major one missing is persistence for the consumers (i.e. when the request is taken from the queue). I was thinking about having separate items for each of the consumers backed by queue - that would require extra marshaling/unmarshaling but is probably the simplest/cleanest

[tigrannajaryan]

The major one missing is persistence for the consumers (i.e. when the request is taken from the queue).

@pmm-sumo sorry for a delayed review.

I think one typical approach to fix this problem is to have a queue with 3 pointers: write, read and delete. Reading does not delete the item from the queue. So the consuming end looks something like this:

for { 
  item := read(queue):
  err := send(item) // this is a blocking call that only returns when sending is done or fails.
  if !err {
    delete(queue, item)
  } else {
    // retry logic
  }
}

I think it is important to validate that the current storage interface allows to efficiently support such behavior.

[pmm-sumo]

The major one missing is persistence for the consumers (i.e. when the request is taken from the queue).

@pmm-sumo sorry for a delayed review.

I think one typical approach to fix this problem is to have a queue with 3 pointers: write, read and delete. Reading does not delete the item from the queue. So the consuming end looks something like this:
(...)

Thank you @tigrannajaryan

With multiple consumers this gets bit more complicated as the range between delete and read is not contiguous. I am finalising implementation right now that does use callbacks and another key/value which stores addresses of items that are currently being processed. Going to run a couple of tests and update the PR soon

[pmalek-sumo]

Hi @tigrannajaryan,

Can we get another review on this PR?

@pmm-sumo is out for a couple of days now but we'll happily address any outstanding issues if there are any.

[pmalek-sumo]

Hi @tigrannajaryan,

Do we want to resolve the issue/challenge of batching in this PR or do we want to park this until later on? (perhaps create an issue for this)

If that's decided then we can think or resolving conflicts/fixing tests which got introduced in 0.29 if I understand the changelog correctly (and #3455 seems to be the main culprit).

[tigrannajaryan]

Do we want to resolve the issue/challenge of batching in this PR or do we want to park this until later on? (perhaps create an issue for this)

Can we discuss a few possible Storage interfaces that support batched Set or Get or even transactions with multiple operations mixed?

I would not want to merge Storage interface which we know is not good enough. It would be best to have the right interface from the beginning. This PR does not necessarily need to be refactored to get advantage of the improved Storage, it can be a separate PR, but it would be again great to prove that the improved Storage serves our needs.

[pmm-sumo]

Please also add unstable build and test to Github actions (if not already).

I have created few more targets in Makefile and currently make gotest executes both test and test-unstable. As for the GitHub Actions, build-and-test creates also unstable set of binaries (though to be honest they are not very useful, one needs persistent storage implementation, which is available in Contrib)

[tigrannajaryan]

@bogdandrutu PTAL.

[tigrannajaryan]

@open-telemetry/collector-approvers if you reviewed this please approve if you are OK with the PR.

[djaglowski]

LGTM

[tigrannajaryan]

Since we have just made a release now is a good time to merge to have enough time till next release to fix any issues.
I am going merge the PR and if Bogdan or other approvers have comments they can be addressed in a future PR.

[tigrannajaryan]

@pmm-sumo thank you for your patience and the great effort to get this done!