state machine, separate config index, suggest & validate

Also, save cold start results for run once visualization.

Signed-off-by: Kaituo Li <kaituo@amazon.com>

Author: kaituo

build.gradle

@@ -164,6 +164,7 @@ dependencies {
     testImplementation "org.opensearch:opensearch-core:${opensearch_version}"
     testRuntimeOnly("org.junit.platform:junit-platform-launcher:1.11.2")
     testCompileOnly 'junit:junit:4.13.2'
+    testImplementation 'org.reflections:reflections:0.10.2'
 }
 
 apply plugin: 'java'

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

@@ -40,7 +40,8 @@ public ADEntityProfileRunner(
             AnalysisType.AD,
             ADEntityProfileAction.INSTANCE,
             ADCommonName.ANOMALY_RESULT_INDEX_ALIAS,
-            AnomalyResult.DETECTOR_ID_FIELD
+            AnomalyResult.DETECTOR_ID_FIELD,
+            ADCommonName.CONFIG_INDEX
         );
     }
 }

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

@@ -13,6 +13,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.ad.constant.ADCommonName;
 import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.ADIndexManagement;
 import org.opensearch.ad.model.ADTask;
@@ -71,7 +72,8 @@ public AnomalyDetectorProfileRunner(
             ProfileName.AD_TASK,
             ADProfileAction.INSTANCE,
             AnomalyDetector::parse,
-            taskProfileRunner
+            taskProfileRunner,
+            ADCommonName.CONFIG_INDEX
         );
     }
 

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

@@ -11,6 +11,7 @@
 
 package org.opensearch.ad;
 
