test

Signed-off-by: Kaituo Li <[email protected]>
kaituo · Aug 22, 2024 · a8637d0 · a8637d0
1 parent dc85dc4
commit a8637d0
Show file tree

Hide file tree

Showing 11 changed files with 83 additions and 24 deletions.
diff --git a/src/main/java/org/opensearch/timeseries/dataprocessor/ImputationOption.java b/src/main/java/org/opensearch/timeseries/dataprocessor/ImputationOption.java
@@ -26,7 +26,7 @@
 public class ImputationOption implements Writeable, ToXContent {
     // field name in toXContent
     public static final String METHOD_FIELD = "method";
-    public static final String DEFAULT_FILL_FIELD = "defaultFill";
+    public static final String DEFAULT_FILL_FIELD = "default_fill";
 
     private final ImputationMethod method;
     private final Map<String, Double> defaultFill;
@@ -152,7 +152,7 @@ public int hashCode() {
 
     @Override
     public String toString() {
-        return new ToStringBuilder(this).append("method", method).append("defaultFill", defaultFill).toString();
+        return new ToStringBuilder(this).append("method", method).append("default_fill", defaultFill).toString();
     }
 
     public ImputationMethod getMethod() {

diff --git a/src/main/java/org/opensearch/timeseries/ml/ModelManager.java b/src/main/java/org/opensearch/timeseries/ml/ModelManager.java
@@ -169,6 +169,7 @@ public <RCFDescriptor extends AnomalyDescriptor> IntermediateResultType score(
             throw e;
         } finally {
             modelState.setLastUsedTime(clock.instant());
+            modelState.setLastProcessedDataEndTime(sample.getDataEndTime());
         }
         return createEmptyResult();
     }

diff --git a/src/main/java/org/opensearch/timeseries/ml/ModelState.java b/src/main/java/org/opensearch/timeseries/ml/ModelState.java
@@ -36,6 +36,7 @@ public class ModelState<T> implements org.opensearch.timeseries.ExpiringState {
     // time when the ML model was used last time
     protected Instant lastUsedTime;
     protected Instant lastCheckpointTime;
+    protected Instant lastProcessedDataEndTime;
     protected Clock clock;
     protected float priority;
     protected Deque<Sample> samples;
@@ -74,6 +75,7 @@ public ModelState(
         this.priority = priority;
         this.entity = entity;
         this.samples = samples;
+        this.lastProcessedDataEndTime = Instant.MIN;
     }
 
     /**
@@ -249,4 +251,12 @@ public Map<String, Object> getModelStateAsMap() {
             }
         };
     }
+
+    public Instant getLastProcessedDataEndTime() {
+        return lastProcessedDataEndTime;
+    }
+
+    public void setLastProcessedDataEndTime(Instant lastProcessedDataEndTime) {
+        this.lastProcessedDataEndTime = lastProcessedDataEndTime;
+    }
 }
diff --git a/src/main/java/org/opensearch/timeseries/ml/RealTimeInferencer.java b/src/main/java/org/opensearch/timeseries/ml/RealTimeInferencer.java
@@ -134,11 +134,13 @@ private boolean processWithTimeout(
     }
 
     private boolean tryProcess(Sample sample, ModelState<RCFModelType> modelState, Config config, String taskId, long curExecutionEnd) {
-        // execution end time (when job starts execution in this interval) >= last used time => the model state is updated in
+        // execution end time (when job starts execution in this interval) >= last processed data end time => the model state is updated in
         // previous intervals
-        // This can happen while scheduled to waiting some other threads have already scored the same interval (e.g., during tests
-        // when everything happens fast)
-        if (curExecutionEnd < modelState.getLastUsedTime().toEpochMilli()) {
+        // This branch being true can happen while scheduled to waiting some other threads have already scored the same interval
+        // (e.g., during tests when everything happens fast)
+        // We cannot use last used time as it will be updated whenever we update its priority in CacheBuffer.update when there is a
+        // PriorityCache.get.
+        if (curExecutionEnd < modelState.getLastProcessedDataEndTime().toEpochMilli()) {
             return false;
         }
         String modelId = modelState.getModelId();

diff --git a/src/main/java/org/opensearch/timeseries/model/Config.java b/src/main/java/org/opensearch/timeseries/model/Config.java
@@ -225,29 +225,36 @@ protected Config(
                 : features.stream().filter(Feature::getEnabled).collect(Collectors.toList());
 
             Map<String, Double> defaultFill = imputationOption.getDefaultFill();
-            if (defaultFill.isEmpty() && enabledFeatures.size() > 0) {
+
+            // Case 1: enabledFeatures == null && defaultFill != null
+            if (enabledFeatures == null && defaultFill != null && !defaultFill.isEmpty()) {
+                issueType = ValidationIssueType.IMPUTATION;
+                errorMessage = "Enabled features list is null, but default fill values are provided.";
+                return;
+            }
+
+            // Case 2: enabledFeatures != null && defaultFill == null
+            if (enabledFeatures != null && (defaultFill == null || defaultFill.isEmpty())) {
                 issueType = ValidationIssueType.IMPUTATION;
-                errorMessage = "No given values for fixed value imputation";
+                errorMessage = "Enabled features are present, but no default fill values are provided.";
                 return;
             }
 
-            // Check if the length of the defaultFill array matches the number of expected features
-            if (enabledFeatures == null || defaultFill.size() != enabledFeatures.size()) {
+            // Case 3: enabledFeatures.size() != defaultFill.size()
+            if (enabledFeatures != null && defaultFill != null && defaultFill.size() != enabledFeatures.size()) {
                 issueType = ValidationIssueType.IMPUTATION;
                 errorMessage = String
                     .format(
                         Locale.ROOT,
-                        "Incorrect number of values to fill. Got: %d. Expected: %d.",
+                        "Mismatch between the number of enabled features and default fill values. Number of default fill values: %d. Number of enabled features: %d.",
                         defaultFill.size(),
-                        enabledFeatures == null ? 0 : enabledFeatures.size()
+                        enabledFeatures.size()
                     );
                 return;
             }
 
-            Map<String, Double> defaultFills = imputationOption.getDefaultFill();
-
             for (int i = 0; i < enabledFeatures.size(); i++) {
-                if (!defaultFills.containsKey(enabledFeatures.get(i).getName())) {
+                if (!defaultFill.containsKey(enabledFeatures.get(i).getName())) {
                     issueType = ValidationIssueType.IMPUTATION;
                     errorMessage = String.format(Locale.ROOT, "Missing feature name: %s.", enabledFeatures.get(i).getName());
                     return;

diff --git a/src/main/resources/mappings/config.json b/src/main/resources/mappings/config.json
@@ -174,7 +174,7 @@
                 "method": {
                     "type": "keyword"
                 },
-                "defaultFill": {
+                "default_fill": {
                     "type": "nested",
                     "properties": {
                         "feature_name": {

diff --git a/src/test/java/org/opensearch/ad/e2e/AbstractMissingSingleFeatureTestCase.java b/src/test/java/org/opensearch/ad/e2e/AbstractMissingSingleFeatureTestCase.java
@@ -66,7 +66,7 @@ protected String genDetector(
             case FIXED_VALUES:
                 sb
                     .append(
-                        "\"imputation_option\": { \"method\": \"fixed_values\", \"defaultFill\": [{ \"feature_name\" : \"feature 1\", \"data\": 1 }] },"
+                        "\"imputation_option\": { \"method\": \"fixed_values\", \"default_fill\": [{ \"feature_name\" : \"feature 1\", \"data\": 1 }] },"
                     );
                 break;
         }

diff --git a/src/test/java/org/opensearch/ad/e2e/MissingMultiFeatureIT.java b/src/test/java/org/opensearch/ad/e2e/MissingMultiFeatureIT.java
@@ -218,7 +218,7 @@ protected String genDetector(
             case FIXED_VALUES:
                 sb
                     .append(
-                        "\"imputation_option\": { \"method\": \"fixed_values\", \"defaultFill\": [{ \"feature_name\" : \"feature 1\", \"data\": 1 }, { \"feature_name\" : \"feature 2\", \"data\": 2 }] },"
+                        "\"imputation_option\": { \"method\": \"fixed_values\", \"default_fill\": [{ \"feature_name\" : \"feature 1\", \"data\": 1 }, { \"feature_name\" : \"feature 2\", \"data\": 2 }] },"
                     );
                 break;
         }

diff --git a/src/test/java/org/opensearch/ad/indices/RolloverTests.java b/src/test/java/org/opensearch/ad/indices/RolloverTests.java
@@ -304,7 +304,7 @@ private void setUpGetConfigs_withNoCustomResultIndexAlias() {
             + "\"feature_attributes\":[{\"feature_id\":\"OTYJs\",\"feature_name\":\"eYYCM\",\"feature_enabled\":false,"
             + "\"aggregation_query\":{\"XzewX\":{\"value_count\":{\"field\":\"ok\"}}}}],\"recency_emphasis\":3342,"
             + "\"history\":62,\"last_update_time\":1717192049845,\"category_field\":[\"Tcqcb\"],\"customResultIndexOrAlias\":"
-            + "\"\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"defaultFill\""
+            + "\"\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"default_fill\""
             + ":[],\"integerSensitive\":false},\"suggested_seasonality\":64,\"detection_interval\":{\"period\":"
             + "{\"interval\":5,\"unit\":\"Minutes\"}},\"detector_type\":\"MULTI_ENTITY\",\"rules\":[]}";
 

diff --git a/src/test/java/org/opensearch/ad/model/AnomalyDetectorTests.java b/src/test/java/org/opensearch/ad/model/AnomalyDetectorTests.java
@@ -43,9 +43,12 @@
 import org.opensearch.timeseries.TestHelpers;
 import org.opensearch.timeseries.common.exception.ValidationException;
 import org.opensearch.timeseries.constant.CommonMessages;
+import org.opensearch.timeseries.dataprocessor.ImputationMethod;
+import org.opensearch.timeseries.dataprocessor.ImputationOption;
 import org.opensearch.timeseries.model.Config;
 import org.opensearch.timeseries.model.Feature;
 import org.opensearch.timeseries.model.IntervalTimeConfiguration;
+import org.opensearch.timeseries.model.ValidationIssueType;
 import org.opensearch.timeseries.settings.TimeSeriesSettings;
 
 import com.google.common.collect.ImmutableList;
@@ -894,7 +897,7 @@ public void testParseAnomalyDetector_withCustomIndex_withCustomResultIndexMinSiz
             + "\"feature_attributes\":[{\"feature_id\":\"OTYJs\",\"feature_name\":\"eYYCM\",\"feature_enabled\":true,"
             + "\"aggregation_query\":{\"XzewX\":{\"value_count\":{\"field\":\"ok\"}}}}],\"recency_emphasis\":3342,"
             + "\"history\":62,\"last_update_time\":1717192049845,\"category_field\":[\"Tcqcb\"],\"result_index\":"
-            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"defaultFill\""
+            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"default_fill\""
             + ":[{\"feature_name\":\"eYYCM\", \"data\": 3}]},\"suggested_seasonality\":64,\"detection_interval\":{\"period\":"
             + "{\"interval\":5,\"unit\":\"Minutes\"}},\"detector_type\":\"MULTI_ENTITY\",\"rules\":[],\"result_index_min_size\":1500}";
         AnomalyDetector parsedDetector = AnomalyDetector.parse(TestHelpers.parser(detectorString), "id", 1L, null, null);
@@ -921,7 +924,7 @@ public void testParseAnomalyDetector_withCustomIndex_withCustomResultIndexMinAge
             + "\"feature_attributes\":[{\"feature_id\":\"OTYJs\",\"feature_name\":\"eYYCM\",\"feature_enabled\":true,"
             + "\"aggregation_query\":{\"XzewX\":{\"value_count\":{\"field\":\"ok\"}}}}],\"recency_emphasis\":3342,"
             + "\"history\":62,\"last_update_time\":1717192049845,\"category_field\":[\"Tcqcb\"],\"result_index\":"
-            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"defaultFill\""
+            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"default_fill\""
             + ":[{\"feature_name\":\"eYYCM\", \"data\": 3}]},\"suggested_seasonality\":64,\"detection_interval\":{\"period\":"
             + "{\"interval\":5,\"unit\":\"Minutes\"}},\"detector_type\":\"MULTI_ENTITY\",\"rules\":[],\"result_index_min_age\":7}";
         AnomalyDetector parsedDetector = AnomalyDetector.parse(TestHelpers.parser(detectorString), "id", 1L, null, null);
@@ -936,7 +939,7 @@ public void testParseAnomalyDetector_withCustomIndex_withCustomResultIndexTTL()
             + "\"feature_attributes\":[{\"feature_id\":\"OTYJs\",\"feature_name\":\"eYYCM\",\"feature_enabled\":true,"
             + "\"aggregation_query\":{\"XzewX\":{\"value_count\":{\"field\":\"ok\"}}}}],\"recency_emphasis\":3342,"
             + "\"history\":62,\"last_update_time\":1717192049845,\"category_field\":[\"Tcqcb\"],\"result_index\":"
-            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"defaultFill\""
+            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"default_fill\""
             + ":[{\"feature_name\":\"eYYCM\", \"data\": 3}]},\"suggested_seasonality\":64,\"detection_interval\":{\"period\":"
             + "{\"interval\":5,\"unit\":\"Minutes\"}},\"detector_type\":\"MULTI_ENTITY\",\"rules\":[],\"result_index_ttl\":30}";
         AnomalyDetector parsedDetector = AnomalyDetector.parse(TestHelpers.parser(detectorString), "id", 1L, null, null);
@@ -951,7 +954,7 @@ public void testParseAnomalyDetector_withCustomIndex_withFlattenResultIndexMappi
             + "\"feature_attributes\":[{\"feature_id\":\"OTYJs\",\"feature_name\":\"eYYCM\",\"feature_enabled\":false,"
             + "\"aggregation_query\":{\"XzewX\":{\"value_count\":{\"field\":\"ok\"}}}}],\"recency_emphasis\":3342,"
             + "\"history\":62,\"last_update_time\":1717192049845,\"category_field\":[\"Tcqcb\"],\"result_index\":"
-            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"defaultFill\""
+            + "\"opensearch-ad-plugin-result-test\",\"imputation_option\":{\"method\":\"FIXED_VALUES\",\"default_fill\""
             + ":[],\"integerSensitive\":false},\"suggested_seasonality\":64,\"detection_interval\":{\"period\":"
             + "{\"interval\":5,\"unit\":\"Minutes\"}},\"detector_type\":\"MULTI_ENTITY\",\"rules\":[],\"flatten_result_index_mapping\":true}";
         AnomalyDetector parsedDetector = AnomalyDetector.parse(TestHelpers.parser(detectorString), "id", 1L, null, null);
@@ -992,4 +995,40 @@ public void testSerializeAndDeserializeAnomalyDetector() throws IOException {
         Assert.assertEquals(deserializedDetector, detector);
         Assert.assertEquals(deserializedDetector.getSeasonIntervals(), detector.getSeasonIntervals());
     }
+
+    public void testNullFixedValue() throws IOException {
+        org.opensearch.timeseries.common.exception.ValidationException e = assertThrows(
+            org.opensearch.timeseries.common.exception.ValidationException.class,
+            () -> new AnomalyDetector(
+                randomAlphaOfLength(5),
+                randomLong(),
+                randomAlphaOfLength(5),
+                randomAlphaOfLength(5),
+                randomAlphaOfLength(5),
+                ImmutableList.of(randomAlphaOfLength(5)),
+                ImmutableList.of(TestHelpers.randomFeature()),
+                TestHelpers.randomQuery(),
+                TestHelpers.randomIntervalTimeConfiguration(),
+                TestHelpers.randomIntervalTimeConfiguration(),
+                null,
+                null,
+                1,
+                Instant.now(),
+                null,
+                TestHelpers.randomUser(),
+                null,
+                new ImputationOption(ImputationMethod.FIXED_VALUES, null),
+                randomIntBetween(1, 10000),
+                randomInt(TimeSeriesSettings.MAX_SHINGLE_SIZE / 2),
+                randomIntBetween(1, 1000),
+                null,
+                null,
+                null,
+                null,
+                null
+            )
+        );
+        assertEquals("Got: " + e.getMessage(), "Enabled features are present, but no default fill values are provided.", e.getMessage());
+        assertEquals("Got :" + e.getType(), ValidationIssueType.IMPUTATION, e.getType());
+    }
 }
diff --git a/src/test/java/org/opensearch/timeseries/dataprocessor/ImputationOptionTests.java b/src/test/java/org/opensearch/timeseries/dataprocessor/ImputationOptionTests.java
@@ -41,7 +41,7 @@ public static void setUpOnce() {
 
         xContent = "{"
             + "\"method\":\"FIXED_VALUES\","
-            + "\"defaultFill\":[{\"feature_name\":\"a\", \"data\":1.0},{\"feature_name\":\"b\", \"data\":2.0},{\"feature_name\":\"c\", \"data\":3.0}]}";
+            + "\"default_fill\":[{\"feature_name\":\"a\", \"data\":1.0},{\"feature_name\":\"b\", \"data\":2.0},{\"feature_name\":\"c\", \"data\":3.0}]}";
     }
 
     private Map<String, Double> randomMap(double[] defaultFill) {