This repository has been archived by the owner on Aug 2, 2022. It is now read-only.
Adding negative cache to throttle extra request https://github.com/opendistro-for-elasticsearch/anomaly-detection/issues/33 #40
There are no files selected for viewing
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
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
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -30,6 +30,7 @@ | |
|
|
||
| import com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetector; | ||
| import com.amazon.opendistroforelasticsearch.ad.model.IntervalTimeConfiguration; | ||
| import com.amazon.opendistroforelasticsearch.ad.transport.ADStateManager; | ||
| import com.amazon.opendistroforelasticsearch.ad.util.ClientUtil; | ||
| import com.amazon.opendistroforelasticsearch.ad.util.ParseUtils; | ||
| import org.apache.logging.log4j.LogManager; | ||
|
|
@@ -114,18 +115,24 @@ public Optional<Long> getLatestDataTime(AnomalyDetector detector) { | |
| } | ||
|
|
||
| /** | ||
| * Gets features for the given time period. This function also add given detector to negative cache before sending es request. | ||
|
There was a problem hiding this comment. Minor. Adds. Also, use descriptive language instead of prescriptive. See java doc. There was a problem hiding this comment. will update, thanks for the suggestion |
||
| * Once we get response/exception within timeout, we treat this request as complete and clear the negative cache. | ||
|
There was a problem hiding this comment. Minor. We is prescriptive (giving orders to code), not descriptive (stating what code does). |
||
| * Otherwise this detector entry remain in the negative to reject further request. | ||
| * | ||
| * @param detector info about indices, documents, feature query | ||
| * @param startTime epoch milliseconds at the beginning of the period | ||
| * @param endTime epoch milliseconds at the end of the period | ||
| * @param stateManager ADStateManager | ||
| * @throws IllegalStateException when unexpected failures happen | ||
| * @return features from search results, empty when no data found | ||
| */ | ||
| public Optional<double[]> getFeaturesForPeriod(AnomalyDetector detector, long startTime, long endTime, ADStateManager stateManager) { | ||
|
There was a problem hiding this comment. The needed dependency can be injected in this class. |
||
| SearchRequest searchRequest = createFeatureSearchRequest(detector, startTime, endTime, Optional.empty()); | ||
| // add (detectorId, filteredQuery) to negative cache | ||
| stateManager.insertFilteredQuery(detector, searchRequest); | ||
| // send throttled request: this request will clear the negative cache if the request finished within timeout | ||
| return clientUtil | ||
| .<SearchRequest, SearchResponse>throttledTimedRequest(searchRequest, logger, client::search, stateManager, detector) | ||
| .flatMap(resp -> parseResponse(resp, detector.getEnabledFeatureIds())); | ||
| } | ||
|
|
||
|
|
@@ -242,20 +249,22 @@ public void getFeatureSamplesForPeriods( | |
| * @param maxSamples the maximum number of samples to return | ||
| * @param maxStride the maximum number of periods between samples | ||
| * @param endTime the end time of the latest period | ||
| * @param stateManager ADStateManager | ||
| * @return sampled features and stride, empty when no data found | ||
| */ | ||
| public Optional<Entry<double[][], Integer>> getFeaturesForSampledPeriods( | ||
| AnomalyDetector detector, | ||
| int maxSamples, | ||
| int maxStride, | ||
| long endTime, | ||
| ADStateManager stateManager | ||
| ) { | ||
| Map<Long, double[]> cache = new HashMap<>(); | ||
| int currentStride = maxStride; | ||
| Optional<double[][]> features = Optional.empty(); | ||
| while (currentStride >= 1) { | ||
| boolean isInterpolatable = currentStride < maxStride; | ||
| features = getFeaturesForSampledPeriods(detector, maxSamples, currentStride, endTime, cache, isInterpolatable, stateManager); | ||
| if (!features.isPresent() || features.get().length > maxSamples / 2 || currentStride == 1) { | ||
| break; | ||
| } else { | ||
|
|
@@ -275,7 +284,8 @@ private Optional<double[][]> getFeaturesForSampledPeriods( | |
| int stride, | ||
| long endTime, | ||
| Map<Long, double[]> cache, | ||
| boolean isInterpolatable, | ||
| ADStateManager stateManager | ||
| ) { | ||
| ArrayDeque<double[]> sampledFeatures = new ArrayDeque<>(maxSamples); | ||
| for (int i = 0; i < maxSamples; i++) { | ||
|
|
@@ -284,7 +294,7 @@ private Optional<double[][]> getFeaturesForSampledPeriods( | |
| if (cache.containsKey(end)) { | ||
| sampledFeatures.addFirst(cache.get(end)); | ||
| } else { | ||
| Optional<double[]> features = getFeaturesForPeriod(detector, end - span, end, stateManager); | ||
| if (features.isPresent()) { | ||
| cache.put(end, features.get()); | ||
| sampledFeatures.addFirst(features.get()); | ||
|
|
||
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
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
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
Oops, something went wrong.
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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Question. From #33 , query cannot be aborted. The comment indicates the opposite. Is it confusing?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Good catch. That's true. Currently anomaly-detection cannot abort running query in elasticsearch. If a query running longer than expected, anomaly-detection will not wait for that even though the query is still running. To solve this issue, we will 1) stop accepting new query if this case happen which is #33 2) daily cron clean up running query using es task management API(https://www.elastic.co/guide/en/elasticsearch/reference/current/tasks.html)