test

Signed-off-by: Kaituo Li <kaituo@amazon.com>
kaituo · Jun 5, 2024 · 10da2a7 · 10da2a7
1 parent 9eed2ff
commit 10da2a7
Show file tree

Hide file tree

Showing 67 changed files with 987 additions and 403 deletions.
diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml
@@ -52,7 +52,7 @@ jobs:
                 -Dtests.seed=B4BA12CCF1D9E825 -Dtests.security.manager=false \
                 -Dtests.jvm.argline='-XX:TieredStopAtLevel=1 -XX:ReservedCodeCacheSize=64m' \
                 -Dtests.locale=ar-JO -Dtests.timezone=Asia/Samarkand -Dmodel-benchmark=true \
-                -Dtests.timeoutSuite=3600000!  -Dtests.logs=true"
+                -Dtests.timeoutSuite=3600000!  -Dtest.logs=true"
               ;;
             hc)
               su `id -un 1000` -c "./gradlew ':test' --tests 'org.opensearch.ad.ml.HCADModelPerfTests' \

diff --git a/build.gradle b/build.gradle
@@ -695,8 +695,8 @@ List<String> jacocoExclusions = [
         'org.opensearch.ad.task.ADTaskManager',
         // TODO: add forecast test coverage before release
         'org.opensearch.forecast.*',
-        'org.opensearch.timeseries.*',
-        'org.opensearch.ad.*',
+        //'org.opensearch.timeseries.*',
+        //'org.opensearch.ad.*',
 ]
 
 

diff --git a/src/main/java/org/opensearch/ad/ADJobProcessor.java b/src/main/java/org/opensearch/ad/ADJobProcessor.java
@@ -81,7 +81,7 @@ protected void validateResultIndexAndRunJob(
         ExecuteADResultResponseRecorder recorder,
         Config detector
     ) {
-        String resultIndex = jobParameter.getCustomResultIndex();
+        String resultIndex = jobParameter.getCustomResultIndexOrAlias();
         if (resultIndex == null) {
             runJob(jobParameter, lockService, lock, executionStartTime, executionEndTime, configId, user, roles, recorder, detector);
             return;

diff --git a/src/main/java/org/opensearch/ad/indices/ADIndex.java b/src/main/java/org/opensearch/ad/indices/ADIndex.java
@@ -37,7 +37,8 @@ public enum ADIndex implements TimeSeriesIndex {
         false,
         ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getCheckpointMappings)
     ),
-    STATE(ADCommonName.DETECTION_STATE_INDEX, false, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getStateMappings));
+    STATE(ADCommonName.DETECTION_STATE_INDEX, false, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getStateMappings)),
+    CUSTOM_RESULT(CUSTOM_RESULT_INDEX, true, ThrowingSupplierWrapper.throwingSupplierWrapper(ADIndexManagement::getResultMappings)),;
 
     private final String indexName;
     // whether we use an alias for the index
@@ -64,10 +65,4 @@ public boolean isAlias() {
     public String getMapping() {
         return mapping;
     }
-
-    @Override
-    public boolean isJobIndex() {
-        return CommonName.JOB_INDEX.equals(indexName);
-    }
-
 }
diff --git a/src/main/java/org/opensearch/ad/indices/ADIndexManagement.java b/src/main/java/org/opensearch/ad/indices/ADIndexManagement.java
@@ -30,12 +30,14 @@
 import org.opensearch.action.delete.DeleteRequest;
 import org.opensearch.action.index.IndexRequest;
 import org.opensearch.ad.constant.ADCommonName;
+import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.ad.model.AnomalyResult;
 import org.opensearch.client.Client;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.xcontent.XContentType;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.core.xcontent.ToXContent;
 import org.opensearch.core.xcontent.XContentBuilder;
 import org.opensearch.threadpool.ThreadPool;
