HCAD model performance benchmark

This PR adds a HCAD model performance benchmark so that we can compare model performance across versions.

Regarding benchmark data, we randomly generated synthetic data with known anomalies inserted throughout the signal. In particular, these are one/two/four dimensional data where each dimension is a noisy cosine wave. Anomalies are inserted into one dimension with 0.003 probability. Anomalies across each dimension can be independent or dependent.  We have approximately 5000 observations per data set. The data set is generated using the same random seed so the result is comparable across versions.

We also backported opensearch-project#600 so that we can capture the performance data in CI output.

Testing done:
1. added unit tests to run the benchmark.

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

.github/workflows/CI.yml

@@ -75,7 +75,7 @@ jobs:
 
       - name: Build and Run Tests
         run: |
-          ./gradlew build -Dopensearch.version=1.0.0
+          ./gradlew build -Dopensearch.version=1.0.0 -Dtest.logs=true
 
       - name: Publish to Maven Local
         run: |

build.gradle

@@ -33,6 +33,10 @@ buildscript {
         opensearch_version = System.getProperty("opensearch.version", "1.0.0")
         common_utils_version = '1.0.0.0'
         job_scheduler_version = '1.0.0.0'
+        // gradle build won't print logs during test by default unless there is a failure.
+        // It is useful to record intermediately information like prediction precision and recall.
+        // This option turn on log printing during tests. 
+        printLogs = "true" == System.getProperty("test.logs", "false")
     }
 
     repositories {
@@ -206,6 +210,12 @@ integTest {
         jvmArgs '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:5005'
     }
 
+    if (printLogs) {
+        testLogging {
+            showStandardStreams = true
+            outputs.upToDateWhen {false}
+        }
+    }
 }
 
 testClusters.integTest {
@@ -485,3 +495,13 @@ validateNebulaPom.enabled = false
 tasks.withType(licenseHeaders.class) {
     additionalLicense 'AL   ', 'Apache', 'Licensed under the Apache License, Version 2.0 (the "License")'
 }
+
+// show test results so that we can record information like precion/recall results of correctness testing.
+if (printLogs) {
+    test {
+        testLogging {
+            showStandardStreams = true
+            outputs.upToDateWhen {false}
+        }
+    }
+}

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

@@ -76,6 +76,7 @@ public AnomalyDetectorRunner(ModelManager modelManager, FeatureManager featureMa
      * @param detector  anomaly detector instance
      * @param startTime detection period start time
      * @param endTime   detection period end time
+     * @param context Thread context when making the call
      * @param listener handle anomaly result
      * @throws IOException - if a user gives wrong query input when defining a detector
      */

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

@@ -97,6 +97,7 @@ public class EntityColdStarter {
     private final Cache<String, Instant> lastColdStartTime;
     private final CheckpointDao checkpointDao;
     private int coolDownMinutes;
+    private long rcfSeed;
 
     /**
      * Constructor
@@ -150,6 +151,60 @@ public EntityColdStarter(
         long maxCacheSize,
         CheckpointDao checkpointDao,
         Settings settings
+    ) {
+        this(
+            clock,
+            threadPool,
+            nodeStateManager,
+            rcfSampleSize,
+            numberOfTrees,
+            rcfTimeDecay,
+            numMinSamples,
+            maxSampleStride,
+            maxTrainSamples,
+            interpolator,
+            searchFeatureDao,
+            shingleSize,
+            thresholdMinPvalue,
+            thresholdMaxRankError,
+            thresholdMaxScore,
+            thresholdNumLogNormalQuantiles,
+            thresholdDownsamples,
+            thresholdMaxSamples,
+            featureManager,
+            lastColdStartTimestampTtl,
+            maxCacheSize,
+            checkpointDao,
+            settings,
+            -1
+        );
+    }
+
+    public EntityColdStarter(
+        Clock clock,
+        ThreadPool threadPool,
+        NodeStateManager nodeStateManager,
+        int rcfSampleSize,
+        int numberOfTrees,
+        double rcfTimeDecay,
+        int numMinSamples,
+        int maxSampleStride,
+        int maxTrainSamples,
+        Interpolator interpolator,
+        SearchFeatureDao searchFeatureDao,
+        int shingleSize,
+        double thresholdMinPvalue,
+        double thresholdMaxRankError,
+        double thresholdMaxScore,
+        int thresholdNumLogNormalQuantiles,
+        int thresholdDownsamples,
+        long thresholdMaxSamples,
+        FeatureManager featureManager,
+        Duration lastColdStartTimestampTtl,
+        long maxCacheSize,
+        CheckpointDao checkpointDao,
+        Settings settings,
+        long rcfSeed
     ) {
         this.clock = clock;
         this.lastThrottledColdStartTime = Instant.MIN;
@@ -180,6 +235,7 @@ public EntityColdStarter(
             .build();
         this.checkpointDao = checkpointDao;
         this.coolDownMinutes = (int) (COOLDOWN_MINUTES.get(settings).getMinutes());
+        this.rcfSeed = rcfSeed;
     }
 
     /**
@@ -268,15 +324,19 @@ private void trainModelFromDataSegments(List<double[][]> dataPoints, String mode
         }
 
         int rcfNumFeatures = dataPoints.get(0)[0].length;
-        RandomCutForest rcf = RandomCutForest
+        RandomCutForest.Builder rcfBuilder = RandomCutForest
             .builder()
             .dimensions(rcfNumFeatures)
             .sampleSize(rcfSampleSize)
             .numberOfTrees(numberOfTrees)
             .lambda(rcfTimeDecay)
             .outputAfter(numMinSamples)
-            .parallelExecutionEnabled(false)
-            .build();
+            .parallelExecutionEnabled(false);
+
+        if (rcfSeed != -1) {
+            rcfBuilder.randomSeed(rcfSeed);
+        }
+        RandomCutForest rcf = rcfBuilder.build();
         List<double[]> allScores = new ArrayList<>();
         int totalLength = 0;
         // get continuous data points and send for training

src/test/java/org/opensearch/ad/TestHelpers.java

@@ -51,6 +51,7 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.Callable;
@@ -1028,4 +1029,133 @@ public static Map<String, Object> parseStatsResult(String statsResult) throws IO
         }
         return adStats;
     }
+
+    public static class AnomalyDetectorBuilder {
+        private String detectorId = randomAlphaOfLength(10);
+        private Long version = randomLong();
+        private String name = randomAlphaOfLength(20);
+        private String description = randomAlphaOfLength(30);
+        private String timeField = randomAlphaOfLength(5);
+        private List<String> indices = ImmutableList.of(randomAlphaOfLength(10).toLowerCase(Locale.ROOT));
+        private List<Feature> featureAttributes = ImmutableList.of(randomFeature(true));
+        private QueryBuilder filterQuery;
+        private TimeConfiguration detectionInterval = randomIntervalTimeConfiguration();
+        private TimeConfiguration windowDelay = randomIntervalTimeConfiguration();
+        private Integer shingleSize = randomIntBetween(1, 100);
+        private Map<String, Object> uiMetadata = null;
+        private Integer schemaVersion = randomInt();
+        private Instant lastUpdateTime = Instant.now().truncatedTo(ChronoUnit.SECONDS);
+        private List<String> categoryFields = null;
+        private User user = randomUser();
+        private String resultIndex = null;
+
+        public static AnomalyDetectorBuilder newInstance() throws IOException {
+            return new AnomalyDetectorBuilder();
+        }
+
+        private AnomalyDetectorBuilder() throws IOException {
+            filterQuery = randomQuery();
+        }
+
+        public AnomalyDetectorBuilder setDetectorId(String detectorId) {
+            this.detectorId = detectorId;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setVersion(Long version) {
+            this.version = version;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setName(String name) {
+            this.name = name;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setDescription(String description) {
+            this.description = description;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setTimeField(String timeField) {
+            this.timeField = timeField;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setIndices(List<String> indices) {
+            this.indices = indices;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setFeatureAttributes(List<Feature> featureAttributes) {
+            this.featureAttributes = featureAttributes;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setFilterQuery(QueryBuilder filterQuery) {
+            this.filterQuery = filterQuery;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setDetectionInterval(TimeConfiguration detectionInterval) {
+            this.detectionInterval = detectionInterval;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setWindowDelay(TimeConfiguration windowDelay) {
+            this.windowDelay = windowDelay;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setShingleSize(Integer shingleSize) {
+            this.shingleSize = shingleSize;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setUiMetadata(Map<String, Object> uiMetadata) {
+            this.uiMetadata = uiMetadata;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setSchemaVersion(Integer schemaVersion) {
+            this.schemaVersion = schemaVersion;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setLastUpdateTime(Instant lastUpdateTime) {
+            this.lastUpdateTime = lastUpdateTime;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setCategoryFields(List<String> categoryFields) {
+            this.categoryFields = categoryFields;
+            return this;
+        }
+
+        public AnomalyDetectorBuilder setUser(User user) {
+            this.user = user;
+            return this;
+        }
+
+        public AnomalyDetector build() {
+            return new AnomalyDetector(
+                detectorId,
+                version,
+                name,
+                description,
+                timeField,
+                indices,
+                featureAttributes,
+                filterQuery,
+                detectionInterval,
+                windowDelay,
+                shingleSize,
+                uiMetadata,
+                schemaVersion,
+                lastUpdateTime,
+                categoryFields,
+                user
+            );
+        }
+    }
 }

src/test/java/org/opensearch/ad/e2e/DetectionResultEvalutationIT.java

@@ -44,6 +44,8 @@
 
 import org.apache.http.HttpHeaders;
 import org.apache.http.message.BasicHeader;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Logger;
 import org.opensearch.ad.ODFERestTestCase;
 import org.opensearch.ad.TestHelpers;
 import org.opensearch.client.Request;
@@ -57,6 +59,7 @@
 import com.google.gson.JsonParser;
 
 public class DetectionResultEvalutationIT extends ODFERestTestCase {
+    protected static final Logger LOG = (Logger) LogManager.getLogger(DetectionResultEvalutationIT.class);
 
     public void testDataset() throws Exception {
         // TODO: this test case will run for a much longer time and timeout with security enabled
@@ -112,6 +115,7 @@ private void verifyTestResults(
         assertTrue(recall >= minRecall);
 
         assertTrue(errors <= maxError);
+        LOG.info("Precision: {}, Window recall: {}", precision, recall);
     }
 
     private int isAnomaly(Instant time, List<Entry<Instant, Instant>> labels) {