build flatten resuilt index enabled into the ResultWriteRequest

Signed-off-by: Jackie Han <hnyng@amazon.com>

Author: jackiehanyang

src/main/java/org/opensearch/ad/ratelimit/ADResultWriteRequest.java

@@ -25,9 +25,10 @@ public ADResultWriteRequest(
         String detectorId,
         RequestPriority priority,
         AnomalyResult result,
-        String resultIndex
+        String resultIndex,
+        boolean flattenResultIndex
     ) {
-        super(expirationEpochMs, detectorId, priority, result, resultIndex);
+        super(expirationEpochMs, detectorId, priority, result, resultIndex, flattenResultIndex);
     }
 
     public ADResultWriteRequest(StreamInput in) throws IOException {

src/main/java/org/opensearch/ad/ratelimit/ADResultWriteWorker.java

@@ -103,8 +103,9 @@ protected ADResultWriteRequest createResultWriteRequest(
         String configId,
         RequestPriority priority,
         AnomalyResult result,
-        String resultIndex
+        String resultIndex,
+        boolean flattenResultIndex
     ) {
-        return new ADResultWriteRequest(expirationEpochMs, configId, priority, result, resultIndex);
+        return new ADResultWriteRequest(expirationEpochMs, configId, priority, result, resultIndex, flattenResultIndex);
     }
 }

src/main/java/org/opensearch/ad/ratelimit/ADSaveResultStrategy.java

@@ -86,7 +86,8 @@ public void saveResult(AnomalyResult result, Config config) {
                     config.getId(),
                     result.getAnomalyGrade() > 0 ? RequestPriority.HIGH : RequestPriority.MEDIUM,
                     result,
-                    config.getCustomResultIndexOrAlias()
+                    config.getCustomResultIndexOrAlias(),
+                    config.getFlattenResultIndexMapping()
                 )
             );
     }

src/main/java/org/opensearch/ad/transport/ADResultBulkTransportAction.java

@@ -120,7 +120,10 @@ protected BulkRequest prepareBulkRequest(float indexingPressurePercent, ADResult
 
             if (shouldAddResult(indexingPressurePercent, result)) {
                 addResult(bulkRequest, result, resultIndex);
-                addToFlattenedIndexIfExists(bulkRequest, result, resultIndex);
+                if (resultWriteRequest.getFlattenResultIndex()) {
+                    String flattenedResultIndexAlias = resultIndex + "_flattened_" + result.getDetectorId().toLowerCase(Locale.ROOT);
+                    addResult(bulkRequest, result, flattenedResultIndexAlias);
+                }
             }
         }
 
@@ -141,20 +144,6 @@ private boolean shouldAddResult(float indexingPressurePercent, AnomalyResult res
         }
     }
 
