[pkg/stanza] - Finalize the design discussion

[VihasMakwana]

@djaglowski I'm starting this discussion because our previous github issue has lots of people watching and it would spam their mailboxes ;(

So, what I have in mind right now:

• Get the thread pool PR merged, that would remove some complexity in my upcoming PRs
• Raise PRs for adding the core functionality of feature gate, which would include:
  • PR for handling lost readers
  • PR for handling fingerprint matches
  • PR for handling saving/reloading offsets
  • PR for test cases. (benchmarks included)
  • A final PR for updating user docs and add feature gate option to registry. (I'm keeping this as last, PRs merged before this would have no effect on current functionality whatsoever)

[djaglowski]

@VihasMakwana, we have gone through many different designs, most of them defined as a difference from a previous attempt. As I mentioned here, I don't understand your most recent design. I think I have gotten lost in all the changes. Perhaps you could put together a design document which describes your design and articulates how it manages to support the current functionality.

[VihasMakwana]

Okay. Makes sense

[VihasMakwana]

Before I attach the design, I would like us to start from scratch and treat it as a new feature.

[VihasMakwana]

@djaglowski talking from the design perspectiv, here's high level overview of my design:

A pictorial representation of the beginning of every poll()

2nd stage of the poll() i.e. start creating readers in batches and submit them to our pool:

Our thread pool's function:

[VihasMakwana]

The steps per poll() in the foreground include:

1. Read all the previously opened files, we read all open files.
2. Get all the matches and start creating readers
3. Add the fingerprint to trie
4. Queue our newly created readers to the thread pool
5. Sync the knownFiles array
6. Repeat steps 1st to 5th until completion/exit

The steps in the thread pool include:

1. Wait on channel to receive new readers
2. Read to end
3. Add reader to knownFiles
4. Remove fingerprint from trie
5. Repeat steps 1st to 4th until channel is closed.

All of this is done while considering potential race conditions and proper locking is used.

[djaglowski]

@VihasMakwana, will you please include a detailed explanation for each of the following:

1. How will your design keep track of which files are open?
2. When we open a file, how do we establish that we should read it vs close it immediately?
3. If we decide to read file which we've just opened, how do we determine the offset?
4. Under what conditions is a file closed?
5. How does this design ensure that we respect max_concurrent_files?

[VihasMakwana]

@djaglowski

1. How will your design keep track of which files are open?

   • We use knownFiles to keep track of known open files from previous poll() cycles.
   • Also, an observation. If a reader is closed, its file member will be nil. This check is used to determine an open/closed reader.
   • Ref: https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/stanza/fileconsumer/internal/reader/reader.go#L151

2. When we open a file, how do we establish that we should read it vs. close it immediately?

   • Based on the deleteAfterRead flag.
   • We don't keep track of deleted files, just like the current implementation.

3. If we decide to read a file that we've just opened, how do we determine the offset?

   • Depends on multiple scenarios, but all of them are the same as the current implementation. Nothing new here:
     1. If the file is an old one, then we decide from its previous offset from knownFiles (same as the current implementation).
     2. If it's a newly discovered file and doesn't match any of the files from knownFiles, then the offset is selected based on the startAt flag. startAt: beginning means the offset is 0, startAt: end means the offset is the end of the file.

4. Under what conditions is a file closed?

   • In the first stage of calling the consume() on a given batch of files, we loop through m.knownFiles and acquire the mutex.
   • We read all the files and close them after reading to end and release the mutex.
   • Mutex ensures that the underlying array is not changed while looping.
   • We also wait for all files to be read till the end and closed before proceeding. This would ensure that the offsets are up to date and there's no reading from the outdated offset twice.

   Note:

   • We hold files open as long as possible and close them one last time before proceeding with new readers
   • Since we have a completely asynchronous solution, we don't have a clear understanding of whether an open file is from a previous batch or two batches before.
   • Hence, we can't perform the rollReaders() check like the current implementation, as it requires all the readers from the current batch and all the readers from the previous batch.
   • In a complete asynchronous solution, there's no such guarantee.
   • The best way is to read all the open files. One of two scenarios can occur:
     • Either the open file is a lost reader. In this case, it's best to read till the end and close the file.
     • Either the open file is not a lost reader. In this case, if we read and close the file, it wouldn't have any effect. The new batch will copy the offset and open the file once again before submitting it to threadpool. We will indeed have an up-to-date offset.

5. How does this design ensure that we respect max_concurrent_files?

   • Assumptions and initial observation:
     • Let's consider the worst-case scenario i.e. we discover new files, always.
     • Our workers are limited by maxConcurrentFiles/2, which implies that there can be at most maxConcurrentFiles/2 active readers.
   • We run clearOldReaders before the start of a new batch. Which is the 4th answer.
   • The maximum number of open files in the thread pool is limited by maxConcurrentFiles/2 i.e. the batch size.
   • At the start of a new batch, we can have a maximum maxConcurrentFiles/2 number of open files in knownFiles from the previous batch.
   • We then call clearOldReaders and clean open files which will release file descriptors. This way we respect max_concurrent_files

[VihasMakwana]

@djaglowski did you get a chance to review the design?

[djaglowski]

@djaglowski

1. How will your design keep track of which files are open?

   • We use knownFiles to keep track of known open files from previous poll() cycles.
   • Also, an observation. If a reader is closed, its file member will be nil. This check is used to determine an open/closed reader.

knownFiles are not necessarily open. I would like to move to a model where we have a concrete list of open files, but we do not have it today.

If your proposal to Read all the previously opened files, we read all open files. depends on iterating over knownFiles, then this implies we will duplicate reads from multiple handles to the same file.

2. When we open a file, how do we establish that we should read it vs. close it immediately?

   • Based on the deleteAfterRead flag.
   • We don't keep track of deleted files, just like the current implementation.

This makes no sense to me. What does the deleteAfterRead flag have to do with opening files?

I'm asking how your design would asynchronously handle what we currently do with checkDuplicates.

3. If we decide to read a file that we've just opened, how do we determine the offset?

   • Depends on multiple scenarios, but all of them are the same as the current implementation. Nothing new here:
     1. If the file is an old one, then we decide from its previous offset from knownFiles (same as the current implementation).

I need to understand how, in an asynchronous solution, we can safely and correctly determine the most recent offset. Based on later answers, it looks like you're proposing to mutex the entire knownFiles slice, which would allow you to iterate through it safely. However, if you do this I don't understand how you intend to open, read, and close, files asynchronously while keeping knownFiles up to date.

4. Under what conditions is a file closed?

   • In the first stage of calling the consume() on a given batch of files, we loop through m.knownFiles and acquire the mutex.
   • We read all the files and close them after reading to end and release the mutex.
   • Mutex ensures that the underlying array is not changed while looping.
   • We also wait for all files to be read till the end and closed before proceeding. This would ensure that the offsets are up to date and there's no reading from the outdated offset twice.

What is the point of your proposal if we are not able to get away from this restriction?

Note:

• We hold files open as long as possible and close them one last time before proceeding with new readers

• Since we have a completely asynchronous solution, we don't have a clear understanding of whether an open file is from a previous batch or two batches before.

• Hence, we can't perform the rollReaders() check like the current implementation, as it requires all the readers from the current batch and all the readers from the previous batch.

• In a complete asynchronous solution, there's no such guarantee.

• The best way is to read all the open files. One of two scenarios can occur:

  • Either the open file is a lost reader. In this case, it's best to read till the end and close the file.
  • Either the open file is not a lost reader. In this case, if we read and close the file, it wouldn't have any effect. The new batch will copy the offset and open the file once again before submitting it to threadpool. We will indeed have an up-to-date offset.

All of this misses the following points which I've reiterated multiple times:

1. We cannot just read from every file handle.
2. We need to establish that a given file handle is the most recent handle to the file.
3. We need to establish that a file handle does not lead to a truncated file.

5. How does this design ensure that we respect max_concurrent_files?

   • Assumptions and initial observation:

     • Let's consider the worst-case scenario i.e. we discover new files, always.
     • Our workers are limited by maxConcurrentFiles/2, which implies that there can be at most maxConcurrentFiles/2 active readers.

   • We run clearOldReaders before the start of a new batch. Which is the 4th answer.

   • The maximum number of open files in the thread pool is limited by maxConcurrentFiles/2 i.e. the batch size.

   • At the start of a new batch, we can have a maximum maxConcurrentFiles/2 number of open files in knownFiles from the previous batch.

   • We then call clearOldReaders and clean open files which will release file descriptors. This way we respect max_concurrent_files

In an asynchronous solution, why do we need to rely on batching? Can't we just close one file immediately before opening another?

[VihasMakwana]

In an asynchronous solution, why do we need to rely on batching? Can't we just close one file immediately before opening another?

If we close them immediately and proceed further, then we won't be able to track moved-away files.
It would work fine for files in pattern, but for files that moved away, it wouldn't work as expected.
It causes TestTrackMovedAwayFiles and TestRotatedOutOfPatternMoveCreate to fail in rotation.go

[VihasMakwana]

I think for a complete asynchronous solution, it might be acceptable. Not sure.
We can always decrease the pollInterval if the file rotation is too quick.

[VihasMakwana]

When we open a file, how do we establish that we should read it vs close it immediately?

I misunderstood this one, my bad.
The reader creation is done synchronously in my proposal, via makeReader. We keep track of fingerprints discovered in the current poll() and perform Equal check on fingerprints.
We will close the file immediately if we find a duplicate.

We also close the file immediately, if we detect a prefix match in trie.

[VihasMakwana]

If your proposal to Read all the previously opened files, we read all open files. depends on iterating over knownFiles, then this implies we will duplicate reads from multiple handles to the same file.

We close the file after reading, it would result in

opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/reader/reader.go

Line 152 in fea5ba8

r.file = nil

.
Next time, if we arrive to same reader, we would have known if we already closed it or not based on r.file

[VihasMakwana]

I need to understand how, in an asynchronous solution, we can safely and correctly determine the most recent offset. Based on later answers, it looks like you're proposing to mutex the entire knownFiles slice, which would allow you to iterate through it safely.

That's correct. We use mutex to guarantee thread safety.

[VihasMakwana]

Under what conditions is a file closed?

In the first stage of calling the consume() on a given batch of files, we loop through m.knownFiles and acquire the mutex.
We read all the files and close them after reading to end and release the mutex.
Mutex ensures that the underlying array is not changed while looping.
We also wait for all files to be read till the end and closed before proceeding. This would ensure that the offsets are up to date and there's no reading from the outdated offset twice.

What is the point of your proposal if we are not able to get away from this restriction?

@djaglowski a comment from my other PR explains why I keep this restriction, pasting the same here:

We wait for all aggregated readers to finish reading before moving forward.

Observations:

- If the number of files matched is significantly higher than maxConcurrentFiles then we are likely going to batch the files more than a couple of times.
- This causes workers() to overwhelm and they'll all be busy.
- If we try to send already finished readers to the workers() anyway, it will cause a bottleneck, and benchmarks show a significant delay in receiving logs.
- It's better to create separate goroutines to process finished readers and let the workers() loosen themselves up for the upcoming poll() cycle.
- In this way, we respect max_concurrent_files, workers() are happy, all's good. The benchmark performs as expected.

---

@djaglowski we CAN get rid of this restriction, but I choose not to based on poor performance of benchmarks in extreme cases.

However, there's an alternative. We can create another thread pool for lost readers.
This will also be limited by maxBatchFiles. Performance-wise, we still see a 15-20% improvement in log throughput.

[VihasMakwana]

I ran the following function to calculate open file descriptors of the current process with various maxConcurrentFiles. It came to observation the maxConcurrentFiles is always respected in my PR (this also includes scenarios where the number of files matched is significantly higher than maxConcurrentFiles)

func getOpenFds() string {
	pid := os.Getpid()
	grep := exec.Command("egrep", "REG|DIR")
	lsof := exec.Command("lsof", "-a", "-d", "0-2147483647", "-p", fmt.Sprintf("%d", pid))
	pipe, _ := lsof.StdoutPipe()
	defer pipe.Close()
	grep.Stdin = pipe
	lsof.Start()

	out, _ := grep.Output()
	return string(out)
}

[VihasMakwana]

• We cannot just read from every file handle.
  I understand, this is why we place a check before reading the handle to avoid dereferencing the null pointer here

• We need to establish that a file handle does not lead to a truncated file.
  This will be addressed by a check which you introduced via [pkg/stanza] Skip rotated file handles if truncated #27064

[VihasMakwana]

• We need to establish that a given file handle is the most recent handle for the file.

To understand this, let's consider the following:

To discuss the above-mentioned scenario, we remove a reader from m.knownFiles if we detect a match here.
Also, while copying a reader, we don't copy the generation.
So, every time a reader is copied, we remove the old reader from m.knownFiles and create a fresh reader with generation set to 0.

The scenario would look something like this with knownFiles:

Time	Actual Content	Offset After Reading	knownFiles (before poll())	knownFiles (after poll())
t0	foo0\n	5	[]	[reader{offset: 5, generation: 0}]
t1	foo0\nfoo1\n	10	[reader{offset: 5, generation: 0}]	[reader{offset: 10 generation: 0}]
t2	foo0\nfoo1\nfoo2\n	15	[reader{offset: 15, generation: 0}]	[reader{offset: 15, generation: 0}]
t3	foo0\nfoo1\nfoo2\nfoo3\n	20	[reader{offset: 20, generation: 0}]	[reader{offset: 20 generation: 0}]

---

This means that if a reader is present in knownFiles, it's the most up-to-date copy of that file.
And if a reader is ready to be queued or it is in worker(), it means it has been removed from knownFiles(here) and added to trie.
Subsequently, poll() will detect that we're already consuming the file. Hence, they will close the file.

Inside the worker, we add the reader with updated offset to knownFiles => the up-to-date file handle.

At a particular moment, there can be NO two readers pointing to the same file in knownFiles. This behavior was introduced in #20749

[VihasMakwana]

@djaglowski I saw that your PR got merged and it looks like we have a straightforward situation with open/closed files.
To reiterate the questions, based on new changes:

• How will your design keep track of which files are open?

  • We now have m.previousPollFiles which keep track of open files from previous poll()

• When we open a file, how do we establish that we should read it vs close it immediately?

  • File will be closed if any of the following is met:
    • If we have detected a fingerprint match from the current poll() itself
    • If we find out that it's an empty file
    • If the file is currently being consumed in the goroutine. This is done by checking the trie with proper locking.
  • In other cases, the fingerprint will be added to the trie, and the file is sent to the channel respectively.

• If we decide to read a file which we've just opened, how do we determine the offset?

  • Based on m.previousPollFiles or m.knownFiles.
  • If we don't find a match in either of them, it's treated as a new file.
  • If we find a match in any of them (first preference to m.previousPollFiles), we copy the metadata and create the reader

• Under what conditions is a file closed?

  • At the beginning of every poll(), we go through m.previousPollFiles, and read them to the end.
  • [pkg/stanza] - Finalize the design discussion #27359 (comment) explains this scenario.

• How does this design ensure that we respect max_concurrent_files?

  • Our thread pool is limited by maxConcurrentFiles/2, which implies that there can be at most maxConcurrentFiles/2 open files in the thread pool (worst case scenario).
  • We also close previously opened files before moving forward, which would be maxConcurrentFiles/2 in size at maximum (from the previous batch)

• How is a given file handle the most recent handle for the file?

  • If a file is present in m.knownFiles or m.previousPollFiles, then it would imply that it is not currently being consumed.
  • Conversely, if a file is currently being consumed, it wouldn't be present in either m.knownFiles or m.previousPollFiles (and we will close the file immediately).
    • This would necessarily mean that it is indeed the most recent offset of that particular file.
  • The file is only added m.previousPollFiles after it has been read to end.
  • The file is also removed from m.previousPollFiles or m.knownFiles if we find a find a match.
  • [pkg/stanza] - Finalize the design discussion #27359 (comment) explains this.

• We need to establish that a file handle does not lead to a truncated file.

  • This logic has been moved to Validate() in reader.go AFAIK and it would be sufficient to establish the validity of the file.

[VihasMakwana]

@djaglowski ping ^

[djaglowski]

@VihasMakwana, I think we should table this discussion until we complete other changes to the fileconsumer package. We have too many things in flight and I do not think it should be a priority to iterate on your proposal until the underlying code has been stabilized. Specifically, I'd like to see us resolve changes mentioned here.

[VihasMakwana]

@djaglowski it turns out we can't use Trie for fingerprints as it adds too much complexity. I'll work on adding a simplified management struct for this.
After that, we can come to this proposal again.
I hope that sounds good.