GetMappingsView API - index pattern/alias/datastream support

src/main/java/org/opensearch/securityanalytics/SecurityAnalyticsPlugin.java

@@ -121,7 +121,7 @@ public Collection<Object> createComponents(Client client,
                                                Supplier<RepositoriesService> repositoriesServiceSupplier) {
         detectorIndices = new DetectorIndices(client.admin(), clusterService, threadPool);
         ruleTopicIndices = new RuleTopicIndices(client, clusterService);
-        mapperService = new MapperService(client.admin().indices());
+        mapperService = new MapperService(client.admin().indices(), clusterService);
         ruleIndices = new RuleIndices(client, clusterService, threadPool);
         return List.of(detectorIndices, ruleTopicIndices, ruleIndices, mapperService);
     }

src/main/java/org/opensearch/securityanalytics/mapper/MapperService.java

@@ -5,19 +5,24 @@
 
 package org.opensearch.securityanalytics.mapper;
 
-import java.util.Locale;
+import java.util.Collection;
+import java.util.Optional;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.OpenSearchStatusException;
-import org.opensearch.ResourceNotFoundException;
 import org.opensearch.action.ActionListener;
+import org.opensearch.action.admin.indices.get.GetIndexRequest;
+import org.opensearch.action.admin.indices.get.GetIndexResponse;
 import org.opensearch.action.admin.indices.mapping.get.GetMappingsRequest;
 import org.opensearch.action.admin.indices.mapping.get.GetMappingsResponse;
 import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest;
+import org.opensearch.action.support.GroupedActionListener;
 import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.client.IndicesAdminClient;
+import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.metadata.MappingMetadata;
+import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.collect.ImmutableOpenMap;
 import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.rest.RestStatus;
@@ -30,6 +35,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.opensearch.securityanalytics.action.GetMappingsViewResponse;
+import org.opensearch.securityanalytics.util.IndexUtils;
 import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
 
 
