Adding remote index and multi-index checks in validation (#1290) (#1294)

* Adding remote index and multi index checks in validation



* adding more tests



---------


(cherry picked from commit ec16d53)

Signed-off-by: Amit Galitzky <amgalitz@amazon.com>
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

src/main/java/org/opensearch/timeseries/constant/CommonMessages.java

@@ -37,6 +37,7 @@ public static String getTooManyCategoricalFieldErr(int limit) {
         "Result index name has invalid character. Valid characters are a-z, 0-9, -(hyphen) and _(underscore)";
     public static String FAIL_TO_VALIDATE = "failed to validate";
     public static String INVALID_TIMESTAMP = "Timestamp field: (%s) must be of type date";
+    public static String NON_EXISTENT_TIMESTAMP_IN_INDEX = "Timestamp field: (%s) is not found in the (%s) index mapping";
     public static String NON_EXISTENT_TIMESTAMP = "Timestamp field: (%s) is not found in index mapping";
     public static String INVALID_NAME = "Valid characters for name are a-z, A-Z, 0-9, -(hyphen), _(underscore) and .(period)";
     // change this error message to make it compatible with old version's integration(nexus) test
@@ -74,6 +75,9 @@ public static String getTooManyCategoricalFieldErr(int limit) {
         + " characters.";
     public static final String INDEX_NOT_FOUND = "index does not exist";
     public static final String FAIL_TO_GET_MAPPING_MSG = "Fail to get the index mapping of %s";
+    public static final String FAIL_TO_GET_MAPPING = "Fail to get the index mapping";
+    public static final String TIMESTAMP_VALIDATION_FAILED = "Validation failed for timefield of %s, ";
+
     public static final String FAIL_TO_GET_CONFIG_MSG = "Fail to get config";
 
     // ======================================

src/main/java/org/opensearch/timeseries/rest/RestValidateAction.java

@@ -84,7 +84,6 @@ private Boolean validationTypesAreAccepted(String validationType) {
     public ValidateConfigRequest prepareRequest(RestRequest request, NodeClient client, String typesStr) throws IOException {
         XContentParser parser = request.contentParser();
         ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser);
-
         // if type param isn't blank and isn't a part of possible validation types throws exception
         if (!StringUtils.isBlank(typesStr)) {
             if (!validationTypesAreAccepted(typesStr)) {

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

@@ -8,10 +8,7 @@
 import static org.opensearch.timeseries.util.ParseUtils.checkFilterByBackendRoles;
 
 import java.time.Clock;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
+import java.util.*;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -205,7 +202,6 @@ public void validateExecute(
         storedContext.restore();
         Config config = request.getConfig();
         ActionListener<ValidateConfigResponse> validateListener = ActionListener.wrap(response -> {
-            logger.debug("Result of validation process " + response);
             // forcing response to be empty
             listener.onResponse(new ValidateConfigResponse((ConfigValidationIssue) null));
         }, exception -> {

src/main/java/org/opensearch/timeseries/util/CrossClusterConfigUtils.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.timeseries.util;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.client.Client;
+import org.opensearch.client.node.NodeClient;
+import org.opensearch.cluster.service.ClusterService;
+
+public class CrossClusterConfigUtils {
+    private static final Logger logger = LogManager.getLogger(ParseUtils.class);
+
+    /**
+     * Uses the clusterName to determine whether the target client is the local or a remote client,
+     * and returns the appropriate client.
+     * @param clusterName The name of the cluster to evaluate.
+     * @param client The local {@link NodeClient}.
+     * @param localClusterName The name of the local cluster.
+     * @return The local {@link NodeClient} for the local cluster, or a remote client for a remote cluster.
+     */
+    public static Client getClientForCluster(String clusterName, Client client, String localClusterName) {
+        return clusterName.equals(localClusterName) ? client : client.getRemoteClusterClient(clusterName);
+    }
+
+    /**
+     * Uses the clusterName to determine whether the target client is the local or a remote client,
+     * and returns the appropriate client.
+     * @param clusterName The name of the cluster to evaluate.
+     * @param client The local {@link NodeClient}.
+     * @param clusterService Used to retrieve the name of the local cluster.
+     * @return The local {@link NodeClient} for the local cluster, or a remote client for a remote cluster.
+     */
+    public static Client getClientForCluster(String clusterName, Client client, ClusterService clusterService) {
+        return getClientForCluster(clusterName, client, clusterService.getClusterName().value());
+    }
+
+    /**
+     * Parses the list of indexes into a map of cluster_name to List of index names
+     * @param indexes A list of index names in cluster_name:index_name format.
+     *      Local indexes can also be in index_name format.
+     * @param clusterService Used to retrieve the name of the local cluster.
+     * @return A map of cluster_name:index names
+     */
+    public static HashMap<String, List<String>> separateClusterIndexes(List<String> indexes, ClusterService clusterService) {
+        return separateClusterIndexes(indexes, clusterService.getClusterName().value());
+    }
+
+    /**
+     * Parses the list of indexes into a map of cluster_name to list of index_name
+     * @param indexes A list of index names in cluster_name:index_name format.
+     * @param localClusterName The name of the local cluster.
+     * @return A map of cluster_name to List index_name
+     */
+    public static HashMap<String, List<String>> separateClusterIndexes(List<String> indexes, String localClusterName) {
+        HashMap<String, List<String>> output = new HashMap<>();
+        for (String index : indexes) {
+            // Use the refactored method to get both cluster and index names in one call
+            Pair<String, String> clusterAndIndex = parseClusterAndIndexName(index);
+            String clusterName = clusterAndIndex.getKey();
+            String indexName = clusterAndIndex.getValue();
+            logger.info("clusterName: " + clusterName);
+            logger.info("indexName: " + indexName);
+
+            // If the index entry does not have a cluster_name, it indicates the index is on the local cluster.
+            if (clusterName.isEmpty()) {
+                clusterName = localClusterName;
+            }
+            output.computeIfAbsent(clusterName, k -> new ArrayList<>()).add(indexName);
+        }
+        return output;
+    }
+
+    /**
+     * Parses the cluster and index names from the given input string.
+     * The input can be in either "cluster_name:index_name" format or just "index_name".
+     * @param index The name of the index to evaluate.
+     * @return A Pair where the left is the cluster name (or empty if not present), and the right is the index name.
+     */
+    public static Pair<String, String> parseClusterAndIndexName(String index) {
+        if (index.contains(":")) {
+            String[] parts = index.split(":", 2);
+            String clusterName = parts[0];
+            String indexName = parts.length > 1 ? parts[1] : "";
+            return Pair.of(clusterName, indexName);
+        } else {
+            return Pair.of("", index);
+        }
+    }
+}

src/test/java/org/opensearch/action/admin/indices/mapping/get/AbstractForecasterActionHandlerTestCase.java

@@ -73,6 +73,8 @@ public class AbstractForecasterActionHandlerTestCase extends AbstractTimeSeriesT
     protected ThreadContext threadContext;
     protected SecurityClientUtil clientUtil;
     protected String categoricalField;
+    // @Mock
+    protected ClusterName clusterName;
 
     @SuppressWarnings("unchecked")
     @Override
@@ -85,6 +87,9 @@ public void setUp() throws Exception {
 
         clusterService = mock(ClusterService.class);
         ClusterName clusterName = new ClusterName("test");
+        clusterName = mock(ClusterName.class);
+        when(clusterService.getClusterName()).thenReturn(clusterName);
+        when(clusterName.value()).thenReturn("test");
         ClusterState clusterState = ClusterState.builder(clusterName).metadata(Metadata.builder().build()).build();
         when(clusterService.state()).thenReturn(clusterState);
 

src/test/java/org/opensearch/action/admin/indices/mapping/get/IndexAnomalyDetectorActionHandlerTests.java

@@ -100,6 +100,7 @@ public class IndexAnomalyDetectorActionHandlerTests extends AbstractTimeSeriesTe
     private RestRequest.Method method;
     private ADTaskManager adTaskManager;
     private SearchFeatureDao searchFeatureDao;
+    private ClusterName clusterName;
 
     @BeforeClass
     public static void beforeClass() {
@@ -157,6 +158,10 @@ public void setUp() throws Exception {
 
         searchFeatureDao = mock(SearchFeatureDao.class);
 
+        clusterName = mock(ClusterName.class);
+        when(clusterService.getClusterName()).thenReturn(clusterName);
+        when(clusterName.value()).thenReturn("test");
+
         handler = new IndexAnomalyDetectorActionHandler(
             clusterService,
             clientMock,

src/test/java/org/opensearch/action/admin/indices/mapping/get/IndexForecasterActionHandlerTests.java

@@ -40,6 +40,7 @@
 import org.opensearch.action.support.replication.ReplicationResponse;
 import org.opensearch.action.support.replication.ReplicationResponse.ShardInfo;
 import org.opensearch.client.node.NodeClient;
+import org.opensearch.cluster.ClusterName;
 import org.opensearch.cluster.routing.AllocationId;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.action.ActionListener;
@@ -240,7 +241,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
         verify(clientSpy, times(1)).execute(eq(GetAction.INSTANCE), any(), any());
     }
 
-    public void testFaiToParse() throws InterruptedException {
+    public void testFailToParse() throws InterruptedException {
         NodeClient client = new NodeClient(Settings.EMPTY, threadPool) {
             @Override
             public <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
@@ -273,6 +274,9 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
             }
         };
         NodeClient clientSpy = spy(client);
+        clusterName = mock(ClusterName.class);
+        when(clusterService.getClusterName()).thenReturn(clusterName);
+        when(clusterName.value()).thenReturn("test");
 
         method = RestRequest.Method.PUT;
 
@@ -508,6 +512,9 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
             }
         };
         NodeClient clientSpy = spy(client);
+        clusterName = mock(ClusterName.class);
+        when(clusterService.getClusterName()).thenReturn(clusterName);
+        when(clusterName.value()).thenReturn("test");
 
         method = RestRequest.Method.POST;
 

src/test/java/org/opensearch/action/admin/indices/mapping/get/ValidateAnomalyDetectorActionHandlerTests.java

@@ -43,6 +43,7 @@
 import org.opensearch.ad.task.ADTaskManager;
 import org.opensearch.client.Client;
 import org.opensearch.client.node.NodeClient;
+import org.opensearch.cluster.ClusterName;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
@@ -91,6 +92,7 @@ public class ValidateAnomalyDetectorActionHandlerTests extends AbstractTimeSerie
     @Mock
     protected ThreadPool threadPool;
     protected ThreadContext threadContext;
+    protected ClusterName mockClusterName;
 
     @SuppressWarnings("unchecked")
     @Override
@@ -106,7 +108,9 @@ public void setUp() throws Exception {
 
         anomalyDetectionIndices = mock(ADIndexManagement.class);
         when(anomalyDetectionIndices.doesConfigIndexExist()).thenReturn(true);
-
+        mockClusterName = mock(ClusterName.class);
+        when(clusterService.getClusterName()).thenReturn(mockClusterName);
+        when(mockClusterName.value()).thenReturn("test");
         detectorId = "123";
         seqNo = 0L;
         primaryTerm = 0L;

src/test/java/org/opensearch/action/admin/indices/mapping/get/ValidateForecasterActionHandlerTests.java

@@ -22,6 +22,7 @@
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.forecast.rest.handler.ValidateForecasterActionHandler;
+import org.opensearch.timeseries.common.exception.ValidationException;
 import org.opensearch.timeseries.model.ValidationAspect;
 
 public class ValidateForecasterActionHandlerTests extends AbstractForecasterActionHandlerTestCase {
@@ -100,7 +101,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
             assertTrue("should not reach here", false);
             inProgressLatch.countDown();
         }, e -> {
-            assertTrue(e instanceof IllegalArgumentException);
+            assertTrue(e instanceof ValidationException);
             inProgressLatch.countDown();
         }));
         assertTrue(inProgressLatch.await(100, TimeUnit.SECONDS));

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

@@ -1370,6 +1370,73 @@ public void testValidateAnomalyDetectorWithNoTimeField() throws Exception {
         assertEquals("time field missing", CommonMessages.NULL_TIME_FIELD, messageMap.get("time_field").get("message"));
     }
 
+    public void testValidateAnomalyDetectorWithMultipleIndicesOneNotFound() throws Exception {
+        TestHelpers.createIndex(client(), "test-index", TestHelpers.toHttpEntity("{\"timestamp\": " + Instant.now().toEpochMilli() + "}"));
+        Response resp = TestHelpers
+            .makeRequest(
+                client(),
+                "POST",
+                TestHelpers.AD_BASE_DETECTORS_URI + "/_validate",
+                ImmutableMap.of(),
+                TestHelpers
+                    .toHttpEntity(
+                        "{\"name\":\""
+                            + "test-detector"
+                            + "\",\"description\":\"Test detector\",\"time_field\":\"timestamp\","
+                            + "\"indices\":[\"test-index\", \"test-index-2\"],\"feature_attributes\":[{\"feature_name\":\"cpu-sum\",\""
+                            + "feature_enabled\":true,\"aggregation_query\":{\"total_cpu\":{\"sum\":{\"field\":\"cpu\"}}}},"
+                            + "{\"feature_name\":\"error-sum\",\"feature_enabled\":true,\"aggregation_query\":"
+                            + "{\"total_error\":"
+                            + "{\"sum\":{\"field\":\"error\"}}}}],\"filter_query\":{\"bool\":{\"filter\":[{\"exists\":"
+                            + "{\"field\":"
+                            + "\"cpu\",\"boost\":1}}],\"adjust_pure_negative\":true,\"boost\":1}},\"detection_interval\":"
+                            + "{\"period\":{\"interval\":1,\"unit\":\"Minutes\"}},"
+                            + "\"window_delay\":{\"period\":{\"interval\":2,\"unit\":\"Minutes\"}},"
+                            + "\"shingle_size\": 8}"
+                    ),
+                null
+            );
+        Map<String, Object> responseMap = entityAsMap(resp);
+
+        @SuppressWarnings("unchecked")
+        Map<String, Map<String, String>> messageMap = (Map<String, Map<String, String>>) XContentMapValues
+            .extractValue("detector", responseMap);
+        String errorMessage = "index does not exist";
+        assertEquals("index does not exist", errorMessage, messageMap.get("indices").get("message"));
+    }
+
+    public void testValidateAnomalyDetectorWithMultipleIndices() throws Exception {
+        TestHelpers.createIndexWithTimeField(client(), "test-index", TIME_FIELD);
+        TestHelpers.createIndexWithTimeField(client(), "test-index-2", TIME_FIELD);
+
+        Response resp = TestHelpers
+            .makeRequest(
+                client(),
+                "POST",
+                TestHelpers.AD_BASE_DETECTORS_URI + "/_validate",
+                ImmutableMap.of(),
+                TestHelpers
+                    .toHttpEntity(
+                        "{\"name\":\""
+                            + "test-detector"
+                            + "\",\"description\":\"Test detector\",\"time_field\":\"timestamp\","
+                            + "\"indices\":[\"test-index\", \"test-index-2\"],\"feature_attributes\":[{\"feature_name\":\"cpu-sum\",\""
+                            + "feature_enabled\":true,\"aggregation_query\":{\"total_cpu\":{\"sum\":{\"field\":\"cpu\"}}}},"
+                            + "{\"feature_name\":\"error-sum\",\"feature_enabled\":true,\"aggregation_query\":"
+                            + "{\"total_error\":"
+                            + "{\"sum\":{\"field\":\"error\"}}}}],\"filter_query\":{\"bool\":{\"filter\":[{\"exists\":"
+                            + "{\"field\":"
+                            + "\"cpu\",\"boost\":1}}],\"adjust_pure_negative\":true,\"boost\":1}},\"detection_interval\":"
+                            + "{\"period\":{\"interval\":1,\"unit\":\"Minutes\"}},"
+                            + "\"window_delay\":{\"period\":{\"interval\":2,\"unit\":\"Minutes\"}},"
+                            + "\"shingle_size\": 8}"
+                    ),
+                null
+            );
+        Map<String, Object> responseMap = entityAsMap(resp);
+        assertEquals("no issue, empty response body", new HashMap<String, Object>(), responseMap);
+    }
+
     public void testValidateAnomalyDetectorWithIncorrectShingleSize() throws Exception {
         TestHelpers.createIndex(client(), "test-index", TestHelpers.toHttpEntity("{\"timestamp\": " + Instant.now().toEpochMilli() + "}"));
         Response resp = TestHelpers

src/test/java/org/opensearch/ad/transport/ValidateAnomalyDetectorTransportActionTests.java

@@ -488,9 +488,8 @@ public void testValidateAnomalyDetectorWithNonExistentTimefield() throws IOExcep
         ValidateConfigResponse response = client().execute(ValidateAnomalyDetectorAction.INSTANCE, request).actionGet(5_000);
         assertEquals(ValidationIssueType.TIMEFIELD_FIELD, response.getIssue().getType());
         assertEquals(ValidationAspect.DETECTOR, response.getIssue().getAspect());
-        assertEquals(
-            String.format(Locale.ROOT, CommonMessages.NON_EXISTENT_TIMESTAMP, anomalyDetector.getTimeField()),
-            response.getIssue().getMessage()
+        assertTrue(
+            response.getIssue().getMessage().contains("Timestamp field: (" + anomalyDetector.getTimeField() + ") is not found in the")
         );
     }
 
@@ -511,9 +510,11 @@ public void testValidateAnomalyDetectorWithNonDateTimeField() throws IOException
         ValidateConfigResponse response = client().execute(ValidateAnomalyDetectorAction.INSTANCE, request).actionGet(5_000);
         assertEquals(ValidationIssueType.TIMEFIELD_FIELD, response.getIssue().getType());
         assertEquals(ValidationAspect.DETECTOR, response.getIssue().getAspect());
-        assertEquals(
-            String.format(Locale.ROOT, CommonMessages.INVALID_TIMESTAMP, anomalyDetector.getTimeField()),
-            response.getIssue().getMessage()
+        assertTrue(
+            response
+                .getIssue()
+                .getMessage()
+                .contains(String.format(Locale.ROOT, CommonMessages.INVALID_TIMESTAMP, anomalyDetector.getTimeField()))
         );
     }