Refactor ADTask and Related Components (#969)

* Refactor ADTask and Related Components This PR includes several key refactoring changes: - Extracts common code from ADTask into TimeSeriesTask, creating ForecastTask for forecasting-specific logic. - Consolidates common code from ADTaskType into TaskType and introduces ForecastTaskType for forecasting-related purposes. - Renames ADTaskState to TaskState for consistent code reuse. - Renames the method getId in ADTask to getConfigId to differentiate it from other IDs like task id. Testing done: 1. Added unit tests for the new code to ensure functionality. 2. Executed a successful Gradle build. Signed-off-by: Kaituo Li <[email protected]> * add comments and address compiler errors Signed-off-by: Kaituo Li <[email protected]> * address Amit's comments and address compiler failure Signed-off-by: Kaituo Li <[email protected]> --------- Signed-off-by: Kaituo Li <[email protected]>
opensearch-project · Aug 7, 2023 · d4946f0 · d4946f0
1 parent 1130a1b
commit d4946f0
Show file tree

Hide file tree

Showing 49 changed files with 1,545 additions and 692 deletions.
diff --git a/src/main/java/org/opensearch/ad/AnomalyDetectorJobRunner.java b/src/main/java/org/opensearch/ad/AnomalyDetectorJobRunner.java
@@ -32,7 +32,6 @@
 import org.opensearch.action.index.IndexRequest;
 import org.opensearch.action.support.WriteRequest;
 import org.opensearch.ad.indices.ADIndexManagement;
-import org.opensearch.ad.model.ADTaskState;
 import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.ad.settings.AnomalyDetectorSettings;
 import org.opensearch.ad.task.ADTaskManager;
@@ -64,6 +63,7 @@
 import org.opensearch.timeseries.constant.CommonName;
 import org.opensearch.timeseries.function.ExecutorFunction;
 import org.opensearch.timeseries.model.Job;
+import org.opensearch.timeseries.model.TaskState;
 import org.opensearch.timeseries.util.SecurityUtil;
 
 import com.google.common.base.Throwables;
@@ -509,7 +509,7 @@ private void stopAdJobForEndRunException(
                 executionStartTime,
                 error,
                 true,
-                ADTaskState.STOPPED.name(),
+                TaskState.STOPPED.name(),
                 recorder,
                 detector
             )

diff --git a/src/main/java/org/opensearch/ad/cluster/ADDataMigrator.java b/src/main/java/org/opensearch/ad/cluster/ADDataMigrator.java
@@ -15,9 +15,9 @@
 import static org.opensearch.ad.model.ADTask.DETECTOR_ID_FIELD;
 import static org.opensearch.ad.model.ADTask.IS_LATEST_FIELD;
 import static org.opensearch.ad.model.ADTask.TASK_TYPE_FIELD;
-import static org.opensearch.ad.model.ADTaskType.taskTypeToString;
 import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_DETECTOR_UPPER_LIMIT;
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
+import static org.opensearch.timeseries.model.TaskType.taskTypeToString;
 import static org.opensearch.timeseries.util.RestHandlerUtils.XCONTENT_WITH_TYPE;
 import static org.opensearch.timeseries.util.RestHandlerUtils.createXContentParserFromRegistry;
 
@@ -39,7 +39,6 @@
 import org.opensearch.ad.constant.ADCommonName;
 import org.opensearch.ad.indices.ADIndexManagement;
 import org.opensearch.ad.model.ADTask;
-import org.opensearch.ad.model.ADTaskState;
 import org.opensearch.ad.model.ADTaskType;
 import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.ad.model.DetectorInternalState;
@@ -59,6 +58,7 @@
 import org.opensearch.timeseries.constant.CommonName;
 import org.opensearch.timeseries.function.ExecutorFunction;
 import org.opensearch.timeseries.model.Job;
+import org.opensearch.timeseries.model.TaskState;
 import org.opensearch.timeseries.util.ExceptionUtil;
 
 /**
@@ -245,15 +245,15 @@ private void createRealtimeADTask(Job job, String error, ConcurrentLinkedQueue<J
                     Instant now = Instant.now();
                     String userName = job.getUser() != null ? job.getUser().getName() : null;
                     ADTask adTask = new ADTask.Builder()
-                        .detectorId(detector.getId())
+                        .configId(detector.getId())
                         .detector(detector)
                         .error(error)
                         .isLatest(true)
                         .taskType(taskType.name())
                         .executionStartTime(now)
                         .taskProgress(0.0f)
                         .initProgress(0.0f)
-                        .state(ADTaskState.CREATED.name())
+                        .state(TaskState.CREATED.name())
                         .lastUpdateTime(now)
                         .startedBy(userName)
                         .coordinatingNode(null)

diff --git a/src/main/java/org/opensearch/ad/cluster/ClusterManagerEventListener.java b/src/main/java/org/opensearch/ad/cluster/ClusterManagerEventListener.java
@@ -20,7 +20,7 @@
 import org.opensearch.client.Client;
 import org.opensearch.cluster.LocalNodeClusterManagerListener;
 import org.opensearch.cluster.service.ClusterService;
-import org.opensearch.common.component.LifecycleListener;
+import org.opensearch.common.lifecycle.LifecycleListener;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;

diff --git a/src/main/java/org/opensearch/ad/cluster/HashRing.java b/src/main/java/org/opensearch/ad/cluster/HashRing.java
@@ -46,8 +46,8 @@
 import org.opensearch.cluster.routing.Murmur3HashFunction;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.settings.Settings;
-import org.opensearch.common.transport.TransportAddress;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.transport.TransportAddress;
 import org.opensearch.plugins.PluginInfo;
 import org.opensearch.timeseries.common.exception.TimeSeriesException;
 import org.opensearch.timeseries.constant.CommonName;