Adding remote index and multi-index checks in validation

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,102 @@
+/*
+ * 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.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) {
+        logger.info("clusterName1: " + clusterName);
+        logger.info("clusterService.getClusterName().value(): " + clusterService.getClusterName().value());
+
+        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) {
+            String clusterName = parseClusterName(index);
+            String indexName = parseIndexName(index);
+
+            // 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;
+    }
+
+    /**
+     * @param index The name of the index to evaluate.
+     *      Can be in either cluster_name:index_name or index_name format.
+     * @return The index name.
+     */
+    public static String parseIndexName(String index) {
+        if (index.contains(":")) {
+            String[] parts = index.split(":");
+            return parts.length > 1 ? parts[1] : index;
+        } else {
+            return index;
+        }
+    }
+
+    /**
+     * @param index The name of the index to evaluate.
+     *      Can be in either cluster_name:index_name or index_name format.
+     * @return The index name.
+     */
+    public static String parseClusterName(String index) {
+        return index.contains(":") ? index.substring(0, index.indexOf(':')) : "";
+    }
+}

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

@@ -11,8 +11,6 @@
 
 package org.opensearch.timeseries.util;
 
-import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.atomic.AtomicInteger;

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/ad/transport/ValidateAnomalyDetectorTransportActionTests.java

@@ -490,9 +490,11 @@ 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(String.format(Locale.ROOT, CommonMessages.INVALID_TIMESTAMP, anomalyDetector.getTimeField()))
         );
     }
 
@@ -513,9 +515,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()))
         );
     }
 

src/test/java/org/opensearch/timeseries/util/CrossClusterConfigUtilsTests.java

@@ -0,0 +1,66 @@
+package org.opensearch.timeseries.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+
+import org.mockito.Mock;
+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.test.OpenSearchTestCase;
+
+public class CrossClusterConfigUtilsTests extends OpenSearchTestCase {
+
+    @Mock
+    private Client clientMock;
+
+    public void testGetClientForClusterLocalCluster() {
+        String clusterName = "localCluster";
+        Client mockClient = mock(NodeClient.class);
+        String localClusterName = "localCluster";
+
+        Client result = CrossClusterConfigUtils.getClientForCluster(clusterName, mockClient, localClusterName);
+
+        assertEquals(mockClient, result);
+    }
+
+    public void testGetClientForClusterRemoteCluster() {
+        String clusterName = "remoteCluster";
+        Client mockClient = mock(NodeClient.class);
+        // Client mockRemoteClient = mock(Client.class);
+
+        when(mockClient.getRemoteClusterClient(clusterName)).thenReturn(mockClient);
+
+        Client result = CrossClusterConfigUtils.getClientForCluster(clusterName, mockClient, "localCluster");
+
+        assertEquals(mockClient, result);
+    }
+
+    public void testSeparateClusterIndexesRemoteCluster() {
+        List<String> indexes = Arrays.asList("remoteCluster:index1", "index2");
+        ClusterService mockClusterService = mock(ClusterService.class);
+        when(mockClusterService.getClusterName()).thenReturn(new ClusterName("localCluster"));
+
+        HashMap<String, List<String>> result = CrossClusterConfigUtils.separateClusterIndexes(indexes, mockClusterService);
+
+        assertEquals(2, result.size());
+        assertEquals(Arrays.asList("index1"), result.get("remoteCluster"));
+        assertEquals(Arrays.asList("index2"), result.get("localCluster"));
+    }
+
+    public void testParseIndexName() {
+        assertEquals("index1", CrossClusterConfigUtils.parseIndexName("remoteCluster:index1"));
+        assertEquals("index2", CrossClusterConfigUtils.parseIndexName("index2"));
+    }
+
+    public void testParseClusterName() {
+        assertEquals("remoteCluster", CrossClusterConfigUtils.parseClusterName("remoteCluster:index1"));
+        assertEquals("", CrossClusterConfigUtils.parseClusterName("index2"));
+    }
+}