test

Signed-off-by: Kaituo Li <[email protected]>

build.gradle

@@ -139,7 +139,7 @@ dependencies {
     implementation group: 'io.protostuff', name: 'protostuff-runtime', version: '1.8.0'
     implementation group: 'io.protostuff', name: 'protostuff-api', version: '1.8.0'
     implementation group: 'io.protostuff', name: 'protostuff-collectionschema', version: '1.8.0'
-    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.12.0'
+    implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.13.0'
 
     implementation "org.jacoco:org.jacoco.agent:0.8.5"
     implementation ("org.jacoco:org.jacoco.ant:0.8.5") {
@@ -403,6 +403,7 @@ testClusters.integTest {
                 @Override
                 File getAsFile() {
                     return configurations.zipArchive.asFileTree.getSingleFile()
+                    //return fileTree("src/test/resources/job-scheduler").getSingleFile()
                 }
             }
         }

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

@@ -0,0 +1,46 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.ad;
+
+import org.opensearch.ad.constant.ADCommonName;
+import org.opensearch.ad.model.AnomalyDetector;
+import org.opensearch.ad.model.AnomalyResult;
+import org.opensearch.ad.settings.ADNumericSetting;
+import org.opensearch.ad.transport.ADEntityProfileAction;
+import org.opensearch.client.Client;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.timeseries.AnalysisType;
+import org.opensearch.timeseries.EntityProfileRunner;
+import org.opensearch.timeseries.util.SecurityClientUtil;
+
+public class ADEntityProfileRunner extends EntityProfileRunner<ADEntityProfileAction> {
+
+    public ADEntityProfileRunner(
+        Client client,
+        SecurityClientUtil clientUtil,
+        NamedXContentRegistry xContentRegistry,
+        long requiredSamples
+    ) {
+        super(
+            client,
+            clientUtil,
+            xContentRegistry,
+            requiredSamples,
+            AnomalyDetector::parse,
+            ADNumericSetting.maxCategoricalFields(),
+            AnalysisType.AD,
+            ADEntityProfileAction.INSTANCE,
+            ADCommonName.ANOMALY_RESULT_INDEX_ALIAS,
+            AnomalyResult.DETECTOR_ID_FIELD
+        );
+    }
+}

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

@@ -0,0 +1,94 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ad;
+
+import java.time.Instant;
+import java.util.List;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.ad.indices.ADIndex;
+import org.opensearch.ad.indices.ADIndexManagement;
+import org.opensearch.ad.model.ADTask;
+import org.opensearch.ad.model.ADTaskType;
+import org.opensearch.ad.model.AnomalyResult;
+import org.opensearch.ad.settings.AnomalyDetectorSettings;
+import org.opensearch.ad.task.ADTaskCacheManager;
+import org.opensearch.ad.task.ADTaskManager;
+import org.opensearch.ad.transport.ADProfileAction;
+import org.opensearch.ad.transport.AnomalyResultAction;
+import org.opensearch.ad.transport.AnomalyResultRequest;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.jobscheduler.spi.LockModel;
+import org.opensearch.jobscheduler.spi.utils.LockService;
+import org.opensearch.timeseries.AnalysisType;
+import org.opensearch.timeseries.JobProcessor;
+import org.opensearch.timeseries.TimeSeriesAnalyticsPlugin;
+import org.opensearch.timeseries.common.exception.EndRunException;
+import org.opensearch.timeseries.model.Config;
+import org.opensearch.timeseries.model.Job;
+import org.opensearch.timeseries.transport.ResultRequest;
+
+public class ADJobProcessor extends
+    JobProcessor<ADIndex, ADIndexManagement, ADTaskCacheManager, ADTaskType, ADTask, ADTaskManager, AnomalyResult, ADProfileAction, ExecuteADResultResponseRecorder> {
+
+    private static final Logger log = LogManager.getLogger(ADJobProcessor.class);
+
+    private static ADJobProcessor INSTANCE;
+
+    public static ADJobProcessor getInstance() {
+        if (INSTANCE != null) {
+            return INSTANCE;
+        }
+        synchronized (JobProcessor.class) {
+            if (INSTANCE != null) {
+                return INSTANCE;
+            }
+            INSTANCE = new ADJobProcessor();
+            return INSTANCE;
+        }
+    }
+
+    private ADJobProcessor() {
+        // Singleton class, use getJobRunnerInstance method instead of constructor
+        super(AnalysisType.AD, TimeSeriesAnalyticsPlugin.AD_THREAD_POOL_NAME, AnomalyResultAction.INSTANCE);
+    }
+
+    public void registerSettings(Settings settings) {
+        super.registerSettings(settings, AnomalyDetectorSettings.AD_MAX_RETRY_FOR_END_RUN_EXCEPTION);
+    }
+
+    @Override
+    protected ResultRequest createResultRequest(String configId, long start, long end) {
+        return new AnomalyResultRequest(configId, start, end);
+    }
+
+    @Override
+    protected void validateResultIndexAndRunJob(Job jobParameter, LockService lockService,
+            LockModel lock,
+            Instant executionStartTime,
+            Instant executionEndTime,
+            String configId,
+            String user,
+            List<String> roles,
+            ExecuteADResultResponseRecorder recorder,
+            Config detector) {
+        String resultIndex = jobParameter.getCustomResultIndex();
+        if (resultIndex == null) {
+            runJob(jobParameter, lockService, lock, executionStartTime, executionEndTime, configId, user, roles, recorder, detector);
+            return;
+        }
+        ActionListener<Boolean> listener = ActionListener.wrap(r -> { log.debug("Custom index is valid"); }, e -> {
+            Exception exception = new EndRunException(configId, e.getMessage(), false);
+            handleException(jobParameter, lockService, lock, executionStartTime, executionEndTime, exception, recorder, detector);
+        });
+        indexManagement.validateCustomIndexForBackendJob(resultIndex, configId, user, roles, () -> {
+            listener.onResponse(true);
+            runJob(jobParameter, lockService, lock, executionStartTime, executionEndTime, configId, user, roles, recorder, detector);
+        }, listener);
+    }
+}

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.ad;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.ad.model.ADTask;
+import org.opensearch.ad.model.ADTaskProfile;
+import org.opensearch.ad.transport.ADTaskProfileAction;
+import org.opensearch.ad.transport.ADTaskProfileNodeResponse;
+import org.opensearch.ad.transport.ADTaskProfileRequest;
+import org.opensearch.client.Client;
+import org.opensearch.core.action.ActionListener;
+import org.opensearch.timeseries.TaskProfileRunner;
+import org.opensearch.timeseries.cluster.HashRing;
+import org.opensearch.timeseries.model.EntityTaskProfile;
+
+public class ADTaskProfileRunner implements TaskProfileRunner<ADTask, ADTaskProfile> {
+    public final Logger logger = LogManager.getLogger(ADTaskProfileRunner.class);
+
+    private final HashRing hashRing;
+    private final Client client;
+
+    public ADTaskProfileRunner(HashRing hashRing, Client client) {
+        this.hashRing = hashRing;
+        this.client = client;
+    }
+
+    @Override
+    public void getTaskProfile(ADTask configLevelTask, ActionListener<ADTaskProfile> listener) {
+        String detectorId = configLevelTask.getConfigId();
+
+        hashRing.getAllEligibleDataNodesWithKnownVersion(dataNodes -> {
+            ADTaskProfileRequest adTaskProfileRequest = new ADTaskProfileRequest(detectorId, dataNodes);
+            client.execute(ADTaskProfileAction.INSTANCE, adTaskProfileRequest, ActionListener.wrap(response -> {
+                if (response.hasFailures()) {
+                    listener.onFailure(response.failures().get(0));
+                    return;
+                }
+
+                List<EntityTaskProfile> adEntityTaskProfiles = new ArrayList<>();
+                ADTaskProfile detectorTaskProfile = new ADTaskProfile(configLevelTask);
+                for (ADTaskProfileNodeResponse node : response.getNodes()) {
+                    ADTaskProfile taskProfile = node.getAdTaskProfile();
+                    if (taskProfile != null) {
+                        if (taskProfile.getNodeId() != null) {
+                            // HC detector: task profile from coordinating node
+                            // Single entity detector: task profile from worker node
+                            detectorTaskProfile.setTaskId(taskProfile.getTaskId());
+                            detectorTaskProfile.setShingleSize(taskProfile.getShingleSize());
+                            detectorTaskProfile.setRcfTotalUpdates(taskProfile.getRcfTotalUpdates());
+                            detectorTaskProfile.setThresholdModelTrained(taskProfile.getThresholdModelTrained());
+                            detectorTaskProfile.setThresholdModelTrainingDataSize(taskProfile.getThresholdModelTrainingDataSize());
+                            detectorTaskProfile.setModelSizeInBytes(taskProfile.getModelSizeInBytes());
+                            detectorTaskProfile.setNodeId(taskProfile.getNodeId());
+                            detectorTaskProfile.setTotalEntitiesCount(taskProfile.getTotalEntitiesCount());
+                            detectorTaskProfile.setDetectorTaskSlots(taskProfile.getDetectorTaskSlots());
+                            detectorTaskProfile.setPendingEntitiesCount(taskProfile.getPendingEntitiesCount());
+                            detectorTaskProfile.setRunningEntitiesCount(taskProfile.getRunningEntitiesCount());
+                            detectorTaskProfile.setRunningEntities(taskProfile.getRunningEntities());
+                            detectorTaskProfile.setTaskType(taskProfile.getTaskType());
+                        }
+                        if (taskProfile.getEntityTaskProfiles() != null) {
+                            adEntityTaskProfiles.addAll(taskProfile.getEntityTaskProfiles());
+                        }
+                    }
+                }
+                if (adEntityTaskProfiles != null && adEntityTaskProfiles.size() > 0) {
+                    detectorTaskProfile.setEntityTaskProfiles(adEntityTaskProfiles);
+                }
+                listener.onResponse(detectorTaskProfile);
+            }, e -> {
+                logger.error("Failed to get task profile for task " + configLevelTask.getTaskId(), e);
+                listener.onFailure(e);
+            }));
+        }, listener);
+
+    }
+
+}