[ML] Persist data counts and datafeed timing stats asynchronously #93000
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
When an anomaly detection job runs, the majority of results originate from the C++ autodetect process, so can be persisted in bulk. However, there are two types of results, namely data counts and datafeed timing stats, that are generated wholly within the ML Java code and where there are serious downsides to batching them up with the output of the C++ process. (If we batched them and the C++ process stopped generating results then the input side stats would also stall, so it is better that the input side stats are written independently.) The approach used in this PR is to write data counts and datafeed timing stats asynchronously _except_ at certain key points, like job flush and close, and datafeed stop. At these key points the latest stats _are_ persisted synchronously, like before. When large amounts of data are being processed the code will generate updated stats documents faster than they can be indexed. The approach taken here is to skip persistence of the newer document if persistence of the previous document is still in progress. This can lead to the stats being slightly out of date while a job is running. However, at key points like flush and close the data counts will be up-to-date, and the datafeed timing stats will get written at least once per datafeed `frequency`, so should not be more out-of-date than that.
|
Pinging @elastic/ml-core (Team:ML) |
|
Hi @droberts195, I've created a changelog YAML for you. |
davidkyle
approved these changes
Jan 18, 2023
| logger.trace("[{}] not persisting datafeed timing stats as persistence is disallowed", jobId); | ||
| return; | ||
| } | ||
| if (persistInProgressLatch != null && persistInProgressLatch.await(1, TimeUnit.NANOSECONDS) == false) { |
There was a problem hiding this comment.
persistInProgressLatch.await(1, TimeUnit.NANOSECONDS)
This looks like you want to test if the latch is waitable, i.e. the count is greater than zero. In which case it could be replaced with persistInProgressLatch.getCount() > 0
droberts195
added a commit
to droberts195/elasticsearch
that referenced
this pull request
Jan 27, 2023
When a datafeed starts up it looks at the job's data counts to decide where to pick up from where a previous invocation finished. Previously the datafeed was always getting the data counts from the index. In the case where a datafeed has been stopped and restarted while its corresponding job was continually opened this is incorrect. In this case the data counts need to be obtained from the running job. (In the case where the job was closed and reopened while the datafeed was stopped this does not matter, as closing the job will have persisted the up-to-date data counts.) This bug has always existed, yet is made much more likely to cause a noticeable discrepancy by the changes made in elastic#93000. Fixes elastic#93298
elasticsearchmachine
pushed a commit
that referenced
this pull request
Jan 27, 2023
When a datafeed starts up it looks at the job's data counts to decide where to pick up from where a previous invocation finished. Previously the datafeed was always getting the data counts from the index. In the case where a datafeed has been stopped and restarted while its corresponding job was continually opened this is incorrect. In this case the data counts need to be obtained from the running job. (In the case where the job was closed and reopened while the datafeed was stopped this does not matter, as closing the job will have persisted the up-to-date data counts.) This bug has always existed, yet is made much more likely to cause a noticeable discrepancy by the changes made in #93000. Fixes #93298
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
When an anomaly detection job runs, the majority of results originate from the C++ autodetect process, so can be persisted in bulk. However, there are two types of results, namely data counts and datafeed timing stats, that are generated wholly within the ML Java code and where there are serious downsides to batching them up with the output of the C++ process. (If we batched them and the C++ process stopped generating results then the input side stats would also stall, so it is better that the input side stats are written independently.)
The approach used in this PR is to write data counts and datafeed timing stats asynchronously except at certain key points, like job flush and close, and datafeed stop. At these key points the latest stats are persisted synchronously, like before. When large amounts of data are being processed the code will generate updated stats documents faster than they can be indexed. The approach taken here is to skip persistence of the newer document if persistence of the previous document is still in progress. This can lead to the stats being slightly out of date while a job is running. However, at key points like flush and close the data counts will be up-to-date, and the datafeed timing stats will get written at least once per datafeed
frequency, so should not be more out-of-date than that.