Change to use callbacks in cold start #208

kaituo · 2020-08-11T01:28:11Z

Issue #, if available:
#78

Description of changes:

This PR changes the code path of cold start in the transport layer to use callbacks.

Previously, I created AD’s ExecutorService that has one thread for cold starts in ColdStartRunner. When triggering a cold start, we can submit a task in the ExecutorService, and consult a hash map (keyed by detector Id) that cached the results of recent cold start results. Since I have to invoke the cold start thread in various callbacks, ColdStartRunner does not work anymore. I created a cold start thread pool and put the cold start result in the transport state instead.

This PR also handles new exceptions like invalid queries introduced by recent changes on ModelManager and FeatureManager.

This PR lowers the severity of a couple of log messages in HashRing and RCFPollingTransportAction to avoid overwhelming log file readers. These log messages are common.

This PR corrects typos and updates known causes of EndRunException in comments.

Testing done:

Simulated cold start failures: Exceptions of cold starts can be seen by the transport layer. EndRunException can cause AD jobs to be terminated.
Happy case of a cold start still works.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

codecov · 2020-08-11T01:29:28Z

Codecov Report

Merging #208 into master will decrease coverage by 0.01%.
The diff coverage is 72.67%.

@@             Coverage Diff              @@
##             master     #208      +/-   ##
============================================
- Coverage     72.26%   72.25%   -0.02%     
- Complexity     1283     1290       +7     
============================================
  Files           140      139       -1     
  Lines          5931     5979      +48     
  Branches        466      469       +3     
============================================
+ Hits           4286     4320      +34     
- Misses         1440     1454      +14     
  Partials        205      205

Flag	Coverage Δ	Complexity Δ
#cli	`79.74% <ø> (ø)`	`0.00 <ø> (ø)`
#plugin	`71.38% <72.67%> (-0.01%)`	`1290.00 <41.00> (+7.00)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ	Complexity Δ
...oforelasticsearch/ad/AnomalyDetectorJobRunner.java	`76.47% <0.00%> (ø)`	`35.00 <0.00> (ø)`
...stroforelasticsearch/ad/AnomalyDetectorPlugin.java	`93.54% <ø> (-0.07%)`	`10.00 <0.00> (ø)`
...opendistroforelasticsearch/ad/ml/ModelManager.java	`92.05% <ø> (ø)`	`112.00 <0.00> (ø)`
...sticsearch/ad/transport/TransportStateManager.java	`73.68% <61.76%> (-2.73%)`	`29.00 <13.00> (+3.00)`	⬇️
...rch/ad/transport/AnomalyResultTransportAction.java	`78.46% <67.74%> (-0.23%)`	`59.00 <12.00> (+14.00)`	⬇️
...oforelasticsearch/ad/transport/TransportState.java	`88.88% <90.90%> (+11.46%)`	`13.00 <11.00> (-3.00)`	⬆️
...pendistroforelasticsearch/ad/cluster/HashRing.java	`78.94% <100.00%> (ø)`	`10.00 <0.00> (ø)`
...troforelasticsearch/ad/feature/FeatureManager.java	`96.68% <100.00%> (+0.05%)`	`96.00 <3.00> (ø)`
...search/ad/transport/RCFPollingTransportAction.java	`90.00% <100.00%> (ø)`	`6.00 <0.00> (ø)`
...ch/ad/transport/handler/DetectionStateHandler.java	`87.23% <100.00%> (ø)`	`8.00 <2.00> (ø)`
... and 4 more

VijayanB · 2020-08-11T17:54:34Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorPlugin.java

@@ -142,6 +141,7 @@
    public static final String AD_BASE_URI = "/_opendistro/_anomaly_detection";
    public static final String AD_BASE_DETECTORS_URI = AD_BASE_URI + "/detectors";
    public static final String AD_THREAD_POOL_NAME = "ad-threadpool";
+    public static final String COLD_START_THREAD_POOL_NAME = "ad-cold-start-threadpool";


minor: Shall we add this as last constant? ( constants arranged alphabetically )

