utlizing a node state manager when writing results into flattened result index

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

Author: jackiehanyang

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

@@ -31,8 +31,12 @@
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.index.IndexingPressure;
+import org.opensearch.timeseries.AnalysisType;
+import org.opensearch.timeseries.NodeStateManager;
+import org.opensearch.timeseries.model.Config;
 import org.opensearch.timeseries.transport.ResultBulkTransportAction;
 import org.opensearch.timeseries.util.RestHandlerUtils;
 import org.opensearch.transport.TransportService;
@@ -50,7 +54,8 @@ public ADResultBulkTransportAction(
         IndexingPressure indexingPressure,
         Settings settings,
         ClusterService clusterService,
-        Client client
+        Client client,
+        NodeStateManager stateManager
     ) {
         super(
             ADResultBulkAction.NAME,
@@ -62,7 +67,8 @@ public ADResultBulkTransportAction(
             AD_INDEX_PRESSURE_SOFT_LIMIT.get(settings),
             AD_INDEX_PRESSURE_HARD_LIMIT.get(settings),
             ADCommonName.ANOMALY_RESULT_INDEX_ALIAS,
-            ADResultBulkRequest::new
+            ADResultBulkRequest::new,
+            stateManager
         );
         this.clusterService = clusterService;
         this.client = client;
@@ -137,9 +143,16 @@ 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);
-        if (clusterService.state().metadata().hasAlias(flattenedResultIndexAlias)) {
-            addResult(bulkRequest, result, flattenedResultIndexAlias);
-        }
+        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) {

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

@@ -26,6 +26,7 @@
 import org.opensearch.forecast.model.ForecastResult;
 import org.opensearch.forecast.ratelimit.ForecastResultWriteRequest;
 import org.opensearch.index.IndexingPressure;
+import org.opensearch.timeseries.NodeStateManager;
 import org.opensearch.timeseries.transport.ResultBulkTransportAction;
 import org.opensearch.transport.TransportService;
 
@@ -39,7 +40,8 @@ public ForecastResultBulkTransportAction(
         IndexingPressure indexingPressure,
         Settings settings,
         ClusterService clusterService,
-        Client client
+        Client client,
+        NodeStateManager stateManager
     ) {
         super(
             ForecastResultBulkAction.NAME,
@@ -51,7 +53,8 @@ public ForecastResultBulkTransportAction(
             FORECAST_INDEX_PRESSURE_SOFT_LIMIT.get(settings),
             FORECAST_INDEX_PRESSURE_HARD_LIMIT.get(settings),
             ForecastIndex.RESULT.getIndexName(),
-            ForecastResultBulkRequest::new
+            ForecastResultBulkRequest::new,
+            stateManager
         );
         clusterService.getClusterSettings().addSettingsUpdateConsumer(FORECAST_INDEX_PRESSURE_SOFT_LIMIT, it -> softLimit = it);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(FORECAST_INDEX_PRESSURE_HARD_LIMIT, it -> hardLimit = it);

src/main/java/org/opensearch/timeseries/transport/ResultBulkTransportAction.java

@@ -34,6 +34,7 @@
 import org.opensearch.index.IndexingPressure;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.timeseries.NodeStateManager;
 import org.opensearch.timeseries.model.IndexableResult;
 import org.opensearch.timeseries.ratelimit.ResultWriteRequest;
 import org.opensearch.timeseries.util.BulkUtil;
@@ -51,6 +52,7 @@ public abstract class ResultBulkTransportAction<ResultType extends IndexableResu
     protected String indexName;
     private Client client;
     protected Random random;
+    protected NodeStateManager nodeStateManager;
 
     public ResultBulkTransportAction(
         String actionName,
@@ -62,7 +64,8 @@ public ResultBulkTransportAction(
         float softLimit,
         float hardLimit,
         String indexName,
-        Writeable.Reader<ResultBulkRequestType> requestReader
+        Writeable.Reader<ResultBulkRequestType> requestReader,
+        NodeStateManager nodeStateManager
     ) {
         super(actionName, transportService, actionFilters, requestReader, ThreadPool.Names.SAME);
         this.indexingPressure = indexingPressure;
@@ -72,6 +75,7 @@ public ResultBulkTransportAction(
         this.softLimit = softLimit;
         this.hardLimit = hardLimit;
         this.indexName = indexName;
+        this.nodeStateManager = nodeStateManager;
 
         // random seed is 42. Can be any number
         this.random = new Random(42);

src/test/java/org/opensearch/ad/rest/AnomalyDetectorRestApiIT.java

@@ -241,7 +241,7 @@ public void testCreateAnomalyDetector_withFlattenedResultIndex() throws Exceptio
             .format(Locale.ROOT, "/opensearch-ad-plugin-result-test_flattened_%s", id.toLowerCase(Locale.ROOT));
         // wait for the detector starts writing result
         try {
-            Thread.sleep(60 * 1000);
+            Thread.sleep(30 * 1000);
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
             throw new RuntimeException("Thread was interrupted while waiting", e);