Adding negative cache to throttle extra request https://github.com/opendistro-for-elasticsearch/anomaly-detection/issues/33 #40

zhanghg08 · 2020-01-30T23:51:28Z

Description:
Adding negative cache to throttle extra request #33

Updated on Feb 05:

Re-structured the code for a better dependency injection
Now there is a new class Throttler handle the negative cache stuff.
ClientUtil will inject the dependency of Throttler instead of ADStateManager.

Test:

./gradlew build
Created a single node domain with NAB data. Created one AD detector and trigger the run. Verified the negative cache work as expected.

…stop AD request" This reverts commit 363daa6, reversing changes made to caebb97. This also reverts commit ff190fa

opendistro-for-elasticsearch#33

wnbts · 2020-01-31T18:27:52Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/cluster/HourlyCron.java

@@ -43,6 +43,8 @@ public HourlyCron(ClusterService clusterService, Client client) {
    public void run() {
        DiscoveryNode[] dataNodes = clusterService.state().nodes().getDataNodes().values().toArray(DiscoveryNode.class);

+        // we also add the cancel query function here based on query text from the negative cache.


Question. From #33 , query cannot be aborted. The comment indicates the opposite. Is it confusing?

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)

wnbts · 2020-01-31T18:29:58Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/feature/FeatureManager.java

     * @return unprocessed features and processed features for the current data point
     */
    @Deprecated
-    public SinglePointFeatures getCurrentFeatures(AnomalyDetector detector, long startTime, long endTime) {
+    public SinglePointFeatures getCurrentFeatures(AnomalyDetector detector, long startTime, long endTime, ADStateManager stateManager) {


The added parameter should be a dependency injected rather than passed all the way down the stack.

Agree. We should not pass state manager around.

wnbts · 2020-01-31T18:31:00Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/feature/SearchFeatureDao.java

     * @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) {
+    public Optional<double[]> getFeaturesForPeriod(AnomalyDetector detector, long startTime, long endTime, ADStateManager stateManager) {


The needed dependency can be injected in this class.

wnbts · 2020-01-31T18:42:03Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/ADStateManager.java

+            return Optional.empty();
+        }
+        if (negativeCache.containsKey(detector.getDetectorId())) {
+            return Optional.of(negativeCache.get(detector.getDetectorId()));


Minor. This method can be simplified to this line.

wnbts · 2020-01-31T18:46:09Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/ADStateManager.java

@@ -119,6 +124,47 @@ public int getPartitionNumber(String adID) throws InterruptedException {
        return partitionNum;
    }

+    /**
+     * Get negative cache value(QueryBuilder, Instant) for given detector
+     * If detectorId is null, return Optional.empty()


Minor. This can be assumed to be unlikely. If this is a real concern that must be addressed, the proper behavior to expect is to throw an exception.

wnbts · 2020-01-31T18:48:22Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/ADStateManager.java

+        if (detector.getDetectorId() == null) {
+            return;
+        }
+        negativeCache.putIfAbsent(detector.getDetectorId(), new SimpleEntry<>(searchRequest, clock.instant()));


Question. Is put more expected for the client? Or, the insert call returns but the entry is still not updated. If that's by design, the documentation should make it clear.

Yes, I think put is better than putIfAbsent here. Will update.

wnbts · 2020-01-31T18:49:24Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/ADStateManager.java

+        if (detector.getDetectorId() == null) {
+            return;
+        }
+        negativeCache.keySet().removeIf(key -> key.equals(detector.getDetectorId()));


Minor. Map::remove should work.

wnbts

Suggestion. Throttling can be separated into its own module as a throttler, which can be injected into state manager and client util.

ylwu-amzn · 2020-02-04T00:29:25Z

...in/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java

+            Optional<Entry<SearchRequest, Instant>> queryEntry = stateManager.getFilteredQuery(anomalyDetector);
+            if (queryEntry.isPresent()) {
+                LOG.info("There is one query running for detectorId: {}", anomalyDetector.getDetectorId());
+                listener.onResponse(new AnomalyResultResponse(Double.NaN, Double.NaN, new ArrayList<FeatureData>()));


Client can't distinguish this case with "No data in current detection window"(line 295) as both return empty result. How about we throw an exception like line243

Good catch! Yes, we should throw exception in this case. Will update in next revision.

ylwu-amzn · 2020-02-04T00:36:12Z

...in/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java

@@ -249,6 +251,12 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
                return;
            }
            AnomalyDetector anomalyDetector = detector.get();
+            Optional<Entry<SearchRequest, Instant>> queryEntry = stateManager.getFilteredQuery(anomalyDetector);
+            if (queryEntry.isPresent()) {
+                LOG.info("There is one query running for detectorId: {}", anomalyDetector.getDetectorId());


If a query stuck for a long time, that will impact AD realtime detection. How about we cancel the stuck/running query and run the new coming query?

You are correct. Canceling the stuck query whenever new query comes is one way. However it can be expensive since right now we can only use string compare(query.getDescription()) to find which query to kill. Also killing query using task management API is not effective instantly. It will only happen when moving to next segment.

wnbts

Most stylish issues. Consider race conditions.

wnbts · 2020-02-18T21:28:28Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/ADStateManager.java

+    /**
+     * Check if there is running query on given detector
+     * @param detector Anomaly Detector
+     * @return boolean


Minor. The doc is incomplete.

will update

wnbts · 2020-02-18T21:30:49Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/feature/SearchFeatureDao.java

@@ -114,7 +114,9 @@ public SearchFeatureDao(
    }

    /**
-     * Gets features for the given time period.
+     * Gets features for the given time period. This function also add given detector to negative cache before sending es request.


Minor. Adds.

Also, use descriptive language instead of prescriptive. See java doc.

will update, thanks for the suggestion

wnbts · 2020-02-18T21:33:24Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/ClientUtil.java

+        AnomalyDetector detector
+    ) {
+        try {
+            throttler.insertFilteredQuery(detector, request);


What should happen if the detector is already in cache?

It won't happen here since if a detector is already in cache, we will throw exception in AnomalyResultTransportAction

This can be a race condition when concurrent threads have finished the check at the higher level and try to execute queries. The method needs to be synchronized.

wnbts · 2020-02-18T21:38:06Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/ClientUtil.java

+     * @return boolean
+     */
+    public boolean hasRunningQuery(AnomalyDetector detector) {
+        Optional<Map.Entry<ActionRequest, Instant>> queryEntry = throttler.getFilteredQuery(detector);


Minor. The method can be simplified to one line. return throttler.getFilteredQuery(detector).isPresent()

will update

wnbts · 2020-02-18T21:38:26Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

+import java.util.concurrent.ConcurrentHashMap;
+import org.elasticsearch.action.ActionRequest;
+
+public class Throttler {


Minor. The class java doc is missing.

will update

wnbts · 2020-02-18T21:43:53Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

+        if (negativeCache.containsKey(detector.getDetectorId())) {
+            return Optional.of(negativeCache.get(detector.getDetectorId()));


MInor. The whole method can be simplified to this line return Optional.of(negativeCache.get(detector.getDetectorId()));

will update

Should be Optional.ofNullable

wnbts · 2020-02-18T21:44:49Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

+
+    /**
+     * Insert the negative cache entry for given detector
+     * If detectorId is null, do nothing


Minor. The java doc is outdated.

will update

ylwu-amzn · 2020-02-18T22:41:45Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

-            return Optional.of(negativeCache.get(detector.getDetectorId()));
-        }
-        return Optional.empty();
+        return Optional.of(negativeCache.get(detector.getDetectorId()));


Optional.ofNullable

will update

wnbts

this pr has some conflicts with current branch to resolve

wnbts · 2020-02-20T00:39:19Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/feature/SearchFeatureDao.java

@@ -115,6 +115,9 @@ public SearchFeatureDao(

    /**
     * Gets features for the given time period.
+     * This function also adds given detector to negative cache before sending es request.
+     * Once we get response/exception within timeout, we treat this request as complete and clear the negative cache.


Minor. We is prescriptive (giving orders to code), not descriptive (stating what code does).

wnbts · 2020-02-20T00:50:27Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/ClientUtil.java

+        AnomalyDetector detector
+    ) {
+        try {
+            throttler.insertFilteredQuery(detector, request);


This can be a race condition when concurrent threads have finished the check at the higher level and try to execute queries. The method needs to be synchronized.

wnbts · 2020-02-20T00:56:26Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

+
+    /**
+     * Clear the negative cache for given detector.
+     * If detectorId is null, do nothing


Minor. Outdated doc.

wnbts · 2020-02-20T00:57:52Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/Throttler.java

+     * @param detector AnomalyDetector
+     */
+    public void clearFilteredQuery(AnomalyDetector detector) {
+        negativeCache.keySet().removeIf(key -> key.equals(detector.getDetectorId()));


Minor. Why not using Map::remove?

2) Adding check when inserting new entry to negative cache.

wnbts · 2020-02-24T21:24:23Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/ClientUtil.java

+            // if key already exist, reject the request and throws exception
+            if (!throttler.insertFilteredQuery(detector.getDetectorId(), request)) {
+                LOG.error("There is one query running for detectorId: {}", detector.getDetectorId());
+                throw new EndRunException(detector.getDetectorId(), "There is one query running on AnomalyDetector", true);


MInor. This exception should be documented for client.

ylwu-amzn · 2020-02-24T23:01:50Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/util/ClientUtil.java

+            return Optional.ofNullable(respReference.get());
+        } catch (InterruptedException e1) {
+            LOG.error(CommonErrorMessages.WAIT_ERR_MSG);
+            throw new IllegalStateException(e1);


If exception happens, should clear negative cache for this detector

ylwu-amzn

LGTM. Thanks for the change.

zhanghg08 added 2 commits January 30, 2020 10:42

Revert "merge changes from alpha branch: change setting name and fix …

9aa9e46

…stop AD request" This reverts commit 363daa6, reversing changes made to caebb97. This also reverts commit ff190fa

Adding negative cache to AD

bc6a763

opendistro-for-elasticsearch#33

wnbts reviewed Jan 31, 2020

View reviewed changes

ylwu-amzn reviewed Feb 4, 2020

View reviewed changes

zhanghg08 added 2 commits February 5, 2020 10:33

Address feedback from Yaliang and Lai

3a82b22

Remove unused data structure

fe2a193

zhanghg08 requested review from ylwu-amzn and wnbts February 18, 2020 16:35

wnbts reviewed Feb 18, 2020

View reviewed changes

Adding missing java doc and simplify code.

1bb9257

zhanghg08 requested a review from wnbts February 18, 2020 22:21

ylwu-amzn reviewed Feb 18, 2020

View reviewed changes

Address feedback: Optional.ofNullable

fae4c59

wnbts reviewed Feb 20, 2020

View reviewed changes

zhanghg08 added 4 commits February 20, 2020 09:21

Address feedback

c6a048d

Merge remote-tracking branch 'upstream/development' into development

fcab62c

1) Adding test case for throttler

f7ad7c0

2) Adding check when inserting new entry to negative cache.

Merge remote-tracking branch 'upstream/development' into development

ff73fb7

zhanghg08 requested review from ylwu-amzn and wnbts February 21, 2020 17:01

wnbts approved these changes Feb 24, 2020

View reviewed changes

ylwu-amzn reviewed Feb 24, 2020

View reviewed changes

zhanghg08 added 2 commits February 24, 2020 15:20

Add additional clean cache step.

95f3892

Add javadoc for EndRunException

1d086d1

ylwu-amzn approved these changes Feb 24, 2020

View reviewed changes

zhanghg08 merged commit b8ea1a2 into opendistro-for-elasticsearch:development Feb 24, 2020

		if (negativeCache.containsKey(detector.getDetectorId())) {
		return Optional.of(negativeCache.get(detector.getDetectorId()));

Adding negative cache to throttle extra request https://github.com/opendistro-for-elasticsearch/anomaly-detection/issues/33 #40

Adding negative cache to throttle extra request https://github.com/opendistro-for-elasticsearch/anomaly-detection/issues/33 #40

Conversation

zhanghg08 commented Jan 30, 2020 • 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

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

wnbts 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

wnbts 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wnbts 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

ylwu-amzn left a comment

Choose a reason for hiding this comment

zhanghg08 commented Jan 30, 2020 •

edited

Loading