Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Adding remote index and multi-index checks in validation #1290

Merged
merged 2 commits into from
Sep 4, 2024
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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";

// ======================================
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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)) {
Expand Down

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 -> {
Expand Down
Original file line number Diff line number Diff line change
@@ -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());
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are these logs still needed?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed


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);
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you make one call and return a Pair instead of calling similar logic twice?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Changed


// 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(':')) : "";
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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);

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -91,6 +92,7 @@ public class ValidateAnomalyDetectorActionHandlerTests extends AbstractTimeSerie
@Mock
protected ThreadPool threadPool;
protected ThreadContext threadContext;
protected ClusterName mockClusterName;

@SuppressWarnings("unchecked")
@Override
Expand All @@ -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;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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()))
);
}

Expand All @@ -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()))
);
}

Expand Down
Original file line number Diff line number Diff line change
@@ -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"));
}
}
Loading