[Backport to 2.x] Speed up cold start (#753)

build.gradle

@@ -134,7 +134,7 @@ configurations.all {
     if (it.state != Configuration.State.UNRESOLVED) return
     resolutionStrategy {
         force "joda-time:joda-time:${versions.joda}"
-        force "com.fasterxml.jackson.core:jackson-core:2.14.0"
+        force "com.fasterxml.jackson.core:jackson-core:2.14.1"
         force "commons-logging:commons-logging:${versions.commonslogging}"
         force "org.apache.httpcomponents:httpcore:${versions.httpcore}"
         force "commons-codec:commons-codec:${versions.commonscodec}"
@@ -677,9 +677,9 @@ dependencies {
     implementation 'software.amazon.randomcutforest:randomcutforest-core:3.0-rc3'
 
     // force Jackson version to avoid version conflict issue
-    implementation "com.fasterxml.jackson.core:jackson-core:2.14.0"
-    implementation "com.fasterxml.jackson.core:jackson-databind:2.14.0"
-    implementation "com.fasterxml.jackson.core:jackson-annotations:2.14.0"
+    implementation "com.fasterxml.jackson.core:jackson-core:2.14.1"
+    implementation "com.fasterxml.jackson.core:jackson-databind:2.14.1"
+    implementation "com.fasterxml.jackson.core:jackson-annotations:2.14.1"
 
     // used for serializing/deserializing rcf models.
     implementation group: 'io.protostuff', name: 'protostuff-core', version: '1.8.0'

src/main/java/org/opensearch/ad/AnomalyDetectorPlugin.java

@@ -234,11 +234,12 @@ public class AnomalyDetectorPlugin extends Plugin implements ActionPlugin, Scrip
     private ClientUtil clientUtil;
     private DiscoveryNodeFilterer nodeFilter;
     private IndexUtils indexUtils;
-    private ADTaskCacheManager adTaskCacheManager;
     private ADTaskManager adTaskManager;
     private ADBatchTaskRunner adBatchTaskRunner;
     // package private for testing
     GenericObjectPool<LinkedBuffer> serializeRCFBufferPool;
+    private NodeStateManager stateManager;
+    private ExecuteADResultResponseRecorder adResultResponseRecorder;
 
     static {
         SpecialPermission.check();
@@ -259,25 +260,14 @@ public List<RestHandler> getRestHandlers(
         IndexNameExpressionResolver indexNameExpressionResolver,
         Supplier<DiscoveryNodes> nodesInCluster
     ) {
-        AnomalyIndexHandler<AnomalyResult> anomalyResultHandler = new AnomalyIndexHandler<AnomalyResult>(
-            client,
-            settings,
-            threadPool,
-            CommonName.ANOMALY_RESULT_INDEX_ALIAS,
-            anomalyDetectionIndices,
-            this.clientUtil,
-            this.indexUtils,
-            clusterService
-        );
-
         AnomalyDetectorJobRunner jobRunner = AnomalyDetectorJobRunner.getJobRunnerInstance();
         jobRunner.setClient(client);
         jobRunner.setThreadPool(threadPool);
-        jobRunner.setAnomalyResultHandler(anomalyResultHandler);
         jobRunner.setSettings(settings);
         jobRunner.setAnomalyDetectionIndices(anomalyDetectionIndices);
-        jobRunner.setNodeFilter(nodeFilter);
         jobRunner.setAdTaskManager(adTaskManager);
+        jobRunner.setNodeStateManager(stateManager);
+        jobRunner.setExecuteADResultResponseRecorder(adResultResponseRecorder);
 
         RestGetAnomalyDetectorAction restGetAnomalyDetectorAction = new RestGetAnomalyDetectorAction();
         RestIndexAnomalyDetectorAction restIndexAnomalyDetectorAction = new RestIndexAnomalyDetectorAction(settings, clusterService);
@@ -383,7 +373,7 @@ public Collection<Object> createComponents(
             adCircuitBreakerService
         );
 
-        NodeStateManager stateManager = new NodeStateManager(
+        stateManager = new NodeStateManager(
             client,
             xContentRegistry,
             settings,
@@ -568,7 +558,8 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
             AnomalyDetectorSettings.QUEUE_MAINTENANCE,
             entityColdStarter,
             AnomalyDetectorSettings.HOURLY_MAINTENANCE,
-            stateManager
+            stateManager,
+            cacheProvider
         );
 
         ModelManager modelManager = new ModelManager(
@@ -714,7 +705,7 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
 
         anomalyDetectorRunner = new AnomalyDetectorRunner(modelManager, featureManager, AnomalyDetectorSettings.MAX_PREVIEW_RESULTS);
 
-        adTaskCacheManager = new ADTaskCacheManager(settings, clusterService, memoryTracker);
+        ADTaskCacheManager adTaskCacheManager = new ADTaskCacheManager(settings, clusterService, memoryTracker);
         adTaskManager = new ADTaskManager(
             settings,
             clusterService,
@@ -754,6 +745,26 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
 
         ADSearchHandler adSearchHandler = new ADSearchHandler(settings, clusterService, client);
 
+        AnomalyIndexHandler<AnomalyResult> anomalyResultHandler = new AnomalyIndexHandler<AnomalyResult>(
+            client,
+            settings,
+            threadPool,
+            CommonName.ANOMALY_RESULT_INDEX_ALIAS,
+            anomalyDetectionIndices,
+            this.clientUtil,
+            this.indexUtils,
+            clusterService
+        );
+
+        adResultResponseRecorder = new ExecuteADResultResponseRecorder(
+            anomalyDetectionIndices,
+            anomalyResultHandler,
+            adTaskManager,
+            nodeFilter,
+            threadPool,
+            client
+        );
+
         // return objects used by Guice to inject dependencies for e.g.,
         // transport action handler constructors
         return ImmutableList
@@ -795,7 +806,8 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
                 checkpointWriteQueue,
                 coldEntityQueue,
                 entityColdStarter,
-                adTaskCacheManager
+                adTaskCacheManager,
+                adResultResponseRecorder
             );
     }