[Extensions] Increasing AD Extension Request timeout to 20 and fixes exception handling for Delete/Profile

src/main/java/org/opensearch/ad/AnomalyDetectorExtension.java

@@ -249,7 +249,14 @@ public Collection<Object> createComponents(ExtensionsRunner runner) {
 
         Throttler throttler = new Throttler(getClock());
         ClientUtil clientUtil = new ClientUtil(environmentSettings, restClient(), throttler);
-        IndexUtils indexUtils = new IndexUtils(restClient(), clientUtil, sdkClusterService, indexNameExpressionResolver, javaAsyncClient());
+        IndexUtils indexUtils = new IndexUtils(
+            restClient(),
+            clientUtil,
+            sdkClusterService,
+            indexNameExpressionResolver,
+            javaAsyncClient(),
+            environmentSettings
+        );
         nodeFilter = new DiscoveryNodeFilterer(sdkClusterService);
         AnomalyDetectionIndices anomalyDetectionIndices = new AnomalyDetectionIndices(
             sdkRestClient,
@@ -342,6 +349,7 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
         CheckpointDao checkpoint = new CheckpointDao(
             sdkRestClient,
             sdkJavaAsyncClient,
+            environmentSettings,
             clientUtil,
             CommonName.CHECKPOINT_INDEX_NAME,
             gson,

src/main/java/org/opensearch/ad/AnomalyDetectorJobRunner.java

@@ -652,7 +652,9 @@ public void onFailure(Exception exception) {
             }
 
         });
-        Response response = acquireLockResponse.orTimeout(15L, TimeUnit.SECONDS).join();
+        Response response = acquireLockResponse
+            .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+            .join();
 
         log.info("Acquired lock for AD job {}", context.getJobId());
 
@@ -685,7 +687,9 @@ public void onFailure(Exception exception) {
                 }
 
             });