+import java.time.Clock;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Optional;
@@ -50,7 +51,7 @@ public ExecuteADResultResponseRecorder(
         ThreadPool threadPool,
         Client client,
         NodeStateManager nodeStateManager,
-        ADTaskCacheManager taskCacheManager,
+        Clock clock,
         int rcfMinSamples
     ) {
         super(
@@ -62,7 +63,7 @@ public ExecuteADResultResponseRecorder(
             TimeSeriesAnalyticsPlugin.AD_THREAD_POOL_NAME,
             client,
             nodeStateManager,
-            taskCacheManager,
+            clock,
             rcfMinSamples,
             ADIndex.RESULT,
             AnalysisType.AD,
@@ -104,14 +105,15 @@ protected AnomalyResult createErrorResult(
      * Instead, we issue a profile request to poll each model node and get the maximum total updates among all models.
      * @param response response returned from executing AnomalyResultAction
      * @param configId config Id
+     * @param clock Clock to get current time
      */
     @Override
-    protected void updateRealtimeTask(ResultResponse<AnomalyResult> response, String configId) {
+    protected void updateRealtimeTask(ResultResponse<AnomalyResult> response, String configId, Clock clock) {
         if (response.isHC() != null && response.isHC()) {
             if (taskManager.skipUpdateRealtimeTask(configId, response.getError())) {
                 return;
             }
-            delayedUpdate(response, configId);
+            delayedUpdate(response, configId, clock);
         } else {
             log
                 .debug(
@@ -124,7 +126,8 @@ protected void updateRealtimeTask(ResultResponse<AnomalyResult> response, String
                 null,
                 response.getRcfTotalUpdates(),
                 response.getConfigIntervalInMinutes(),
-                response.getError()
+                response.getError(),
+                clock
             );
         }
     }

src/main/java/org/opensearch/ad/constant/ADCommonName.java

@@ -19,9 +19,12 @@ public class ADCommonName {
     public static final String CHECKPOINT_INDEX_NAME = ".opendistro-anomaly-checkpoints";
     // index name for anomaly detection state. Will store AD task in this index as well.
     public static final String DETECTION_STATE_INDEX = ".opendistro-anomaly-detection-state";
+    // config index. We are reusing ad detector index.
+    public static final String CONFIG_INDEX = ".opendistro-anomaly-detectors";
 
     // The alias of the index in which to write AD result history
     public static final String ANOMALY_RESULT_INDEX_ALIAS = ".opendistro-anomaly-results";
+    public static final String CONFIG_INDEX_NAME = "";
 
     // ======================================
     // Anomaly Detector name for X-Opaque-Id header

src/main/java/org/opensearch/ad/indices/ADIndex.java

@@ -30,7 +30,7 @@ public enum ADIndex implements TimeSeriesIndex {
         true,
         ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getResultMappings)
     ),
-    CONFIG(CommonName.CONFIG_INDEX, false, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getConfigMappings)),
+    CONFIG(ADCommonName.CONFIG_INDEX, false, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getConfigMappings)),
     JOB(CommonName.JOB_INDEX, false, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getJobMappings)),
     CHECKPOINT(
         ADCommonName.CHECKPOINT_INDEX_NAME,
@@ -65,4 +65,9 @@ public boolean isAlias() {
     public String getMapping() {
         return mapping;
     }
+
+    @Override
+    public boolean isConfigIndex() {
+        return ADCommonName.CONFIG_INDEX.equals(getIndexName());
+    }
 }

src/main/java/org/opensearch/ad/indices/ADIndexManagement.java

@@ -94,7 +94,8 @@ public ADIndexManagement(
             ADIndex.RESULT.getMapping(),
             xContentRegistry,
             AnomalyDetector::parse,
-            ADCommonName.CUSTOM_RESULT_INDEX_PREFIX
+            ADCommonName.CUSTOM_RESULT_INDEX_PREFIX,
+            ADCommonName.CONFIG_INDEX
         );
 
         this.indexStates = new EnumMap<ADIndex, IndexState>(ADIndex.class);

src/main/java/org/opensearch/ad/ml/ADColdStart.java

@@ -13,15 +13,21 @@
 
 import java.time.Clock;
 import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.ADIndexManagement;
 import org.opensearch.ad.ml.IgnoreSimilarExtractor.ThresholdArrays;
 import org.opensearch.ad.model.AnomalyDetector;
+import org.opensearch.ad.model.AnomalyResult;
 import org.opensearch.ad.ratelimit.ADCheckpointWriteWorker;
+import org.opensearch.forecast.ml.RCFCasterResult;
+import org.opensearch.forecast.model.ForecastResult;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.timeseries.AnalysisType;
 import org.opensearch.timeseries.NodeStateManager;
@@ -34,18 +40,21 @@
 import org.opensearch.timeseries.model.Config;
 import org.opensearch.timeseries.ratelimit.RequestPriority;
 import org.opensearch.timeseries.settings.TimeSeriesSettings;
+import org.opensearch.timeseries.util.ModelUtil;
+import org.opensearch.timeseries.util.ParseUtils;
 
 import com.amazon.randomcutforest.config.ForestMode;
 import com.amazon.randomcutforest.config.Precision;
 import com.amazon.randomcutforest.config.TransformMethod;
+import com.amazon.randomcutforest.parkservices.AnomalyDescriptor;
 import com.amazon.randomcutforest.parkservices.ThresholdedRandomCutForest;
 
 /**
  * Training models for HCAD detectors
  *
  */
 public class ADColdStart extends
-    ModelColdStart<ThresholdedRandomCutForest, ADIndex, ADIndexManagement, ADCheckpointDao, ADCheckpointWriteWorker> {
+    ModelColdStart<ThresholdedRandomCutForest, ADIndex, ADIndexManagement, ADCheckpointDao, ADCheckpointWriteWorker, AnomalyResult> {
     private static final Logger logger = LogManager.getLogger(ADColdStart.class);
 
     /**
@@ -87,7 +96,8 @@ public ADColdStart(
         ADCheckpointWriteWorker checkpointWriteWorker,
         long rcfSeed,
         int maxRoundofColdStart,
-        int coolDownMinutes
+        int coolDownMinutes,
+        int resultSchemaVersion
     ) {
         super(
             modelTtl,
@@ -107,7 +117,8 @@ public ADColdStart(
             featureManager,
             maxRoundofColdStart,
             TimeSeriesAnalyticsPlugin.AD_THREAD_POOL_NAME,
-            AnalysisType.AD
+            AnalysisType.AD,
+            resultSchemaVersion
         );
     }
 
@@ -126,7 +137,8 @@ public ADColdStart(
         Duration modelTtl,
         ADCheckpointWriteWorker checkpointWriteQueue,
         int maxRoundofColdStart,
-        int coolDownMinutes
+        int coolDownMinutes,
+        int resultSchemaVersion
     ) {
         this(
             clock,
@@ -144,7 +156,8 @@ public ADColdStart(
             checkpointWriteQueue,
             -1,
             maxRoundofColdStart,
-            coolDownMinutes
+            coolDownMinutes,
+            resultSchemaVersion
         );
     }
 
@@ -158,7 +171,7 @@ public ADColdStart(
      * training data in result index so that the frontend can plot it.
      */
     @Override
-    protected List<Sample> trainModelFromDataSegments(
+    protected List<AnomalyResult> trainModelFromDataSegments(
         List<Sample> pointSamples,
         ModelState<ThresholdedRandomCutForest> entityState,
         Config config,
@@ -185,6 +198,7 @@ protected List<Sample> trainModelFromDataSegments(
             .numberOfTrees(numberOfTrees)
             .timeDecay(config.getTimeDecay())
             .transformDecay(config.getTimeDecay())
+            // allow enough samples before emitting scores to park service
             .outputAfter(Math.max(shingleSize, numMinSamples))
             .initialAcceptFraction(initialAcceptFraction)
             .parallelExecutionEnabled(false)
@@ -221,21 +235,71 @@ protected List<Sample> trainModelFromDataSegments(
         // use build instead of new TRCF(Builder) because build method did extra validation and initialization
         ThresholdedRandomCutForest trcf = rcfBuilder.build();
 
+     // Prepare for sequential processing
+        double[][] sequentialData = new double[pointSamples.size()][];
+        List<Pair<Instant, Instant>> sequentialTime = new ArrayList<>();
+
+        // Convert the list of Sample objects into a 2D array + a parallel list of time pairs
         for (int i = 0; i < pointSamples.size(); i++) {
             Sample dataSample = pointSamples.get(i);
             double[] dataValue = dataSample.getValueList();
-            // We don't keep missing values during cold start as the actual data may not be reconstructed during the early stage.
-            trcf.process(dataValue, dataSample.getDataEndTime().getEpochSecond());
+
+            sequentialData[i] = dataValue;
+            // Store start and end times together
+            sequentialTime.add(Pair.of(dataSample.getDataStartTime(), dataSample.getDataEndTime()));
+        }
+
+        // Process data in one go
+        List<AnomalyDescriptor> descriptors = trcf.processSequentially(sequentialData, x -> true);
+
+        // Check for size mismatch
+        if (descriptors.size() != sequentialTime.size()) {
+            logger.warn(
+                "processSequentially returned a different size than expected: got [{}], expected [{}].",
+                descriptors.size(),
+                sequentialTime.size()
+            );
+            return null;
+        }
+
+        // Build anomaly results from sequential descriptors
+        List<AnomalyResult> results = new ArrayList<>();
+        for (int i = 0; i < descriptors.size(); i++) {
+            AnomalyDescriptor descriptor = descriptors.get(i);
+            double[] dataValue = sequentialData[i];
+            Pair<Instant, Instant> time = sequentialTime.get(i);
+
+            // Convert the descriptor into a thresholding result, or anomaly result
+            ThresholdingResult thresholdingResult =
+                ModelUtil.toResult(trcf.getForest(), descriptor, dataValue, false, config);
+
+            Instant now = Instant.now();
+            results.addAll(
+                thresholdingResult.toIndexableResults(
+                    config,
+                    time.getLeft(),            // Data start time
+                    time.getRight(),           // Data end time
+                    now,
+                    now,
+                    ParseUtils.getFeatureData(dataValue, config),
+                    entityState.getEntity(),
+                    resultMappingVersion,
+                    entityState.getModelId(),
+                    taskId,
+                    null
+                )
+            );
         }
 
+
         entityState.setModel(trcf);
 
         entityState.setLastUsedTime(clock.instant());
 
         // save to checkpoint
         checkpointWriteWorker.write(entityState, true, RequestPriority.MEDIUM);
 
-        return pointSamples;
+        return results;
     }
 
     public static void applyRule(ThresholdedRandomCutForest.Builder rcfBuilder, AnomalyDetector detector) {