Not blocking detector creation on unknown feature validation error (#1366) (#1371)

amitgalitz · web-flow · commit 94acda13136a · 2024-11-26T08:42:10.000-08:00
* don't fail on unknown exception



* fixing test



* order of needed permissions is changed on latest security version or at least not always consistent now



* refactor customNodeclient



---------

Signed-off-by: Amit Galitzky &lt;amgalitz@amazon.com&gt;
diff --git a/src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java b/src/main/java/org/opensearch/timeseries/rest/handler/AbstractTimeSeriesActionHandler.java
@@ -890,6 +890,7 @@ protected void validateConfigFeatures(String id, boolean indexingDryRun, ActionL
                 feature.getId()
             );
             ssb.aggregation(internalAgg.getAggregatorFactories().iterator().next());
+            ssb.trackTotalHits(false);
             SearchRequest searchRequest = new SearchRequest().indices(config.getIndices().toArray(new String[0])).source(ssb);
             ActionListener<SearchResponse> searchResponseListener = ActionListener.wrap(response -> {
                 Optional<double[]> aggFeatureResult = searchFeatureDao.parseResponse(response, Arrays.asList(feature.getId()), false);
@@ -905,13 +906,19 @@ protected void validateConfigFeatures(String id, boolean indexingDryRun, ActionL
                 }
             }, e -> {
                 String errorMessage;
-                if (isExceptionCausedByInvalidQuery(e)) {
+                if (isExceptionCausedByInvalidQuery(e) || e instanceof TimeSeriesException) {
                     errorMessage = CommonMessages.FEATURE_WITH_INVALID_QUERY_MSG + feature.getName();
+                    logger.error(errorMessage, e);
+                    multiFeatureQueriesResponseListener.onFailure(new OpenSearchStatusException(errorMessage, RestStatus.BAD_REQUEST, e));
                 } else {
                     errorMessage = CommonMessages.UNKNOWN_SEARCH_QUERY_EXCEPTION_MSG + feature.getName();
+                    logger.error(errorMessage, e);
+                    // If we see an unexpected error such as timeout or some task cancellation cause of search backpressure
+                    // we don't want to block detector creation as this is unlikely an error due to wrong configs
+                    // but we want to record what error was seen
+                    multiFeatureQueriesResponseListener
+                        .onResponse(new MergeableList<>(new ArrayList<>(Collections.singletonList(Optional.empty()))));
                 }
-                logger.error(errorMessage, e);
-                multiFeatureQueriesResponseListener.onFailure(new OpenSearchStatusException(errorMessage, RestStatus.BAD_REQUEST, e));
             });
             clientUtil.asyncRequestWithInjectedSecurity(searchRequest, client::search, user, client, context, searchResponseListener);
         }
diff --git a/src/test/java/org/opensearch/action/admin/indices/mapping/get/IndexAnomalyDetectorActionHandlerTests.java b/src/test/java/org/opensearch/action/admin/indices/mapping/get/IndexAnomalyDetectorActionHandlerTests.java
@@ -56,6 +56,7 @@
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.action.ActionResponse;
+import org.opensearch.core.rest.RestStatus;
 import org.opensearch.rest.RestRequest;
 import org.opensearch.threadpool.TestThreadPool;
 import org.opensearch.threadpool.ThreadPool;
@@ -207,7 +208,7 @@ public void testMoreThanTenThousandSingleEntityDetectors() throws IOException, I
 
         // extend NodeClient since its execute method is final and mockito does not allow to mock final methods
         // we can also use spy to overstep the final methods
-        NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
+        NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
         NodeClient clientSpy = spy(client);
         NodeStateManager nodeStateManager = mock(NodeStateManager.class);
         clientUtil = new SecurityClientUtil(nodeStateManager, settings);
@@ -546,10 +547,14 @@ public void testUpdateTextField() throws IOException, InterruptedException {
     public static NodeClient getCustomNodeClient(
         SearchResponse detectorResponse,
         SearchResponse userIndexResponse,
+        SearchResponse configInputIndicesResponse,
+        boolean useConfigInputIndicesResponse,
         AnomalyDetector detector,
         ThreadPool pool
     ) {
         return new NodeClient(Settings.EMPTY, pool) {
+            private int searchCallCount = 0;
+
             @Override
             public <Request extends ActionRequest, Response extends ActionResponse> void doExecute(
                 ActionType<Response> action,
@@ -560,8 +565,19 @@ public <Request extends ActionRequest, Response extends ActionResponse> void doE
                     if (action.equals(SearchAction.INSTANCE)) {
                         assertTrue(request instanceof SearchRequest);
                         SearchRequest searchRequest = (SearchRequest) request;
+                        searchCallCount++;
                         if (searchRequest.indices()[0].equals(CommonName.CONFIG_INDEX)) {
                             listener.onResponse((Response) detectorResponse);
+                        } else if (useConfigInputIndicesResponse
+                            && Arrays.equals(searchRequest.indices(), detector.getIndices().toArray(new String[0]))
+                            && searchRequest.source().aggregations() == null) {
+                            listener.onResponse((Response) configInputIndicesResponse);
+                            // Call for feature validation occurs on the 3rd call and we want to make sure we supplied a response to the
+                            // previous call.
+                        } else if (searchCallCount == 3 && useConfigInputIndicesResponse) {
+                            // This is the third search call, which should be for featureConfig and we want to replicate something like a
+                            // timeout exception
+                            listener.onFailure(new OpenSearchStatusException("timeout", RestStatus.BAD_REQUEST));
                         } else {
                             listener.onResponse((Response) userIndexResponse);
                         }
@@ -590,7 +606,7 @@ public void testMoreThanTenMultiEntityDetectors() throws IOException, Interrupte
         when(userIndexResponse.getHits()).thenReturn(TestHelpers.createSearchHits(userIndexHits));
         // extend NodeClient since its execute method is final and mockito does not allow to mock final methods
         // we can also use spy to overstep the final methods
-        NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
+        NodeClient client = getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
         NodeClient clientSpy = spy(client);
         NodeStateManager nodeStateManager = mock(NodeStateManager.class);
         clientUtil = new SecurityClientUtil(nodeStateManager, settings);
diff --git a/src/test/java/org/opensearch/action/admin/indices/mapping/get/ValidateAnomalyDetectorActionHandlerTests.java b/src/test/java/org/opensearch/action/admin/indices/mapping/get/ValidateAnomalyDetectorActionHandlerTests.java
@@ -31,6 +31,7 @@
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
 import org.opensearch.action.search.SearchResponse;
+import org.opensearch.action.support.PlainActionFuture;
 import org.opensearch.action.support.WriteRequest;
 import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.ADIndexManagement;
@@ -54,6 +55,7 @@
 import org.opensearch.timeseries.AbstractTimeSeriesTest;
 import org.opensearch.timeseries.NodeStateManager;
 import org.opensearch.timeseries.TestHelpers;
+import org.opensearch.timeseries.common.exception.TimeSeriesException;
 import org.opensearch.timeseries.common.exception.ValidationException;
 import org.opensearch.timeseries.feature.SearchFeatureDao;
 import org.opensearch.timeseries.model.ValidationAspect;
@@ -150,7 +152,7 @@ public void testValidateMoreThanThousandSingleEntityDetectorLimit() throws IOExc
         // extend NodeClient since its execute method is final and mockito does not allow to mock final methods
         // we can also use spy to overstep the final methods
         NodeClient client = IndexAnomalyDetectorActionHandlerTests
-            .getCustomNodeClient(detectorResponse, userIndexResponse, singleEntityDetector, threadPool);
+            .getCustomNodeClient(detectorResponse, userIndexResponse, null, false, singleEntityDetector, threadPool);
 
         NodeClient clientSpy = spy(client);
         NodeStateManager nodeStateManager = mock(NodeStateManager.class);
@@ -208,7 +210,7 @@ public void testValidateMoreThanTenMultiEntityDetectorsLimit() throws IOExceptio
         // extend NodeClient since its execute method is final and mockito does not allow to mock final methods
         // we can also use spy to overstep the final methods
         NodeClient client = IndexAnomalyDetectorActionHandlerTests
-            .getCustomNodeClient(detectorResponse, userIndexResponse, detector, threadPool);
+            .getCustomNodeClient(detectorResponse, userIndexResponse, null, false, detector, threadPool);
         NodeClient clientSpy = spy(client);
         NodeStateManager nodeStateManager = mock(NodeStateManager.class);
         SecurityClientUtil clientUtil = new SecurityClientUtil(nodeStateManager, settings);
@@ -250,4 +252,60 @@ public void testValidateMoreThanTenMultiEntityDetectorsLimit() throws IOExceptio
         assertTrue(inProgressLatch.await(100, TimeUnit.SECONDS));
         verify(clientSpy, never()).execute(eq(GetMappingsAction.INSTANCE), any(), any());
     }
+
+    // This test also validates that if we get a non timeseries exception or not an invalid query that we will not completely block
+    // detector creation, this is applicable like things when we get timeout not cause of AD configuration errors but because cluster
+    // is momentarily under utilized.
+    public void testValidateMoreThanTenMultiEntityDetectorsLimitDuplicateNameFailure() throws IOException, InterruptedException {
+        SearchResponse mockResponse = mock(SearchResponse.class);
+        int totalHits = 1;
+        when(mockResponse.getHits()).thenReturn(TestHelpers.createSearchHits(totalHits));
+        SearchResponse detectorResponse = mock(SearchResponse.class);
+        when(detectorResponse.getHits()).thenReturn(TestHelpers.createSearchHits(totalHits));
+        SearchResponse userIndexResponse = mock(SearchResponse.class);
+        when(userIndexResponse.getHits()).thenReturn(TestHelpers.createSearchHits(0));
+        AnomalyDetector singleEntityDetector = TestHelpers.randomAnomalyDetector(TestHelpers.randomUiMetadata(), null, true);
+
+        SearchResponse configInputIndicesResponse = mock(SearchResponse.class);
+        when(configInputIndicesResponse.getHits()).thenReturn(TestHelpers.createSearchHits(2));
+
+        // extend NodeClient since its execute method is final and mockito does not allow to mock final methods
+        // we can also use spy to overstep the final methods
+        NodeClient client = IndexAnomalyDetectorActionHandlerTests
+            .getCustomNodeClient(detectorResponse, userIndexResponse, configInputIndicesResponse, true, singleEntityDetector, threadPool);
+
+        NodeClient clientSpy = spy(client);
+        NodeStateManager nodeStateManager = mock(NodeStateManager.class);
+        SecurityClientUtil clientUtil = new SecurityClientUtil(nodeStateManager, settings);
+
+        handler = new ValidateAnomalyDetectorActionHandler(
+            clusterService,
+            clientSpy,
+            clientUtil,
+            anomalyDetectionIndices,
+            singleEntityDetector,
+            requestTimeout,
+            maxSingleEntityAnomalyDetectors,
+            maxMultiEntityAnomalyDetectors,
+            maxAnomalyFeatures,
+            maxCategoricalFields,
+            method,
+            xContentRegistry(),
+            null,
+            searchFeatureDao,
+            ValidationAspect.DETECTOR.getName(),
+            clock,
+            settings
+        );
+        PlainActionFuture<ValidateConfigResponse> future = PlainActionFuture.newFuture();
+        handler.start(future);
+        try {
+            future.actionGet(100, TimeUnit.SECONDS);
+            fail("should not reach here");
+        } catch (Exception e) {
+            assertTrue(e instanceof TimeSeriesException);
+            assertTrue(e.getMessage().contains("Cannot create anomaly detector with name"));
+        }
+        verify(clientSpy, never()).execute(eq(GetMappingsAction.INSTANCE), any(), any());
+    }
 }
diff --git a/src/test/java/org/opensearch/ad/rest/SecureADRestIT.java b/src/test/java/org/opensearch/ad/rest/SecureADRestIT.java
@@ -408,7 +408,9 @@ public void testCreateAnomalyDetectorWithCustomResultIndex() throws IOException
         Assert
             .assertTrue(
                 "got " + exception.getMessage(),
-                exception.getMessage().contains("no permissions for [indices:admin/aliases, indices:admin/create]")
+                exception.getMessage().contains("indices:admin/aliases")
+                    && exception.getMessage().contains("indices:admin/create")
+                    && exception.getMessage().contains("no permissions for")
             );
 
         // User cat has permission to create index
diff --git a/src/test/java/org/opensearch/ad/transport/ValidateAnomalyDetectorTransportActionTests.java b/src/test/java/org/opensearch/ad/transport/ValidateAnomalyDetectorTransportActionTests.java
@@ -222,7 +222,7 @@ public void testValidateAnomalyDetectorWithInvalidFeatureField() throws IOExcept
     }
 
     @Test
-    public void testValidateAnomalyDetectorWithUnknownFeatureField() throws IOException {
+    public void testValidateAnomalyDetectorWithInvalidFeatureDueToTimeSeriesException() throws IOException {
         AggregationBuilder aggregationBuilder = TestHelpers.parseAggregation("{\"test\":{\"terms\":{\"field\":\"type\"}}}");
         AnomalyDetector anomalyDetector = TestHelpers
             .randomAnomalyDetector(
@@ -245,7 +245,7 @@ public void testValidateAnomalyDetectorWithUnknownFeatureField() throws IOExcept
         assertNotNull(response.getIssue());
         assertEquals(ValidationIssueType.FEATURE_ATTRIBUTES, response.getIssue().getType());
         assertEquals(ValidationAspect.DETECTOR, response.getIssue().getAspect());
-        assertTrue(response.getIssue().getMessage().contains(CommonMessages.UNKNOWN_SEARCH_QUERY_EXCEPTION_MSG));
+        assertTrue(response.getIssue().getMessage().contains(CommonMessages.FEATURE_WITH_INVALID_QUERY_MSG));
         assertTrue(response.getIssue().getSubIssues().containsKey(nameField));
     }
 