-    private void addToFlattenedIndexIfExists(BulkRequest bulkRequest, AnomalyResult result, String resultIndex) {
-        String flattenedResultIndexAlias = resultIndex + "_flattened_" + result.getDetectorId().toLowerCase(Locale.ROOT);
-        String configId = result.getConfigId();
-        nodeStateManager.getConfig(configId, AnalysisType.AD, ActionListener.wrap(configOptional -> {
-            if (configOptional.isEmpty()) {
-                return;
-            }
-            Config config = configOptional.get();
-            if (config.getFlattenResultIndexMapping()) {
-                addResult(bulkRequest, result, flattenedResultIndexAlias);
-            }
-        }, e -> LOG.error("Fail to get config", e)));
-    }
-
     private void addResult(BulkRequest bulkRequest, AnomalyResult result, String resultIndex) {
         String index = resultIndex == null ? indexName : resultIndex;
         try (XContentBuilder builder = jsonBuilder()) {

src/main/java/org/opensearch/ad/transport/ADSingleStreamResultTransportAction.java

@@ -72,7 +72,8 @@ public ADResultWriteRequest createResultWriteRequest(Config config, AnomalyResul
             config.getId(),
             RequestPriority.MEDIUM,
             result,
-            config.getCustomResultIndexOrAlias()
+            config.getCustomResultIndexOrAlias(),
+            config.getFlattenResultIndexMapping()
         );
     }
 

src/main/java/org/opensearch/forecast/ratelimit/ForecastResultWriteRequest.java

@@ -25,9 +25,10 @@ public ForecastResultWriteRequest(
         String forecasterId,
         RequestPriority priority,
         ForecastResult result,
-        String resultIndex
+        String resultIndex,
+        boolean flattenResultIndex
     ) {
-        super(expirationEpochMs, forecasterId, priority, result, resultIndex);
+        super(expirationEpochMs, forecasterId, priority, result, resultIndex, flattenResultIndex);
     }
 
     public ForecastResultWriteRequest(StreamInput in) throws IOException {

src/main/java/org/opensearch/forecast/ratelimit/ForecastResultWriteWorker.java

@@ -103,8 +103,9 @@ protected ForecastResultWriteRequest createResultWriteRequest(
         String configId,
         RequestPriority priority,
         ForecastResult result,
-        String resultIndex
+        String resultIndex,
+        boolean flattenResultIndex
     ) {
-        return new ForecastResultWriteRequest(expirationEpochMs, configId, priority, result, resultIndex);
+        return new ForecastResultWriteRequest(expirationEpochMs, configId, priority, result, resultIndex, flattenResultIndex);
     }
 }

src/main/java/org/opensearch/forecast/ratelimit/ForecastSaveResultStrategy.java

@@ -83,7 +83,8 @@ public void saveResult(ForecastResult result, Config config) {
                     config.getId(),
                     RequestPriority.MEDIUM,
                     result,
-                    config.getCustomResultIndexOrAlias()
+                    config.getCustomResultIndexOrAlias(),
+                    config.getFlattenResultIndexMapping()
                 )
             );
     }

src/main/java/org/opensearch/forecast/transport/ForecastSingleStreamResultTransportAction.java

@@ -76,7 +76,8 @@ public ForecastResultWriteRequest createResultWriteRequest(Config config, Foreca
             config.getId(),
             RequestPriority.MEDIUM,
             result,
-            config.getCustomResultIndexOrAlias()
+            config.getCustomResultIndexOrAlias(),
+            config.getFlattenResultIndexMapping()
         );
     }
 }

src/main/java/org/opensearch/timeseries/ratelimit/ResultWriteRequest.java

@@ -22,16 +22,19 @@ public abstract class ResultWriteRequest<ResultType extends IndexableResult> ext
     private final ResultType result;
     // If resultIndex is null, result will be stored in default result index.
     private final String resultIndex;
+    private final boolean flattenResultIndex;
 
-    public ResultWriteRequest(long expirationEpochMs, String configId, RequestPriority priority, ResultType result, String resultIndex) {
+    public ResultWriteRequest(long expirationEpochMs, String configId, RequestPriority priority, ResultType result, String resultIndex, boolean flattenResultIndex) {
         super(expirationEpochMs, configId, priority);
         this.result = result;
         this.resultIndex = resultIndex;
+        this.flattenResultIndex = flattenResultIndex;
     }
 
-    public ResultWriteRequest(StreamInput in, Writeable.Reader<ResultType> resultReader) throws IOException {
+    public ResultWriteRequest(StreamInput in, Reader<ResultType> resultReader) throws IOException {
         this.result = resultReader.read(in);
         this.resultIndex = in.readOptionalString();
+        this.flattenResultIndex = in.readBoolean();
     }
 
     @Override
@@ -47,4 +50,8 @@ public ResultType getResult() {
     public String getResultIndex() {
         return resultIndex;
     }
+
+    public boolean getFlattenResultIndex() {
+        return flattenResultIndex;
+    }
 }

src/main/java/org/opensearch/timeseries/ratelimit/ResultWriteWorker.java

@@ -199,7 +199,8 @@ private ActionListener<Optional<? extends Config>> onGetConfig(
                     id,
                     resultToRetry.isHighPriority() ? RequestPriority.HIGH : RequestPriority.MEDIUM,
                     resultToRetry,
-                    config.getCustomResultIndexOrAlias()
+                    config.getCustomResultIndexOrAlias(),
+                    config.getFlattenResultIndexMapping()
                 )
             );
 
@@ -216,6 +217,7 @@ protected abstract ResultWriteRequestType createResultWriteRequest(
         String configId,
         RequestPriority priority,
         ResultType result,
-        String resultIndex
+        String resultIndex,
+        boolean flattenResultIndex
     );
 }