Skip to content

Commit 056f06f

Browse files
amitgalitzamitgalitz
committed
Not blocking detector creation on unknown feature validation error (#1366)
* don't fail on unknown exception Signed-off-by: Amit Galitzky <amgalitz@amazon.com> * fixing test Signed-off-by: Amit Galitzky <amgalitz@amazon.com> * order of needed permissions is changed on latest security version or at least not always consistent now Signed-off-by: Amit Galitzky <amgalitz@amazon.com> * refactor customNodeclient Signed-off-by: Amit Galitzky <amgalitz@amazon.com> --------- Signed-off-by: Amit Galitzky <amgalitz@amazon.com>
1 parent c86f541 commit 056f06f

File tree

5 files changed

+93
-10
lines changed

5 files changed

+93
-10
lines changed

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

+10-3
Original file line numberDiff line numberDiff line change
@@ -890,6 +890,7 @@ protected void validateConfigFeatures(String id, boolean indexingDryRun, ActionL
890890
feature.getId()
891891
);
892892
ssb.aggregation(internalAgg.getAggregatorFactories().iterator().next());
893+
ssb.trackTotalHits(false);
893894
SearchRequest searchRequest = new SearchRequest().indices(config.getIndices().toArray(new String[0])).source(ssb);
894895
ActionListener<SearchResponse> searchResponseListener = ActionListener.wrap(response -> {
895896
Optional<double[]> aggFeatureResult = searchFeatureDao.parseResponse(response, Arrays.asList(feature.getId()), false);
@@ -905,13 +906,19 @@ protected void validateConfigFeatures(String id, boolean indexingDryRun, ActionL
905906
}
906907
}, e -> {
907908
String errorMessage;
908-
if (isExceptionCausedByInvalidQuery(e)) {
909+
if (isExceptionCausedByInvalidQuery(e) || e instanceof TimeSeriesException) {
909910
errorMessage = CommonMessages.FEATURE_WITH_INVALID_QUERY_MSG + feature.getName();
911+
logger.error(errorMessage, e);
912+
multiFeatureQueriesResponseListener.onFailure(new OpenSearchStatusException(errorMessage, RestStatus.BAD_REQUEST, e));
910913
} else {
911914
errorMessage = CommonMessages.UNKNOWN_SEARCH_QUERY_EXCEPTION_MSG + feature.getName();
915+
logger.error(errorMessage, e);
916+
// If we see an unexpected error such as timeout or some task cancellation cause of search backpressure
917+
// we don't want to block detector creation as this is unlikely an error due to wrong configs
918+
// but we want to record what error was seen
919+
multiFeatureQueriesResponseListener
920+
.onResponse(new MergeableList<>(new ArrayList<>(Collections.singletonList(Optional.empty()))));
912921
}
913-
logger.error(errorMessage, e);
914-
multiFeatureQueriesResponseListener.onFailure(new OpenSearchStatusException(errorMessage, RestStatus.BAD_REQUEST, e));
915922
});
916923
clientUtil.asyncRequestWithInjectedSecurity(searchRequest, client::search, user, client, context, searchResponseListener);
917924
}

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

+18-2
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@
5656
import org.opensearch.common.unit.TimeValue;
5757
import org.opensearch.core.action.ActionListener;
5858
import org.opensearch.core.action.ActionResponse;
59+
import org.opensearch.core.rest.RestStatus;
5960
import org.opensearch.rest.RestRequest;
6061
import org.opensearch.threadpool.TestThreadPool;
6162
import org.opensearch.threadpool.ThreadPool;
@@ -207,7 +208,7 @@ public void testMoreThanTenThousandSingleEntityDetectors() throws IOException, I
207208

208209
// extend NodeClient since its execute method is final and mockito does not allow to mock final methods
209210
// we can also use spy to overstep the final methods
210-
NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
211+
NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
211212
NodeClient clientSpy = spy(client);
212213
NodeStateManager nodeStateManager = mock(NodeStateManager.class);
213214
clientUtil = new SecurityClientUtil(nodeStateManager, settings);
@@ -546,10 +547,14 @@ public void testUpdateTextField() throws IOException, InterruptedException {
546547
public static NodeClient getCustomNodeClient(
547548
SearchResponse detectorResponse,
548549
SearchResponse userIndexResponse,
550+
SearchResponse configInputIndicesResponse,
551+
boolean useConfigInputIndicesResponse,
549552
AnomalyDetector detector,
550553
ThreadPool pool
551554
) {
552555
return new NodeClient(Settings.EMPTY, pool) {
556+
private int searchCallCount = 0;
557+
553558
@Override
554559
public <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
555560
ActionType<Response> action,
@@ -560,8 +565,19 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
560565
if (action.equals(SearchAction.INSTANCE)) {
561566
assertTrue(request instanceof SearchRequest);
562567
SearchRequest searchRequest = (SearchRequest) request;
568+
searchCallCount++;
563569
if (searchRequest.indices()[0].equals(CommonName.CONFIG_INDEX)) {
564570
listener.onResponse((Response) detectorResponse);
571+
} else if (useConfigInputIndicesResponse
572+
&& Arrays.equals(searchRequest.indices(), detector.getIndices().toArray(new String[0]))
573+
&& searchRequest.source().aggregations() == null) {
574+
listener.onResponse((Response) configInputIndicesResponse);
575+
// Call for feature validation occurs on the 3rd call and we want to make sure we supplied a response to the
576+
// previous call.
577+
} else if (searchCallCount == 3 && useConfigInputIndicesResponse) {
578+
// This is the third search call, which should be for featureConfig and we want to replicate something like a
579+
// timeout exception
580+
listener.onFailure(new OpenSearchStatusException("timeout", RestStatus.BAD_REQUEST));
565581
} else {
566582
listener.onResponse((Response) userIndexResponse);
567583
}
@@ -590,7 +606,7 @@ public void testMoreThanTenMultiEntityDetectors() throws IOException, Interrupte
590606
when(userIndexResponse.getHits()).thenReturn(TestHelpers.createSearchHits(userIndexHits));
591607
// extend NodeClient since its execute method is final and mockito does not allow to mock final methods
592608
// we can also use spy to overstep the final methods
593-
NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
609+
NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
594610
NodeClient clientSpy = spy(client);
595611
NodeStateManager nodeStateManager = mock(NodeStateManager.class);
596612
clientUtil = new SecurityClientUtil(nodeStateManager, settings);

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

+60-2
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
import org.mockito.Mockito;
3232
import org.mockito.MockitoAnnotations;
3333
import org.opensearch.action.search.SearchResponse;
34+
import org.opensearch.action.support.PlainActionFuture;
3435
import org.opensearch.action.support.WriteRequest;
3536
import org.opensearch.ad.indices.ADIndex;
3637
import org.opensearch.ad.indices.ADIndexManagement;
@@ -54,6 +55,7 @@
5455
import org.opensearch.timeseries.AbstractTimeSeriesTest;
5556
import org.opensearch.timeseries.NodeStateManager;
5657
import org.opensearch.timeseries.TestHelpers;
58+
import org.opensearch.timeseries.common.exception.TimeSeriesException;
5759
import org.opensearch.timeseries.common.exception.ValidationException;
5860
import org.opensearch.timeseries.feature.SearchFeatureDao;
5961
import org.opensearch.timeseries.model.ValidationAspect;
@@ -150,7 +152,7 @@ public void testValidateMoreThanThousandSingleEntityDetectorLimit() throws IOExc
150152
// extend NodeClient since its execute method is final and mockito does not allow to mock final methods
151153
// we can also use spy to overstep the final methods
152154
NodeClient client = IndexAnomalyDetectorActionHandlerTests
153-
.getCustomNodeClient(detectorResponse, userIndexResponse, singleEntityDetector, threadPool);
155+
.getCustomNodeClient(detectorResponse, userIndexResponse, null, false, singleEntityDetector, threadPool);
154156

155157
NodeClient clientSpy = spy(client);
156158
NodeStateManager nodeStateManager = mock(NodeStateManager.class);
@@ -208,7 +210,7 @@ public void testValidateMoreThanTenMultiEntityDetectorsLimit() throws IOExceptio
208210
// extend NodeClient since its execute method is final and mockito does not allow to mock final methods
209211
// we can also use spy to overstep the final methods
210212
NodeClient client = IndexAnomalyDetectorActionHandlerTests
211-
.getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
213+
.getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
212214
NodeClient clientSpy = spy(client);
213215
NodeStateManager nodeStateManager = mock(NodeStateManager.class);
214216
SecurityClientUtil clientUtil = new SecurityClientUtil(nodeStateManager, settings);
@@ -250,4 +252,60 @@ public void testValidateMoreThanTenMultiEntityDetectorsLimit() throws IOExceptio
250252
assertTrue(inProgressLatch.await(100, TimeUnit.SECONDS));
251253
verify(clientSpy, never()).execute(eq(GetMappingsAction.INSTANCE), any(), any());
252254
}
255+
256+
// This test also validates that if we get a non timeseries exception or not an invalid query that we will not completely block
257+
// detector creation, this is applicable like things when we get timeout not cause of AD configuration errors but because cluster
258+
// is momentarily under utilized.
259+
public void testValidateMoreThanTenMultiEntityDetectorsLimitDuplicateNameFailure() throws IOException, InterruptedException {
260+
SearchResponse mockResponse = mock(SearchResponse.class);
261+
int totalHits = 1;
262+
when(mockResponse.getHits()).thenReturn(TestHelpers.createSearchHits(totalHits));
263+
SearchResponse detectorResponse = mock(SearchResponse.class);
264+
when(detectorResponse.getHits()).thenReturn(TestHelpers.createSearchHits(totalHits));
265+
SearchResponse userIndexResponse = mock(SearchResponse.class);
266+
when(userIndexResponse.getHits()).thenReturn(TestHelpers.createSearchHits(0));
267+
AnomalyDetector singleEntityDetector = TestHelpers.randomAnomalyDetector(TestHelpers.randomUiMetadata(), null, true);
268+
269+
SearchResponse configInputIndicesResponse = mock(SearchResponse.class);
270+
when(configInputIndicesResponse.getHits()).thenReturn(TestHelpers.createSearchHits(2));
271+
272+
// extend NodeClient since its execute method is final and mockito does not allow to mock final methods
273+
// we can also use spy to overstep the final methods
274+
NodeClient client = IndexAnomalyDetectorActionHandlerTests
275+
.getCustomNodeClient(detectorResponse, userIndexResponse, configInputIndicesResponse, true, singleEntityDetector, threadPool);
276+
277+
NodeClient clientSpy = spy(client);
278+
NodeStateManager nodeStateManager = mock(NodeStateManager.class);
279+
SecurityClientUtil clientUtil = new SecurityClientUtil(nodeStateManager, settings);
280+
281+
handler = new ValidateAnomalyDetectorActionHandler(
282+
clusterService,
283+
clientSpy,
284+
clientUtil,
285+
anomalyDetectionIndices,
286+
singleEntityDetector,
287+
requestTimeout,
288+
maxSingleEntityAnomalyDetectors,
289+
maxMultiEntityAnomalyDetectors,
290+
maxAnomalyFeatures,
291+
maxCategoricalFields,
292+
method,
293+
xContentRegistry(),
294+
null,
295+
searchFeatureDao,
296+
ValidationAspect.DETECTOR.getName(),
297+
clock,
298+
settings
299+
);
300+
PlainActionFuture<ValidateConfigResponse> future = PlainActionFuture.newFuture();
301+
handler.start(future);
302+
try {
303+
future.actionGet(100, TimeUnit.SECONDS);
304+
fail("should not reach here");
305+
} catch (Exception e) {
306+
assertTrue(e instanceof TimeSeriesException);
307+
assertTrue(e.getMessage().contains("Cannot create anomaly detector with name"));
308+
}
309+
verify(clientSpy, never()).execute(eq(GetMappingsAction.INSTANCE), any(), any());
310+
}
253311
}

‎src/test/java/org/opensearch/ad/rest/SecureADRestIT.java

+3-1
Original file line numberDiff line numberDiff line change
@@ -408,7 +408,9 @@ public void testCreateAnomalyDetectorWithCustomResultIndex() throws IOException
408408
Assert
409409
.assertTrue(
410410
"got " + exception.getMessage(),
411-
exception.getMessage().contains("no permissions for [indices:admin/aliases, indices:admin/create]")
411+
exception.getMessage().contains("indices:admin/aliases")
412+
&& exception.getMessage().contains("indices:admin/create")
413+
&& exception.getMessage().contains("no permissions for")
412414
);
413415

414416
// User cat has permission to create index

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

+2-2
Original file line numberDiff line numberDiff line change
@@ -222,7 +222,7 @@ public void testValidateAnomalyDetectorWithInvalidFeatureField() throws IOExcept
222222
}
223223

224224
@Test
225-
public void testValidateAnomalyDetectorWithUnknownFeatureField() throws IOException {
225+
public void testValidateAnomalyDetectorWithInvalidFeatureDueToTimeSeriesException() throws IOException {
226226
AggregationBuilder aggregationBuilder = TestHelpers.parseAggregation("{\"test\":{\"terms\":{\"field\":\"type\"}}}");
227227
AnomalyDetector anomalyDetector = TestHelpers
228228
.randomAnomalyDetector(
@@ -245,7 +245,7 @@ public void testValidateAnomalyDetectorWithUnknownFeatureField() throws IOExcept
245245
assertNotNull(response.getIssue());
246246
assertEquals(ValidationIssueType.FEATURE_ATTRIBUTES, response.getIssue().getType());
247247
assertEquals(ValidationAspect.DETECTOR, response.getIssue().getAspect());
248-
assertTrue(response.getIssue().getMessage().contains(CommonMessages.UNKNOWN_SEARCH_QUERY_EXCEPTION_MSG));
248+
assertTrue(response.getIssue().getMessage().contains(CommonMessages.FEATURE_WITH_INVALID_QUERY_MSG));
249249
assertTrue(response.getIssue().getSubIssues().containsKey(nameField));
250250
}
251251

0 commit comments

Comments
 (0)