@@ -39,13 +45,15 @@
 public class MapperService {
 
     private static final Logger log = LogManager.getLogger(MapperService.class);
+    private ClusterService clusterService;
 
     IndicesAdminClient indicesClient;
 
     public MapperService() {}
 
-    public MapperService(IndicesAdminClient indicesClient) {
+    public MapperService(IndicesAdminClient indicesClient, ClusterService clusterService) {
         this.indicesClient = indicesClient;
+        this.clusterService = clusterService;
     }
 
     void setIndicesAdminClient(IndicesAdminClient client) {
@@ -74,10 +82,52 @@ public void onFailure(Exception e) {
 
     private void createMappingActionContinuation(ImmutableOpenMap<String, MappingMetadata> indexMappings, String ruleTopic, String aliasMappings, boolean partial, ActionListener<AcknowledgedResponse> actionListener) {
 
+        int numOfIndices =  indexMappings.size();
+
+        GroupedActionListener doCreateMappingActionsListener = new GroupedActionListener(new ActionListener<Collection<AcknowledgedResponse>>() {
+            @Override
+            public void onResponse(Collection<AcknowledgedResponse> response) {
+                // We will return ack==false if one of the requests returned that
+                // else return ack==true
+                Optional<AcknowledgedResponse> notAckd = response.stream().filter(e -> e.isAcknowledged() == false).findFirst();
+                AcknowledgedResponse ack = new AcknowledgedResponse(
+                        notAckd.isPresent() ? false : true
+                );
+                actionListener.onResponse(ack);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                actionListener.onFailure(
+                    new SecurityAnalyticsException(
+                        "Failed applying mappings to index", RestStatus.INTERNAL_SERVER_ERROR, e
+                    )
+                );
+            }
+        }, numOfIndices);
+
+        indexMappings.forEach(iter -> {
+            String indexName = iter.key;
+            MappingMetadata mappingMetadata = iter.value;
+            // Try to apply mapping to index
+            doCreateMapping(indexName, mappingMetadata, ruleTopic, aliasMappings, partial, doCreateMappingActionsListener);
+        });
+    }
+
+    /**
+     * Applies alias mappings to index.
+     * @param indexName Index name
+     * @param mappingMetadata Index mappings
+     * @param ruleTopic Rule topic spcifying specific alias templates
+     * @param aliasMappings User-supplied alias mappings
+     * @param partial Partial flag indicating if we should apply mappings partially, in case source index doesn't have all paths specified in alias mappings
+     * @param actionListener actionListener used to return response/error
+     */
+    private void doCreateMapping(String indexName, MappingMetadata mappingMetadata, String ruleTopic, String aliasMappings, boolean partial, ActionListener<AcknowledgedResponse> actionListener) {
+
         PutMappingRequest request;
         try {
 
-            String indexName = indexMappings.iterator().next().key;
             String aliasMappingsJSON;
             // aliasMappings parameter has higher priority then ruleTopic
             if (aliasMappings != null) {
@@ -86,7 +136,7 @@ private void createMappingActionContinuation(ImmutableOpenMap<String, MappingMet
                 aliasMappingsJSON = MapperTopicStore.aliasMappings(ruleTopic);
             }
 
-            List<String> missingPathsInIndex = MapperUtils.validateIndexMappings(indexMappings, aliasMappingsJSON);
+            List<String> missingPathsInIndex = MapperUtils.validateIndexMappings(indexName, mappingMetadata, aliasMappingsJSON);
 
             if(missingPathsInIndex.size() > 0) {
                 // If user didn't allow partial apply, we should error out here
@@ -238,7 +288,34 @@ public void getMappingsViewAction(
             String mapperTopic,
             ActionListener<GetMappingsViewResponse> actionListener
     ) {
-        GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(indexName);
+        try {
+            // We are returning mappings view for only 1 index: writeIndex or latest from the pattern
+            resolveConcreteIndex(indexName, new ActionListener<>() {
+                @Override
+                public void onResponse(String concreteIndex) {
+                    doGetMappingsView(mapperTopic, actionListener, concreteIndex);
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    actionListener.onFailure(e);
+                }
+            });
+
+
+        } catch (IOException e) {
+            throw SecurityAnalyticsException.wrap(e);
+        }
+    }
+
+    /**
+     * Constructs Mappings View of index
+     * @param mapperTopic Mapper Topic describing set of alias mappings
+     * @param actionListener Action Listener
+     * @param concreteIndex Concrete Index name for which we're computing Mappings View
+     */
+    private void doGetMappingsView(String mapperTopic, ActionListener<GetMappingsViewResponse> actionListener, String concreteIndex) {
+        GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(concreteIndex);
         indicesClient.getMappings(getMappingsRequest, new ActionListener<>() {
             @Override
             public void onResponse(GetMappingsResponse getMappingsResponse) {
@@ -291,4 +368,45 @@ public void onFailure(Exception e) {
             }
         });
     }
+
+    /**
+     * Given index name, resolves it to single concrete index, depending on what initial <code>indexName</code> is.
+     * In case of Datastream or Alias, WriteIndex would be returned. In case of index pattern, newest index by creation date would be returned.
+     * @param indexName Datastream, Alias, index patter or concrete index
+     * @param actionListener Action Listener
+     * @throws IOException
+     */
+    private void resolveConcreteIndex(String indexName, ActionListener<String> actionListener) throws IOException {
+
+        indicesClient.getIndex((new GetIndexRequest()).indices(indexName), new ActionListener<>() {
+            @Override
+            public void onResponse(GetIndexResponse getIndexResponse) {
+                String[] indices = getIndexResponse.indices();
+                if (indices.length == 0) {
+                    actionListener.onFailure(
+                            SecurityAnalyticsException.wrap(
+                                    new IllegalArgumentException("Invalid index name: [" + indexName + "]")
+                            )
+                    );
+                } else if (indices.length == 1) {
+                    actionListener.onResponse(indices[0]);
+                } else if (indices.length > 1) {
+                    String writeIndex = IndexUtils.getWriteIndex(indexName, MapperService.this.clusterService.state());
+                    if (writeIndex != null) {
+                        actionListener.onResponse(writeIndex);
+                    } else {
+                        actionListener.onResponse(
+                            IndexUtils.getNewestIndexByCreationDate(indices, MapperService.this.clusterService.state())
+                        );
+                    }
+                }
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                actionListener.onFailure(e);
+            }
+        });
+
+    }
 }

src/main/java/org/opensearch/securityanalytics/mapper/MapperUtils.java

@@ -6,6 +6,7 @@
 package org.opensearch.securityanalytics.mapper;
 
 import java.util.HashMap;
+import java.util.Locale;
 import java.util.Map;
 import org.apache.commons.lang3.tuple.Pair;
 import org.opensearch.cluster.metadata.MappingMetadata;
@@ -89,24 +90,22 @@ public void onError(String error) {
      *   <li>Alias mappings have to have property type=alias and path property has to exist
      *   <li>Paths from alias mappings should exists in index mappings
      * </ul>
-     * @param indexMappings Index Mappings to which alias mappings will be applied
-     * @param aliasMappingsJSON Alias Mappings as JSON string
+     * @param indexName Source index name
+     * @param mappingMetadata Source index mapping to which alias mappings will be applied
+     * @param aliasMappingsJSON Alias mappings as JSON string
      * @return list of alias mappings paths which are missing in index mappings
      * */
-    public static List<String> validateIndexMappings(ImmutableOpenMap<String, MappingMetadata> indexMappings, String aliasMappingsJSON) throws IOException {
+    public static List<String> validateIndexMappings(String indexName, MappingMetadata mappingMetadata, String aliasMappingsJSON) throws IOException {
 
         // Check if index's mapping is empty
-        if (isIndexMappingsEmpty(indexMappings)) {
-            throw new IllegalArgumentException("Index mappings are empty");
+        if (isIndexMappingsEmpty(mappingMetadata)) {
+            throw new IllegalArgumentException(String.format(Locale.ROOT, "Mappings for index [%s] are empty", indexName));
         }
 
         // Get all paths (field names) to which we're going to apply aliases
         List<String> paths = getAllPathsFromAliasMappings(aliasMappingsJSON);
 
         // Traverse Index Mappings and extract all fields(paths)
-        String indexName = indexMappings.iterator().next().key;
-        MappingMetadata mappingMetadata = indexMappings.get(indexName);
-
         List<String> flatFields = getAllNonAliasFieldsFromIndex(mappingMetadata);
         // Return list of paths from Alias Mappings which are missing in Index Mappings
         return paths.stream()
@@ -164,11 +163,8 @@ public static List<String> getAllNonAliasFieldsFromIndex(MappingMetadata mapping
         return mappingsTraverser.extractFlatNonAliasFields();
     }
 
-    public static boolean isIndexMappingsEmpty(ImmutableOpenMap<String, MappingMetadata> indexMappings) {
-        if (indexMappings.iterator().hasNext()) {
-            return indexMappings.iterator().next().value.getSourceAsMap().size() == 0;
-        }
-        throw new IllegalArgumentException("Invalid Index Mappings");
+    public static boolean isIndexMappingsEmpty(MappingMetadata mappingMetadata) {
+        return mappingMetadata.getSourceAsMap().size() == 0;
     }
 
     public static Map<String, Object> getAliasMappingsWithFilter(

src/main/java/org/opensearch/securityanalytics/transport/TransportCreateIndexMappingsAction.java

@@ -43,11 +43,6 @@ public TransportCreateIndexMappingsAction(
     protected void doExecute(Task task, CreateIndexMappingsRequest request, ActionListener<AcknowledgedResponse> actionListener) {
         this.threadPool.getThreadContext().stashContext();
 
-        IndexMetadata index = clusterService.state().metadata().index(request.getIndexName());
-        if (index == null) {
-            actionListener.onFailure(new IllegalStateException("Could not find index [" + request.getIndexName() + "]"));
-            return;
-        }
         mapperService.createMappingAction(
                 request.getIndexName(),
                 request.getRuleTopic(),

src/main/java/org/opensearch/securityanalytics/transport/TransportGetMappingsViewAction.java

@@ -47,17 +47,6 @@ public TransportGetMappingsViewAction(
     @Override
     protected void doExecute(Task task, GetMappingsViewRequest request, ActionListener<GetMappingsViewResponse> actionListener) {
         this.threadPool.getThreadContext().stashContext();
-        IndexMetadata index = clusterService.state().metadata().index(request.getIndexName());
-        if (index == null) {
-            actionListener.onFailure(
-                    SecurityAnalyticsException.wrap(
-                            new OpenSearchStatusException(
-                                    "Could not find index [" + request.getIndexName() + "]", RestStatus.NOT_FOUND
-                            )
-                    )
-            );
-            return;
-        }
-        mapperService.getMappingsViewAction(request.getIndexName(), request.getRuleTopic(), actionListener);
+        this.mapperService.getMappingsViewAction(request.getIndexName(), request.getRuleTopic(), actionListener);
     }
 }

src/main/java/org/opensearch/securityanalytics/util/IndexUtils.java

@@ -4,11 +4,13 @@
  */
 package org.opensearch.securityanalytics.util;
 
+import java.util.SortedMap;
 import org.opensearch.action.ActionListener;
 import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest;
 import org.opensearch.action.support.master.AcknowledgedResponse;
 import org.opensearch.client.IndicesAdminClient;
 import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.metadata.IndexAbstraction;
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.xcontent.LoggingDeprecationHandler;
 import org.opensearch.common.xcontent.NamedXContentRegistry;
@@ -105,4 +107,39 @@ public static void updateIndexMapping(
             }
         }
     }
+
+    public static boolean isDataStream(String name, ClusterState clusterState) {
+        return clusterState.getMetadata().dataStreams().containsKey(name);
+    }
+    public static boolean isAlias(String indexName, ClusterState clusterState) {
+        return clusterState.getMetadata().hasAlias(indexName);
+    }
+    public static String getWriteIndex(String indexName, ClusterState clusterState) {
+        if(isAlias(indexName, clusterState) || isDataStream(indexName, clusterState)) {
+            IndexMetadata metadata = clusterState.getMetadata()
+                    .getIndicesLookup()
+                    .get(indexName).getWriteIndex();
+            if (metadata != null) {
+                return metadata.getIndex().getName();
+            }
+        }
+        return null;
+    }
+
+    public static String getNewestIndexByCreationDate(String[] concreteIndices, ClusterState clusterState) {
+        final SortedMap<String, IndexAbstraction> lookup = clusterState.getMetadata().getIndicesLookup();
+        long maxCreationDate = Long.MIN_VALUE;
+        String newestIndex = null;
+        for (String indexName : concreteIndices) {
+            IndexAbstraction index = lookup.get(indexName);
+            IndexMetadata indexMetadata = clusterState.getMetadata().index(indexName);
+            if(index != null && index.getType() == IndexAbstraction.Type.CONCRETE_INDEX) {
+                if (indexMetadata.getCreationDate() > maxCreationDate) {
+                    maxCreationDate = indexMetadata.getCreationDate();
+                    newestIndex = indexName;
+                }
+            }
+        }
+        return newestIndex;
+    }
 }