-            Response response = releaseLockResponse.orTimeout(15L, TimeUnit.SECONDS).join();
+            Response response = releaseLockResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .join();
 
             boolean lockIsReleased = RestStatus.fromCode(response.getStatusLine().getStatusCode()) == RestStatus.OK ? true : false;
             if (lockIsReleased) {

src/main/java/org/opensearch/ad/indices/AnomalyDetectionIndices.java

@@ -63,6 +63,7 @@
 import org.opensearch.ad.model.AnomalyDetectorJob;
 import org.opensearch.ad.model.AnomalyResult;
 import org.opensearch.ad.rest.handler.AnomalyDetectorFunction;
+import org.opensearch.ad.settings.AnomalyDetectorSettings;
 import org.opensearch.ad.util.DiscoveryNodeFilterer;
 import org.opensearch.client.indices.CreateIndexRequest;
 import org.opensearch.client.indices.CreateIndexResponse;
@@ -121,6 +122,7 @@ public class AnomalyDetectionIndices implements LocalNodeMasterListener {
     private final SDKRestClient adminClient;
     private final OpenSearchAsyncClient sdkJavaAsyncClient;
     private final ThreadPool threadPool;
+    private final Settings environmentSettings;
 
     private volatile TimeValue historyRolloverPeriod;
     private volatile Long historyMaxDocs;
@@ -188,6 +190,7 @@ public AnomalyDetectionIndices(
         this.sdkJavaAsyncClient = sdkJavaAsyncClient;
         this.sdkClusterService = sdkClusterService;
         this.threadPool = threadPool;
+        this.environmentSettings = settings;
         // FIXME Implement this
         // https://github.com/opensearch-project/opensearch-sdk-java/issues/423
         // this.clusterService.addLocalNodeMasterListener(this);
@@ -1098,7 +1101,9 @@ private void updateJobIndexSettingIfNecessary(IndexState jobIndexState, ActionLi
 
         GetIndicesSettingsResponse settingResponse;
         try {
-            settingResponse = getIndicesSettingsResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            settingResponse = getIndicesSettingsResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(environmentSettings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
             // auto expand setting is a range string like "1-all"
             org.opensearch.client.opensearch.indices.IndexState indexState = settingResponse.get(ADIndex.JOB.getIndexName());
             String autoExpandReplica = indexState.settings().autoExpandReplicas();

src/main/java/org/opensearch/ad/ml/CheckpointDao.java

@@ -56,13 +56,15 @@
 import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.AnomalyDetectionIndices;
 import org.opensearch.ad.model.Entity;
+import org.opensearch.ad.settings.AnomalyDetectorSettings;
 import org.opensearch.ad.util.ClientUtil;
 import org.opensearch.client.opensearch.OpenSearchAsyncClient;
 import org.opensearch.client.opensearch._types.BulkIndexByScrollFailure;
 import org.opensearch.client.opensearch._types.Conflicts;
 import org.opensearch.client.opensearch._types.ExpandWildcard;
 import org.opensearch.client.opensearch.core.DeleteByQueryRequest;
 import org.opensearch.client.opensearch.core.DeleteByQueryResponse;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.sdk.SDKClient.SDKRestClient;
 
@@ -114,6 +116,7 @@ public class CheckpointDao {
 
     // configuration
     private final String indexName;
+    private final Settings settings;
 
     private Gson gson;
     private RandomCutForestMapper mapper;
@@ -146,6 +149,7 @@ public class CheckpointDao {
      *
      * @param client ES search client
      * @param sdkJavaAsyncClient OpenSearch Async Client
+     * @param settings Environment Settings
      * @param clientUtil utility with ES client
      * @param indexName name of the index for model checkpoints
      * @param gson accessor to Gson functionality
@@ -163,6 +167,7 @@ public class CheckpointDao {
     public CheckpointDao(
         SDKRestClient client,
         OpenSearchAsyncClient sdkJavaAsyncClient,
+        Settings settings,
         ClientUtil clientUtil,
         String indexName,
         Gson gson,
@@ -179,6 +184,7 @@ public CheckpointDao(
     ) {
         this.client = client;
         this.sdkJavaAsyncClient = sdkJavaAsyncClient;
+        this.settings = settings;
         this.clientUtil = clientUtil;
         this.indexName = indexName;
         this.gson = gson;
@@ -452,7 +458,9 @@ public void deleteModelCheckpointByDetectorId(String detectorID) {
         logger.info("Delete checkpoints of detector {}", detectorID);
         try {
             CompletableFuture<DeleteByQueryResponse> deleteResponse = sdkJavaAsyncClient.deleteByQuery(deleteRequest);
-            DeleteByQueryResponse response = deleteResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            DeleteByQueryResponse response = deleteResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
             if (response.timedOut() || !response.failures().isEmpty()) {
                 logFailure(response, detectorID);
             }

src/main/java/org/opensearch/ad/rest/RestAnomalyDetectorJobAction.java

@@ -215,7 +215,9 @@ public void onFailure(Exception exception) {
 
             });
 
-            Response response = registerJobDetailsResponse.orTimeout(15, TimeUnit.SECONDS).join();
+            Response response = registerJobDetailsResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .join();
             this.registeredJobDetails = RestStatus.fromCode(response.getStatusLine().getStatusCode()) == RestStatus.OK ? true : false;
             LOG.info("Job Details Registered : " + registeredJobDetails);
         }

src/main/java/org/opensearch/ad/settings/LegacyOpenDistroAnomalyDetectorSettings.java

@@ -54,10 +54,11 @@ private LegacyOpenDistroAnomalyDetectorSettings() {}
             Setting.Property.Deprecated
         );
 
+    // TODO : Revert setting value back to 10 seconds once there is support for more targeted cluster state requests
     public static final Setting<TimeValue> REQUEST_TIMEOUT = Setting
         .positiveTimeSetting(
             "opendistro.anomaly_detection.request_timeout",
-            TimeValue.timeValueSeconds(10),
+            TimeValue.timeValueSeconds(20),
             Setting.Property.NodeScope,
             Setting.Property.Dynamic,
             Setting.Property.Deprecated

src/main/java/org/opensearch/ad/task/ADTaskManager.java

@@ -121,6 +121,7 @@
 import org.opensearch.ad.model.Entity;
 import org.opensearch.ad.rest.handler.AnomalyDetectorFunction;
 import org.opensearch.ad.rest.handler.IndexAnomalyDetectorJobActionHandler;
+import org.opensearch.ad.settings.AnomalyDetectorSettings;
 import org.opensearch.ad.transport.ADBatchAnomalyResultAction;
 import org.opensearch.ad.transport.ADBatchAnomalyResultRequest;
 import org.opensearch.ad.transport.ADCancelTaskAction;
@@ -212,6 +213,8 @@ public class ADTaskManager {
     private static int DEFAULT_MAINTAIN_INTERVAL_IN_SECONDS = 5;
     private final Semaphore checkingTaskSlot;
 
+    private final Settings settings;
+
     private volatile Integer maxAdBatchTaskPerNode;
     private volatile Integer maxRunningEntitiesPerDetector;
 
@@ -238,6 +241,7 @@ public ADTaskManager(
         this.nodeFilter = nodeFilter;
         this.sdkClusterService = sdkClusterService;
         this.adTaskCacheManager = adTaskCacheManager;
+        this.settings = settings;
         /* MultiNode support https://github.com/opensearch-project/opensearch-sdk-java/issues/200 */
         // this.hashRing = hashRing;
 
@@ -1377,7 +1381,9 @@ private void resetEntityTasksAsStopped(String detectorTaskId) {
 
         try {
             CompletableFuture<UpdateByQueryResponse> updateByQueryResponse = sdkJavaAsyncClient.updateByQuery(updateByQueryRequest.build());
-            UpdateByQueryResponse queryResponse = updateByQueryResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            UpdateByQueryResponse queryResponse = updateByQueryResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
             List<BulkIndexByScrollFailure> bulkFailures = queryResponse.failures();
             if (isNullOrEmpty(bulkFailures)) {
                 logger.debug("Updated {} child entity tasks state for detector task {}", queryResponse.updated(), detectorTaskId);
@@ -1573,7 +1579,9 @@ private void updateLatestFlagOfOldTasksAndCreateNewTask(
 
         try {
             CompletableFuture<UpdateByQueryResponse> updateByQueryResponse = sdkJavaAsyncClient.updateByQuery(updateByQueryRequest.build());
-            UpdateByQueryResponse queryResponse = updateByQueryResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            UpdateByQueryResponse queryResponse = updateByQueryResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
 
             List<BulkIndexByScrollFailure> bulkFailures = queryResponse.failures();
             if (isNullOrEmpty(bulkFailures)) {
@@ -1776,7 +1784,9 @@ protected <T> void deleteTaskDocs(
                 BulkRequest bulkRequest = new BulkRequest.Builder().operations(operations).build();
                 try {
                     CompletableFuture<BulkResponse> bulkResponse = sdkJavaAsyncClient.bulk(bulkRequest);
-                    BulkResponse res = bulkResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+                    BulkResponse res = bulkResponse
+                        .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                        .get();
 
                     logger.info("Old AD tasks deleted for detector {}", detectorId);
                     List<BulkResponseItem> bulkItemResponses = res.items();
@@ -1833,7 +1843,9 @@ public void cleanChildTasksAndADResultsOfDeletedTask() {
                 .build();
             try {
                 CompletableFuture<DeleteByQueryResponse> deleteADResultsResponse = sdkJavaAsyncClient.deleteByQuery(deleteADResultsRequest);
-                DeleteByQueryResponse res = deleteADResultsResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+                DeleteByQueryResponse res = deleteADResultsResponse
+                    .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                    .get();
 
                 logger.debug("Successfully deleted AD results of task " + taskId);
                 DeleteByQueryRequest deleteChildTasksRequest = new DeleteByQueryRequest.Builder()
@@ -1845,7 +1857,9 @@ public void cleanChildTasksAndADResultsOfDeletedTask() {
                 try {
                     CompletableFuture<DeleteByQueryResponse> deleteChildTasksResponse = sdkJavaAsyncClient
                         .deleteByQuery(deleteChildTasksRequest);
-                    DeleteByQueryResponse deleteByQueryResponse = deleteChildTasksResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+                    DeleteByQueryResponse deleteByQueryResponse = deleteChildTasksResponse
+                        .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                        .get();
 
                     logger.debug("Successfully deleted child tasks of task " + taskId);
                     cleanChildTasksAndADResultsOfDeletedTask();
@@ -2014,7 +2028,9 @@ public void deleteADTasks(String detectorId, AnomalyDetectorFunction function, A
         request.query(q -> q.bool(query.build()));
         try {
             CompletableFuture<DeleteByQueryResponse> deleteByQueryResponse = sdkJavaAsyncClient.deleteByQuery(request.build());
-            DeleteByQueryResponse response = deleteByQueryResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            DeleteByQueryResponse response = deleteByQueryResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
 
             if (response.failures() == null || response.failures().size() == 0) {
                 logger.info("AD tasks deleted for detector {}", detectorId);
@@ -2023,11 +2039,13 @@ public void deleteADTasks(String detectorId, AnomalyDetectorFunction function, A
             } else {
                 listener.onFailure(new OpenSearchStatusException("Failed to delete all AD tasks", RestStatus.INTERNAL_SERVER_ERROR));
             }
-        } catch (IndexNotFoundException e) {
-            deleteADResultOfDetector(detectorId);
-            function.execute();
         } catch (Exception e) {
-            listener.onFailure(e);
+            if (e.getMessage().contains("index_not_found_exception")) {
+                deleteADResultOfDetector(detectorId);
+                function.execute();
+            } else {
+                listener.onFailure(e);
+            }
         }
     }
 
@@ -2045,7 +2063,9 @@ private void deleteADResultOfDetector(String detectorId) {
         try {
             CompletableFuture<DeleteByQueryResponse> deleteADResultsResponse = sdkJavaAsyncClient
                 .deleteByQuery(deleteADResultsRequest.build());
-            DeleteByQueryResponse deleteByQueryResponse = deleteADResultsResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            DeleteByQueryResponse deleteByQueryResponse = deleteADResultsResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
             logger.debug("Successfully deleted AD results of detector " + detectorId);
         } catch (Exception exception) {
             logger.error("Failed to delete AD results of detector " + detectorId, exception);
@@ -3037,7 +3057,9 @@ public void resetLatestFlagAsFalse(List<ADTask> adTasks) {
 
         try {
             CompletableFuture<BulkResponse> bulkResponse = sdkJavaAsyncClient.bulk(bulkRequest);
-            BulkResponse res = bulkResponse.orTimeout(10L, TimeUnit.SECONDS).get();
+            BulkResponse res = bulkResponse
+                .orTimeout(AnomalyDetectorSettings.REQUEST_TIMEOUT.get(settings).getMillis(), TimeUnit.MILLISECONDS)
+                .get();
 
             List<BulkResponseItem> bulkItemResponses = res.items();
             if (bulkItemResponses != null && bulkItemResponses.size() > 0) {

src/main/java/org/opensearch/ad/transport/DeleteAnomalyDetectorTransportAction.java

@@ -44,7 +44,6 @@
 import org.opensearch.ad.util.RestHandlerUtils;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.xcontent.XContentParser;
-import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.rest.RestStatus;
 import org.opensearch.sdk.ExtensionsRunner;
 import org.opensearch.sdk.SDKClient.SDKRestClient;
@@ -147,7 +146,7 @@ private void deleteAnomalyDetectorJobDoc(String detectorId, ActionListener<Delet
             }
         }, exception -> {
             LOG.error("Failed to delete AD job for " + detectorId, exception);
-            if (exception instanceof IndexNotFoundException) {
+            if (exception.getMessage().contains("index_not_found_exception")) {
                 deleteDetectorStateDoc(detectorId, listener);
             } else {
                 LOG.error("Failed to delete anomaly detector job", exception);
@@ -163,7 +162,7 @@ private void deleteDetectorStateDoc(String detectorId, ActionListener<DeleteResp
             // whether deleted state doc or not, continue as state doc may not exist
             deleteAnomalyDetectorDoc(detectorId, listener);
         }, exception -> {
-            if (exception instanceof IndexNotFoundException) {
+            if (exception.getMessage().contains("index_not_found_exception")) {
                 deleteAnomalyDetectorDoc(detectorId, listener);
             } else {
                 LOG.error("Failed to delete detector state", exception);

src/main/java/org/opensearch/ad/util/ExceptionUtil.java

@@ -19,7 +19,6 @@
 import org.opensearch.ExceptionsHelper;
 import org.opensearch.OpenSearchException;
 import org.opensearch.action.ActionListener;
-import org.opensearch.action.NoShardAvailableActionException;
 import org.opensearch.action.UnavailableShardsException;
 import org.opensearch.action.index.IndexResponse;
 import org.opensearch.action.support.replication.ReplicationResponse;
@@ -28,7 +27,6 @@
 import org.opensearch.ad.common.exception.LimitExceededException;
 import org.opensearch.common.io.stream.NotSerializableExceptionWrapper;
 import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException;
-import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.rest.RestStatus;
 
 public class ExceptionUtil {
@@ -186,6 +184,6 @@ public static boolean isIndexNotAvailable(Exception e) {
         if (e == null) {
             return false;
         }
-        return e instanceof IndexNotFoundException || e instanceof NoShardAvailableActionException;
+        return (e.getMessage().contains("index_not_found_exception")) || (e.getMessage().contains("no_shard_available_action_exception"));
     }
 }