VijayanB · 2020-08-11T17:59:42Z

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

@@ -88,8 +99,11 @@ public boolean expired(Duration stateTtl, Instant now) {
        if (checkpoint != null) {
            ans = ans && expired(stateTtl, now, checkpoint);
        }
-        if (lastError != null) {
-            ans = ans && expired(stateTtl, now, lastError.getValue());
+        if (lastDetectionError != null) {


Question: Is it possible to have both (lastDetectionError and lastColdStartException) to be not null ?

yes, it is possible.

VijayanB · 2020-08-11T18:07:13Z

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

+     * @param adID detector id
+     * @return last cold start exception for the detector
+     */
+    public Exception fetchColdStartException(String adID) {


Question: Where is this used?

It is used in AnomalyResultTransportAction.

Got it. For some reason AnomalyResultTransportAction was not loaded .

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

wnbts

code coverage seems to decrease

wnbts · 2020-08-12T22:09:02Z

src/main/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorPlugin.java

                new FixedExecutorBuilder(
                    settings,
                    AD_THREAD_POOL_NAME,
                    Math.max(1, EsExecutors.allocatedProcessors(settings) / 4),
                    AnomalyDetectorSettings.AD_THEAD_POOL_QUEUE_SIZE,
                    "opendistro.ad." + AD_THREAD_POOL_NAME
+                ),
+                new FixedExecutorBuilder(


why keep the overhead of an additional thread pool constantly when model training is at best occasional?

wnbts · 2020-08-12T22:19:04Z

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

        searchFeatureDao
            .getLatestDataTime(
                detector,
-                ActionListener.wrap(latest -> getColdStartSamples(latest, detector, listener), listener::onFailure)
+                new ThreadedActionListener<>(


why should this query call back be run separately?

Ideally, all AD code should run inside AD threadpool. This is long overdue. I am starting with cold start code.

does it mean every callback needs to run this way? can this be simplified at a higher level?

yes. I don't know how to simplify this. Any suggestion?

TransportNodesAction seems to take thread pool name.

You meant the threadpool name in TransportResponseHandler? Our case is not just transport thread. Also, the threadpool name is good for one callback. It does not cover nested call backs.

wnbts · 2020-08-12T22:40:49Z

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

+    /**
+     * Set last cold start error of a detector
+     * @param adID detector id
+     * @param exception exception, can be null


what does a null exception mean? is that an exception or no exception? why not using optional?

null means there is no error, so no exception. I know you are passionate about optional. Here it is unnecessary.

if it's no error, would it suffice to leave the entire record to null? <exception, time> is null seems simpler.

optional question. why optional is unnecessary ? clearly this variable can be present or absent and the need seems to exist.

changed to optional

wnbts · 2020-08-12T22:46:19Z

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

+    public Exception fetchColdStartException(String adID) {
+        if (transportStates.containsKey(adID) && transportStates.get(adID).getLastColdStartException() != null) {
+            Entry<Exception, Instant> errorAndTime = transportStates.get(adID).getLastColdStartException();
+            errorAndTime.setValue(clock.instant());


why getting the exception status involves updating the timestamp?

It is the same as other transport state. We are gonna clear these states if they are not accessed within one hour.

Would one last access time for one entire statue be sufficient? instead of keeping one timestamp for each field and still using the latest one

wnbts · 2020-08-12T23:03:30Z

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

+                        if (exception instanceof AnomalyDetectionException) {
+                            // partitioned model exceeds memory limit


the two cases are different

LimitExceededException is a subtype of AnomalyDetectionException. I am commenting on one example here. Updated comment to make it more clear.

VijayanB

LGTM

This PR changes the code path of cold start in the transport layer to use callbacks. Previously, I created AD’s ExecutorService that has one thread for cold starts in ColdStartRunner. When we need to trigger a cold start, we can submit a task in the ExecutorService, and consult a hash map (keyed by detector Id) that cached the results of recent cold start results. Since I have to invoke the cold start thread in various callbacks, I created a cold start thread pool and put the cold start result in the transport state. This PR also handles new exceptions like invalid queries introduced by recent changes on ModelManager and FeatureManager. This PR lowers the severity of a couple of log messages in HashRing and RCFPollingTransportAction to avoid overwhelming readers of log files. These log messages are common. This PR corrects typos and updates known causes of EndRunException in comments. Testing done: 1. Simulated cold start failures: Exceptions of cold starts can be seen by the transport layer. EndRunException can cause AD jobs to be terminated. 2. Happy case of a cold start still works.

kaituo · 2020-08-14T20:30:09Z

code coverage seems to decrease

Yes. Maybe because I deleted a class.

ylwu-amzn · 2020-08-15T00:40:39Z

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

-            } catch (Exception ex) {
-                throw new EndRunException(detector.getDetectorId(), "Error while cold start", ex, false);
+        if (previousException != null) {
+            LOG.error("Previous exception of {}: {}", () -> detectorId, () -> previousException);


Why use () -> detectorId? detectorId should work.

See https://www.baeldung.com/log4j-2-lazy-logging

From the document you shared ,

The lambda expression is only evaluated if the corresponding log level is enabled. This is referred to as lazy logging.

I think error log level is always enabled, so the lambda () -> detectorId will always be executed?

you are right. Changed.

ylwu-amzn · 2020-08-15T00:48:15Z

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

-                throw new EndRunException(detector.getDetectorId(), "Error while cold start", ex, false);
+        if (previousException != null) {
+            LOG.error("Previous exception of {}: {}", () -> detectorId, () -> previousException);
+            if (previousException instanceof EndRunException && ((EndRunException) previousException).isEndNow()) {


If it's EndRunException and isEndNow() is true, JobRunner will stop detector job. Will the previousException be cleaned when stop detector job? Why need to check previous exception ? If user wait for some time and restart detector, cluster state may change, and user may not get previous exception any more.

Yes, it will be cleared. We need to check previous exception because cold start is asynchronous. We don't know cold start error instantly.

If transport state gets cleared up, next few runs would trigger the same error.

ylwu-amzn · 2020-08-15T00:51:24Z

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

+                LOG.info("Trigger cold start for {}", detectorId);
+                coldStart(detector);
+            } else {
+                LOG.error(String.format("Fail to get checkpoint state for %s", detectorId), exception);


Should we throw exception ? Why not record this exception as previousException ?

This is in callback. We cannot throw exception directly as it is asynchronous and won't be caught by AD job. What do we expect the job runner to do here? For example, we can reach here if cold start does not happen before and the system is under extreme heavy load and get request gets rejected. We can log the error and retry later.

Job runner will end detector job if getsEndRunException and isEndNow is true. If isEndNow is false, will still stop detector job if the count of EndRunException exceeds MAX_RETRY_FOR_END_RUN_EXCEPTION(default 6 currently). For other exceptions except for EndRunException, job runner will record it in AD result.

How about we return this exception to JobRunner so it can be logged into AD result ?

In this change, cold start is put into another thread pool and run asynchronously; will check previousException if no checkpoint or no model. How about we check EndRunException which isEndNow is false and return it to JobRunner, so JobRunner can stop detector job if find too many EndRunExceptions ? That can avoid endlessly run cold start.

changed as we discussed offline

ylwu-amzn · 2020-08-15T00:55:21Z

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

    // detector has changed or not. If changed, trigger indexing.
-    private Entry<String, Instant> lastError;
+    private Entry<String, Instant> lastDetectionError;
+    // last training error. Used to save cold error by a concurrent cold start thread.


cold error means cold start error?

Good catch. Fixed

ylwu-amzn · 2020-08-15T00:59:24Z

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

+        }
+
+        if (!(exp instanceof ResourceNotFoundException)) {
+            throw exp;


Why not return this exception ?

The returned exception is for cold start related failure only.

…nsport state

wnbts · 2020-08-18T01:06:50Z

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

        searchFeatureDao
            .getLatestDataTime(
                detector,
-                ActionListener.wrap(latest -> getColdStartSamples(latest, detector, listener), listener::onFailure)
+                new ThreadedActionListener<>(logger, threadPool, AnomalyDetectorPlugin.AD_THREAD_POOL_NAME, latestTimeListener, false)


minor. why not using configured executor name instead of hardcoding it in all places?

I don't understand. Could you explain more?

AnomalyDetectorPlugin.AD_THREAD_POOL_NAME is the name.

why not using a configured name like threadpool in configuration? instead of hardcoding it.

What do you mean by configuration?

the value is configurable using constructor.

The benefit to make it configurable is to make testing easy.

My concern is we may not have just one threadpool to use in this class. For example, I want to add a dedicated cold start threadpool for high cardinality. So we should add threadpool_ad, threadpool_coldStart for this?

if there are additional dependency configuration (threadpool and executor names), they can also be passed via constructor and the implementation just needs to pick the corresponding configuration (threadpool and executor name combination)

wnbts · 2020-08-18T01:11:55Z

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

 import java.time.Duration;
 import java.time.Instant;
-import java.util.Map.Entry;
+import java.util.Optional;

 import com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetector;

 public class TransportState {


minor. documentation is missing for the interface of this class.

also all the public methods (interface)

wnbts · 2020-08-18T01:14:15Z

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

+        lastAccessTime = clock.instant();
+    }
+
+    public boolean expired(Duration stateTtl, Instant now) {


minor. the input can be simplified to one parameter as the expiration time.

I don't understand. Could you explain?

why using two input parameters for this method when one parameter such as expiration time is sufficient?

wnbts · 2020-08-18T01:27:11Z

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

@@ -397,32 +396,46 @@ protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<
     *
     * @param failure  object that may contain exceptions thrown
     * @param detector detector object
-     * @return whether cold start runs
+     * @return exception if we get resource not found exception


minor.

when resource is not found, what exception does this method return? why the general exception instead of the more specific one? if other exceptions occur, what does this method return?

also who is we? does it refer to the reader of the doc? the client of the interface? the developer of the implementation? or the software implementation? please avoid incorrect/ambiguous use of we when the referee is software.

Changed to new AnomalyDetectionException.

Is there any place saying we cannot use we?

it was the question about who we refers to. https://www.oracle.com/technical-resources/articles/java/javadoc-tool.html

Which part of the doc do you want me to read? Any particular part related?

Anyway, I changed to "AD job execution".

See Use 3rd person (descriptive) not 2nd person (prescriptive).

good to know

wnbts · 2020-08-18T01:42:07Z

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

            }
        }

+        stateManager.getDetectorCheckpoint(detectorId, ActionListener.wrap(checkpointExists -> {
+            if (!checkpointExists) {


question. what should be returned when the checkpoint exists?

This is a callback. We don't need to return anything.

wnbts · 2020-08-18T01:54:05Z

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

-                } else {
-                    throw new EndRunException(detectorId, "Cannot get training data", false);
-                }
+        Optional<Exception> previousException = stateManager.fetchColdStartException(detectorId);


question. why should exception be cleared only when data point is missing?

What do you mean by "data point is missing"?

look at when the method coldStartIfNoCheckPoint gets called

There is another method calling fetchColdStartException: coldStartIfNoModel

ylwu-amzn · 2020-08-18T19:51:41Z

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

            }
        }
-        return false;
+        LOG.info("Trigger cold start for {}", detector.getDetectorId());
+        coldStart(detector);


Is it possible that last cold start still not finished when execute coldStart ? Should we check the state of last cold start ?

It is possible. Let me add a check.

… last cold start exception, and more

wnbts · 2020-08-19T00:20:52Z

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

@@ -187,7 +188,7 @@ public void maintenance() {
            String detectorId = entry.getKey();
            try {
                TransportState state = entry.getValue();
-                if (state.expired(stateTtl, clock.instant())) {
+                if (state.expired(stateTtl)) {


minor. why not do expired(expirationdate=now-ttl)? ttl is an odd parameter.

expirationdate is a date, while ttl is a duration. ttl means time to leave.

the method can just take one input parameter as the final expiration time, which can be calculated by the client however they want. in this case, it is now-tll. the method is easier to understand and use and more reusable.

I don't understand. Expiration date = last access time + ttl. Why is it now-ttl?

so this method takes one parameter that is the expiration time. if the last access time is before it, the method returns true, otherwise it returns false.

In that case, all the caller has to compute the expiration time. Current implementation looks easier that encapsulate the logic inside.

wnbts · 2020-08-19T00:29:01Z

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

 import java.time.Duration;
 import java.time.Instant;
-import java.util.Map.Entry;
+import java.util.Optional;

 import com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetector;

 public class TransportState {


also all the public methods (interface)

ylwu-amzn

LGTM. Thanks for the change.

* Change to use callbacks in cold start This PR changes the code path of cold start in the transport layer to use callbacks. Previously, I created AD’s ExecutorService that has one thread for cold starts in ColdStartRunner. When we need to trigger a cold start, we can submit a task in the ExecutorService, and consult a hash map (keyed by detector Id) that cached the results of recent cold start results. Since I have to invoke the cold start thread in various callbacks, I created a cold start thread pool and put the cold start result in the transport state. This PR also handles new exceptions like invalid queries introduced by recent changes on ModelManager and FeatureManager. This PR lowers the severity of a couple of log messages in HashRing and RCFPollingTransportAction to avoid overwhelming readers of log files. These log messages are common. This PR corrects typos and updates known causes of EndRunException in comments. Testing done: 1. Simulated cold start failures: Exceptions of cold starts can be seen by the transport layer. EndRunException can cause AD jobs to be terminated. 2. Happy case of a cold start still works.

kaituo requested review from wnbts and ylwu-amzn August 11, 2020 01:28

VijayanB reviewed Aug 11, 2020

View reviewed changes

...in/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java Outdated Show resolved Hide resolved

VijayanB reviewed Aug 11, 2020

View reviewed changes

...in/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyResultTransportAction.java Show resolved Hide resolved

kaituo added the enhancement New feature or request label Aug 12, 2020

wnbts reviewed Aug 12, 2020

View reviewed changes

VijayanB approved these changes Aug 13, 2020

View reviewed changes

kaituo force-pushed the coldstart branch from 7e3065f to 49f82fe Compare August 13, 2020 22:37

ylwu-amzn reviewed Aug 15, 2020

View reviewed changes

kaituo added 2 commits August 17, 2020 11:34

Change error to use Optional and use a common last update time in tra…

8ec1870

…nsport state

Update comment and log

2f63d9e

wnbts reviewed Aug 18, 2020

View reviewed changes

ylwu-amzn reviewed Aug 18, 2020

View reviewed changes

Make threadpool name configurable and add get checkpoint exception to…

72dc0e8

… last cold start exception, and more

wnbts approved these changes Aug 19, 2020

View reviewed changes

Add comments to transport state methods

bb86d96

ylwu-amzn approved these changes Aug 19, 2020

View reviewed changes

kaituo merged commit 92a5867 into opendistro-for-elasticsearch:master Aug 19, 2020

		if (exception instanceof AnomalyDetectionException) {
		// partitioned model exceeds memory limit

Change to use callbacks in cold start #208

Change to use callbacks in cold start #208

Conversation

kaituo commented Aug 11, 2020

codecov bot commented Aug 11, 2020 • edited

Codecov Report

VijayanB Aug 11, 2020 • edited

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

kaituo Aug 17, 2020 • edited

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

VijayanB left a comment

Choose a reason for hiding this comment

kaituo commented Aug 14, 2020

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 Aug 15, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kaituo Aug 17, 2020 • edited

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

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

kaituo Aug 18, 2020 • edited

codecov bot commented Aug 11, 2020 •

edited

VijayanB Aug 11, 2020 •

edited

kaituo Aug 17, 2020 •

edited

ylwu-amzn Aug 15, 2020 •

edited

kaituo Aug 17, 2020 •

edited

kaituo Aug 18, 2020 •

edited