@@ -64,6 +66,7 @@ public class ADIndexManagement extends IndexManagement<ADIndex> {
      * @param settings       OS cluster setting
      * @param nodeFilter     Used to filter eligible nodes to host AD indices
      * @param maxUpdateRunningTimes max number of retries to update index mapping and setting
+     * @param xContentRegistry registry for json parser
      * @throws IOException
      */
     public ADIndexManagement(
@@ -72,7 +75,8 @@ public ADIndexManagement(
         ThreadPool threadPool,
         Settings settings,
         DiscoveryNodeFilterer nodeFilter,
-        int maxUpdateRunningTimes
+        int maxUpdateRunningTimes,
+        NamedXContentRegistry xContentRegistry
     )
         throws IOException {
         super(
@@ -87,7 +91,9 @@ public ADIndexManagement(
             AD_RESULT_HISTORY_ROLLOVER_PERIOD.get(settings),
             AD_RESULT_HISTORY_MAX_DOCS_PER_SHARD.get(settings),
             AD_RESULT_HISTORY_RETENTION_PERIOD.get(settings),
-            ADIndex.RESULT.getMapping()
+            ADIndex.RESULT.getMapping(),
+            xContentRegistry,
+            AnomalyDetector::parse
         );
         this.clusterService.addLocalNodeClusterManagerListener(this);
 
@@ -181,7 +187,7 @@ public void initDefaultResultIndexDirectly(ActionListener<CreateIndexResponse> a
             AD_RESULT_HISTORY_INDEX_PATTERN,
             ADCommonName.ANOMALY_RESULT_INDEX_ALIAS,
             true,
-            AD_RESULT_HISTORY_INDEX_PATTERN,
+            true,
             ADIndex.RESULT,
             actionListener
         );
@@ -270,6 +276,6 @@ protected DeleteRequest createDummyDeleteRequest(String resultIndex) throws IOEx
 
     @Override
     public void initCustomResultIndexDirectly(String resultIndex, ActionListener<CreateIndexResponse> actionListener) {
-        initResultIndexDirectly(resultIndex, null, false, AD_RESULT_HISTORY_INDEX_PATTERN, ADIndex.RESULT, actionListener);
+        initResultIndexDirectly(getCustomResultIndexPattern(resultIndex), resultIndex, false, false, ADIndex.RESULT, actionListener);
     }
 }
diff --git a/src/main/java/org/opensearch/ad/model/ADTask.java b/src/main/java/org/opensearch/ad/model/ADTask.java
@@ -336,7 +336,7 @@ public static ADTask parse(XContentParser parser, String taskId) throws IOExcept
                 detector.getLastUpdateTime(),
                 detector.getCategoryFields(),
                 detector.getUser(),
-                detector.getCustomResultIndex(),
+                detector.getCustomResultIndexOrAlias(),
                 detector.getImputationOption(),
                 detector.getRecencyEmphasis(),
                 detector.getSeasonIntervals(),

diff --git a/src/main/java/org/opensearch/ad/model/AnomalyDetector.java b/src/main/java/org/opensearch/ad/model/AnomalyDetector.java
@@ -265,7 +265,7 @@ public AnomalyDetector(StreamInput input) throws IOException {
         } else {
             this.uiMetadata = null;
         }
-        customResultIndex = input.readOptionalString();
+        customResultIndexOrAlias = input.readOptionalString();
         if (input.readBoolean()) {
             this.imputationOption = new ImputationOption(input);
         } else {
@@ -326,7 +326,7 @@ public void writeTo(StreamOutput output) throws IOException {
         } else {
             output.writeBoolean(false);
         }
-        output.writeOptionalString(customResultIndex);
+        output.writeOptionalString(customResultIndexOrAlias);
         if (imputationOption != null) {
             output.writeBoolean(true);
             imputationOption.writeTo(output);

diff --git a/src/main/java/org/opensearch/ad/ratelimit/ADSaveResultStrategy.java b/src/main/java/org/opensearch/ad/ratelimit/ADSaveResultStrategy.java
@@ -87,7 +87,7 @@ public void saveResult(AnomalyResult result, Config config) {
                     config.getId(),
                     result.getAnomalyGrade() > 0 ? RequestPriority.HIGH : RequestPriority.MEDIUM,
                     result,
-                    config.getCustomResultIndex()
+                    config.getCustomResultIndexOrAlias()
                 )
             );
     }

diff --git a/src/main/java/org/opensearch/ad/rest/AbstractADSearchAction.java b/src/main/java/org/opensearch/ad/rest/AbstractADSearchAction.java
@@ -13,7 +13,7 @@
 import org.opensearch.ad.constant.ADCommonMessages;
 import org.opensearch.ad.settings.ADEnabledSetting;
 import org.opensearch.core.xcontent.ToXContentObject;
-import org.opensearch.timeseries.AbstractSearchAction;
+import org.opensearch.timeseries.rest.AbstractSearchAction;
 
 public abstract class AbstractADSearchAction<T extends ToXContentObject> extends AbstractSearchAction<T> {
 

diff --git a/src/main/java/org/opensearch/ad/rest/handler/AbstractAnomalyDetectorActionHandler.java b/src/main/java/org/opensearch/ad/rest/handler/AbstractAnomalyDetectorActionHandler.java
@@ -231,7 +231,7 @@ protected AnomalyDetector copyConfig(User user, Config config) {
             Instant.now(),
             config.getCategoryFields(),
             user,
-            config.getCustomResultIndex(),
+            config.getCustomResultIndexOrAlias(),
             config.getImputationOption(),
             config.getRecencyEmphasis(),
             config.getSeasonIntervals(),

diff --git a/src/main/java/org/opensearch/ad/task/ADBatchTaskRunner.java b/src/main/java/org/opensearch/ad/task/ADBatchTaskRunner.java
@@ -1158,7 +1158,7 @@ private void detectAnomaly(
             user = adTask.getUser().getName();
             roles = adTask.getUser().getRoles();
         }
-        String resultIndex = adTask.getDetector().getCustomResultIndex();
+        String resultIndex = adTask.getDetector().getCustomResultIndexOrAlias();
 
         if (resultIndex == null) {
             // if result index is null, store anomaly result directly

diff --git a/src/main/java/org/opensearch/ad/task/ADTaskManager.java b/src/main/java/org/opensearch/ad/task/ADTaskManager.java
@@ -571,31 +571,58 @@ protected void scaleTaskLaneOnCoordinatingNode(
         TransportService transportService,
         ActionListener<JobResponse> listener
     ) {
-        DiscoveryNode coordinatingNode = getCoordinatingNode(adTask);
         transportService
             .sendRequest(
-                coordinatingNode,
+                getCoordinatingNode(adTask),
                 ForwardADTaskAction.NAME,
                 new ForwardADTaskRequest(adTask, approvedTaskSlot, ADTaskAction.SCALE_ENTITY_TASK_SLOTS),
                 transportRequestOptions,
                 new ActionListenerResponseHandler<>(listener, JobResponse::new)
             );
     }
 
+    /**
+     * Retrieves the coordinating node for the given ADTask.
+     *
+     * This method looks for a node in the list of eligible data nodes that matches the coordinating node ID
+     * and version specified in the ADTask. The matching criteria are:
+     * 1. The node ID must be equal to the coordinating node ID.
+     * 2. Both node versions must be either null or equal.
+     *
+     * If the coordinating node ID and the local node have different software versions, this method will
+     * throw a ResourceNotFoundException.
+     *
+     * @param adTask the ADTask containing the coordinating node information.
+     * @return a DiscoveryNode containing the matching DiscoveryNode if found, or throws ResourceNotFoundException if no match is found.
+     * The caller is supposed to handle the thrown exception.
+     * @throws ResourceNotFoundException if the coordinating node has a different version than the local node or if the coordinating node is not found.
+     */
     private DiscoveryNode getCoordinatingNode(ADTask adTask) {
-        String coordinatingNode = adTask.getCoordinatingNode();
+        String coordinatingNodeId = adTask.getCoordinatingNode();
+        Version coordinatingNodeVersion = hashRing.getVersion(coordinatingNodeId);
+        Version localNodeVersion = hashRing.getVersion(clusterService.localNode().getId());
+        if (!isSameVersion(coordinatingNodeVersion, localNodeVersion)) {
+            throw new ResourceNotFoundException(adTask.getConfigId(), "AD task coordinating node has different version than local node");
+        }
+
         DiscoveryNode[] eligibleDataNodes = nodeFilter.getEligibleDataNodes();
-        DiscoveryNode targetNode = null;
+
         for (DiscoveryNode node : eligibleDataNodes) {
-            if (node.getId().equals(coordinatingNode)) {
-                targetNode = node;
-                break;
+            String nodeId = node.getId();
+            if (nodeId == null) {
+                continue;
+            }
+
+            if (nodeId.equals(coordinatingNodeId)) {
+                return node;
             }
         }
-        if (targetNode == null) {
-            throw new ResourceNotFoundException(adTask.getConfigId(), "AD task coordinating node not found");
-        }
-        return targetNode;
+
+        throw new ResourceNotFoundException(adTask.getConfigId(), "AD task coordinating node not found");
+    }
+
+    private boolean isSameVersion(Version version1, Version version2) {
+        return (version1 == null && version2 == null) || (version1 != null && version2 != null && version1.compareTo(version2) == 0);
     }
 
     @Override
@@ -791,7 +818,7 @@ public <T> void cleanConfigCache(
         } catch (ResourceNotFoundException e) {
             logger
                 .warn(
-                    "Task coordinating node left cluster, taskId: {}, detectorId: {}, coordinatingNode: {}",
+                    "Task coordinating node left cluster or has different software version, taskId: {}, detectorId: {}, coordinatingNode: {}",
                     taskId,
                     detectorId,
                     coordinatingNode

diff --git a/src/main/java/org/opensearch/ad/transport/ADResultBulkTransportAction.java b/src/main/java/org/opensearch/ad/transport/ADResultBulkTransportAction.java
@@ -68,7 +68,7 @@ public ADResultBulkTransportAction(
     @Override
     protected BulkRequest prepareBulkRequest(float indexingPressurePercent, ADResultBulkRequest request) {
         BulkRequest bulkRequest = new BulkRequest();
-        List<ADResultWriteRequest> results = request.getAnomalyResults();
+        List<ADResultWriteRequest> results = request.getResults();
 
         if (indexingPressurePercent <= softLimit) {
             for (ADResultWriteRequest resultWriteRequest : results) {

diff --git a/src/main/java/org/opensearch/ad/transport/ADSingleStreamResultTransportAction.java b/src/main/java/org/opensearch/ad/transport/ADSingleStreamResultTransportAction.java
@@ -80,7 +80,7 @@ public ADResultWriteRequest createResultWriteRequest(Config config, AnomalyResul
             config.getId(),
             RequestPriority.MEDIUM,
             result,
-            config.getCustomResultIndex()
+            config.getCustomResultIndexOrAlias()
         );
     }
 

diff --git a/src/main/java/org/opensearch/ad/transport/SearchTopAnomalyResultTransportAction.java b/src/main/java/org/opensearch/ad/transport/SearchTopAnomalyResultTransportAction.java
@@ -301,10 +301,10 @@ protected void doExecute(Task task, SearchTopAnomalyResultRequest request, Actio
             SearchRequest searchRequest = generateSearchRequest(request);
 
             // Adding search over any custom result indices
-            String rawCustomResultIndex = getAdResponse.getDetector().getCustomResultIndex();
-            String customResultIndex = rawCustomResultIndex == null ? null : rawCustomResultIndex.trim();
-            if (!Strings.isNullOrEmpty(customResultIndex)) {
-                searchRequest.indices(defaultIndex, customResultIndex);
+            String rawCustomResultIndexPattern = getAdResponse.getDetector().getCustomResultIndexPattern();
+            String customResultIndexPattern = rawCustomResultIndexPattern == null ? null : rawCustomResultIndexPattern.trim();
+            if (!Strings.isNullOrEmpty(customResultIndexPattern)) {
+                searchRequest.indices(defaultIndex, customResultIndexPattern);
             }
 
             // Utilizing the existing search() from SearchHandler to handle security permissions. Both user role
@@ -321,7 +321,7 @@ protected void doExecute(Task task, SearchTopAnomalyResultRequest request, Actio
                         clock.millis() + TOP_ANOMALY_RESULT_TIMEOUT_IN_MILLIS,
                         request.getSize(),
                         orderType,
-                        customResultIndex
+                        customResultIndexPattern
                     )
                 );
 

diff --git a/...ain/java/org/opensearch/ad/transport/handler/ADIndexMemoryPressureAwareResultHandler.java b/...ain/java/org/opensearch/ad/transport/handler/ADIndexMemoryPressureAwareResultHandler.java
@@ -15,6 +15,8 @@
 import org.apache.logging.log4j.Logger;
 import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.ADIndexManagement;
+import org.opensearch.ad.model.AnomalyResult;
+import org.opensearch.ad.ratelimit.ADResultWriteRequest;
 import org.opensearch.ad.transport.ADResultBulkAction;
 import org.opensearch.ad.transport.ADResultBulkRequest;
 import org.opensearch.client.Client;
@@ -27,7 +29,7 @@
 import org.opensearch.timeseries.transport.handler.IndexMemoryPressureAwareResultHandler;
 
 public class ADIndexMemoryPressureAwareResultHandler extends
-    IndexMemoryPressureAwareResultHandler<ADResultBulkRequest, ResultBulkResponse, ADIndex, ADIndexManagement> {
+    IndexMemoryPressureAwareResultHandler<AnomalyResult, ADResultWriteRequest, ADResultBulkRequest, ResultBulkResponse, ADIndex, ADIndexManagement> {
     private static final Logger LOG = LogManager.getLogger(ADIndexMemoryPressureAwareResultHandler.class);
 
     @Inject

diff --git a/src/main/java/org/opensearch/forecast/ForecastJobProcessor.java b/src/main/java/org/opensearch/forecast/ForecastJobProcessor.java
@@ -85,7 +85,7 @@ protected void validateResultIndexAndRunJob(
             Exception exception = new EndRunException(configId, e.getMessage(), false);
             handleException(jobParameter, lockService, lock, executionStartTime, executionEndTime, exception, recorder, detector);
         });
-        String resultIndex = jobParameter.getCustomResultIndex();
+        String resultIndex = jobParameter.getCustomResultIndexOrAlias();
         if (resultIndex == null) {
             indexManagement.validateDefaultResultIndexForBackendJob(configId, user, roles, () -> {
                 listener.onResponse(true);

diff --git a/src/main/java/org/opensearch/forecast/indices/ForecastIndex.java b/src/main/java/org/opensearch/forecast/indices/ForecastIndex.java
@@ -37,7 +37,8 @@ public enum ForecastIndex implements TimeSeriesIndex {
         ForecastCommonName.FORECAST_STATE_INDEX,
         false,
         ThrowingSupplierWrapper.throwingSupplierWrapper(ForecastIndexManagement::getStateMappings)
-    );
+    ),
+    CUSTOM_RESULT(CUSTOM_RESULT_INDEX, true, ThrowingSupplierWrapper.throwingSupplierWrapper(ForecastIndexManagement::getResultMappings));
 
     private final String indexName;
     // whether we use an alias for the index
@@ -64,9 +65,4 @@ public boolean isAlias() {
     public String getMapping() {
         return mapping;
     }
-
-    @Override
-    public boolean isJobIndex() {
-        return CommonName.JOB_INDEX.equals(indexName);
-    }
 }