address comments

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

Author: jackiehanyang

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

@@ -22,6 +22,7 @@
 
 import java.io.IOException;
 import java.util.EnumMap;
+import java.util.Map;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -45,6 +46,8 @@
 import org.opensearch.timeseries.indices.IndexManagement;
 import org.opensearch.timeseries.util.DiscoveryNodeFilterer;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 /**
  * This class provides utility methods for various anomaly detection indices.
  */
@@ -57,6 +60,8 @@ public class ADIndexManagement extends IndexManagement<ADIndex> {
     // The index name pattern to query all AD result, history and current AD result
     public static final String ALL_AD_RESULTS_INDEX_PATTERN = ".opendistro-anomaly-results*";
 
+    // private static final ObjectMapper objectMapper = new ObjectMapper();
+
     /**
      * Constructor function
      *
@@ -122,6 +127,22 @@ public static String getResultMappings() throws IOException {
         return getMappings(ANOMALY_RESULTS_INDEX_MAPPING_FILE);
     }
 
+    /**
+     * Retrieves the JSON mapping for the flattened result index with the "dynamic" field set to true
+     * @return JSON mapping for the flattened result index.
+     * @throws IOException if the mapping file cannot be read.
+     */
+    public static String getFlattenedResultMappings() throws IOException {
+        ObjectMapper objectMapper = new ObjectMapper();
+
+        Map<String, Object> mapping = objectMapper
+            .readValue(ADIndexManagement.class.getClassLoader().getResourceAsStream(ANOMALY_RESULTS_INDEX_MAPPING_FILE), Map.class);
+
+        mapping.put("dynamic", true);
+
+        return objectMapper.writeValueAsString(mapping);
+    }
+
     /**
      * Get anomaly detector state index mapping json content.
      *

src/main/java/org/opensearch/timeseries/indices/IndexManagement.java

@@ -11,7 +11,7 @@
 
 package org.opensearch.timeseries.indices;
 
-import static org.opensearch.ad.settings.AnomalyDetectorSettings.FLATTENED_ANOMALY_RESULTS_INDEX_MAPPING_FILE;
+import static org.opensearch.ad.indices.ADIndexManagement.getFlattenedResultMappings;
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.opensearch.timeseries.util.RestHandlerUtils.createXContentParserFromRegistry;
 
@@ -90,6 +90,7 @@
 import org.opensearch.timeseries.settings.TimeSeriesSettings;
 import org.opensearch.timeseries.util.DiscoveryNodeFilterer;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Charsets;
 import com.google.common.io.Resources;
 
@@ -137,6 +138,7 @@ public abstract class IndexManagement<IndexType extends Enum<IndexType> & TimeSe
     private NamedXContentRegistry xContentRegistry;
     protected BiCheckedFunction<XContentParser, String, ? extends Config, IOException> configParser;
     protected String customResultIndexPrefix;
+    private final ObjectMapper objectMapper = new ObjectMapper();
 
     protected class IndexState {
         // keep track of whether the mapping version is up-to-date
@@ -1016,21 +1018,28 @@ public <T> void initCustomResultIndexAndExecute(String resultIndexOrAlias, Execu
 
     /**
      * creates flattened result index
-     * @param indexName the index name
+     * @param flattenedResultIndexAlias the flattened result index alias
      * @param actionListener the action listener
      * @throws IOException
      */
-    public void initFlattenedResultIndex(String indexName, ActionListener<CreateIndexResponse> actionListener) throws IOException {
+    public void initFlattenedResultIndex(String flattenedResultIndexAlias, ActionListener<CreateIndexResponse> actionListener)
+        throws IOException {
+        String indexName = getCustomResultIndexPattern(flattenedResultIndexAlias);
         logger.info("Initializing flattened result index: {}", indexName);
 
         CreateIndexRequest request = new CreateIndexRequest(indexName)
-            .mapping(getFlattenedResultIndexMappings(), XContentType.JSON)
+            .mapping(getFlattenedResultMappings(), XContentType.JSON)
             .settings(settings);
+
+        if (flattenedResultIndexAlias != null) {
+            request.alias(new Alias(flattenedResultIndexAlias));
+        }
+
         choosePrimaryShards(request, false);
 
         adminClient.indices().create(request, ActionListener.wrap(response -> {
             if (response.isAcknowledged()) {
-                logger.info("Successfully created flattened result index: {}", indexName);
+                logger.info("Successfully created flattened result index: {} with alias: {}", indexName, flattenedResultIndexAlias);
                 actionListener.onResponse(response);
             } else {
                 String errorMsg = "Index creation not acknowledged for index: " + indexName;
@@ -1043,15 +1052,6 @@ public void initFlattenedResultIndex(String indexName, ActionListener<CreateInde
         }));
     }
 
-    /**
-     * Get flattened result index mapping json content
-     * @return flattened result index mapping
-     * @throws IOException
-     */
-    public String getFlattenedResultIndexMappings() throws IOException {
-        return getMappings(FLATTENED_ANOMALY_RESULTS_INDEX_MAPPING_FILE);
-    }
-
     public <T> void validateCustomIndexForBackendJob(
         String resultIndexOrAlias,
         String securityLogId,

src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java

@@ -454,6 +454,14 @@ protected void prepareConfigIndexing(boolean indexingDryRun, ActionListener<T> l
         }
     }
 
+    private String getFlattenedResultIndexAlias(String configId) {
+        return config.getCustomResultIndexOrAlias() + "_flattened_" + configId.toLowerCase(Locale.ROOT);
+    }
+
+    private String getFlattenResultIndexIngestPipelineId(String configId) {
+        return "flatten_result_index_ingest_pipeline" + configId.toLowerCase(Locale.ROOT);
+    }
+
     private void handlePutRequest(boolean indexingDryRun, ActionListener<T> listener) {
         handler.confirmJobRunning(clusterService, client, id, listener, () -> {
             handleFlattenResultIndexMappingUpdate(listener);
@@ -465,17 +473,17 @@ private void handlePostRequest(boolean indexingDryRun, ActionListener<T> listene
         createConfig(indexingDryRun, ActionListener.wrap(createConfigResponse -> {
             if (shouldHandleFlattening(indexingDryRun, createConfigResponse)) {
                 IndexAnomalyDetectorResponse response = (IndexAnomalyDetectorResponse) createConfigResponse;
-                String detectorId = response.getId();
-                String indexName = config.getCustomResultIndexOrAlias() + "_flattened_" + detectorId.toLowerCase(Locale.ROOT);
-                String pipelineId = "anomaly_detection_ingest_pipeline_" + detectorId.toLowerCase(Locale.ROOT);
+                String configId = response.getId();
+                String flattenedResultIndexAlias = getFlattenedResultIndexAlias(configId);
+                String pipelineId = getFlattenResultIndexIngestPipelineId(configId);
 
                 timeSeriesIndices
                     .initFlattenedResultIndex(
-                        indexName,
-                        ActionListener.wrap(initResponse -> setupIngestPipeline(detectorId, ActionListener.wrap(pipelineResponse -> {
+                        flattenedResultIndexAlias,
+                        ActionListener.wrap(initResponse -> setupIngestPipeline(configId, ActionListener.wrap(pipelineResponse -> {
                             updateResultIndexSetting(
                                 pipelineId,
-                                indexName,
+                                flattenedResultIndexAlias,
                                 ActionListener.wrap(updateResponse -> listener.onResponse(createConfigResponse), listener::onFailure)
                             );
                         }, listener::onFailure)), listener::onFailure)
@@ -493,12 +501,12 @@ private boolean shouldHandleFlattening(boolean indexingDryRun, Object createConf
             && createConfigResponse instanceof IndexAnomalyDetectorResponse;
     }
 
-    protected void setupIngestPipeline(String detectorId, ActionListener<T> listener) {
-        String indexName = config.getCustomResultIndexOrAlias() + "_flattened_" + detectorId.toLowerCase(Locale.ROOT);
-        String pipelineId = "anomaly_detection_ingest_pipeline_" + detectorId.toLowerCase(Locale.ROOT);
+    protected void setupIngestPipeline(String configId, ActionListener<T> listener) {
+        String flattenedResultIndexAlias = getFlattenedResultIndexAlias(configId);
+        String pipelineId = getFlattenResultIndexIngestPipelineId(configId);
 
         try {
-            BytesReference pipelineSource = createPipelineDefinition(indexName);
+            BytesReference pipelineSource = createPipelineDefinition(flattenedResultIndexAlias);
 
             PutPipelineRequest putPipelineRequest = new PutPipelineRequest(pipelineId, pipelineSource, XContentType.JSON);
 
@@ -576,10 +584,9 @@ private void handleFlattenResultIndexMappingUpdate(ActionListener<T> listener) {
             return;
         }
         if (config.getFlattenResultIndexMapping() != null && config.getFlattenResultIndexMapping()) {
-            // if field value is true, create the pipeline. No need to get and compare with previous value
             setupIngestPipeline(id, listener);
         } else {
-            String pipelineId = "anomaly_detection_ingest_pipeline_" + config.getId();
+            String pipelineId = getFlattenResultIndexIngestPipelineId(config.getId());
             client.admin().cluster().deletePipeline(new DeletePipelineRequest(pipelineId), new ActionListener<AcknowledgedResponse>() {
 
                 @Override