[Feature/Extensions] Profile detector

build.gradle

@@ -776,7 +776,13 @@ List<String> jacocoExclusions = [
         'org.opensearch.ad.ratelimit.ResultWriteRequest',
         'org.opensearch.ad.AnomalyDetectorJobRunner.*',
         'org.opensearch.ad.util.RestHandlerUtils',
-        'org.opensearch.ad.transport.SearchAnomalyDetectorInfoTransportAction.*'
+        'org.opensearch.ad.transport.SearchAnomalyDetectorInfoTransportAction.*',
+        'org.opensearch.ad.transport.RCFPollingAction',
+        'org.opensearch.ad.transport.RCFPollingRequest',
+        'org.opensearch.ad.transport.RCFPollingTransportAction',
+        'org.opensearch.ad.transport.RCFPollingTransportAction.*',
+        'org.opensearch.ad.transport.RCFPollingResponse',
+
 ]
 
 

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

@@ -92,6 +92,8 @@
 import org.opensearch.ad.transport.ADResultBulkTransportAction;
 import org.opensearch.ad.transport.ADStatsNodesAction;
 import org.opensearch.ad.transport.ADStatsNodesTransportAction;
+import org.opensearch.ad.transport.ADTaskProfileAction;
+import org.opensearch.ad.transport.ADTaskProfileTransportAction;
 import org.opensearch.ad.transport.AnomalyDetectorJobAction;
 import org.opensearch.ad.transport.AnomalyDetectorJobTransportAction;
 import org.opensearch.ad.transport.AnomalyResultAction;
@@ -112,6 +114,8 @@
 import org.opensearch.ad.transport.PreviewAnomalyDetectorTransportAction;
 import org.opensearch.ad.transport.ProfileAction;
 import org.opensearch.ad.transport.ProfileTransportAction;
+import org.opensearch.ad.transport.RCFPollingAction;
+import org.opensearch.ad.transport.RCFPollingTransportAction;
 import org.opensearch.ad.transport.RCFResultAction;
 import org.opensearch.ad.transport.RCFResultTransportAction;
 import org.opensearch.ad.transport.SearchADTasksAction;
@@ -578,7 +582,7 @@ public PooledObject<LinkedBuffer> wrap(LinkedBuffer obj) {
             xContentRegistry,
             anomalyDetectionIndices,
             nodeFilter,
-            null, // hashRing
+            // null, //hashring MultiNode support https://github.com/opensearch-project/opensearch-sdk-java/issues/200
             adTaskCacheManager,
             threadPool
         );
@@ -801,7 +805,9 @@ public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
                 new ActionHandler<>(DeleteModelAction.INSTANCE, DeleteModelTransportAction.class),
                 new ActionHandler<>(ForwardADTaskAction.INSTANCE, ForwardADTaskTransportAction.class),
                 new ActionHandler<>(ADBatchAnomalyResultAction.INSTANCE, ADBatchAnomalyResultTransportAction.class),
-                new ActionHandler<>(ADCancelTaskAction.INSTANCE, ADCancelTaskTransportAction.class)
+                new ActionHandler<>(ADCancelTaskAction.INSTANCE, ADCancelTaskTransportAction.class),
+                new ActionHandler<>(RCFPollingAction.INSTANCE, RCFPollingTransportAction.class),
+                new ActionHandler<>(ADTaskProfileAction.INSTANCE, ADTaskProfileTransportAction.class)
             );
     }
 

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

@@ -49,8 +49,10 @@
 import org.opensearch.ad.settings.AnomalyDetectorSettings;
 import org.opensearch.ad.settings.NumericSetting;
 import org.opensearch.ad.task.ADTaskManager;
+import org.opensearch.ad.transport.ProfileAction;
 import org.opensearch.ad.transport.ProfileRequest;
 import org.opensearch.ad.transport.ProfileResponse;
+import org.opensearch.ad.transport.RCFPollingAction;
 import org.opensearch.ad.transport.RCFPollingRequest;
 import org.opensearch.ad.transport.RCFPollingResponse;
 import org.opensearch.ad.util.DiscoveryNodeFilterer;
@@ -78,23 +80,23 @@
 
 public class AnomalyDetectorProfileRunner extends AbstractProfileRunner {
     private final Logger logger = LogManager.getLogger(AnomalyDetectorProfileRunner.class);
-    private SDKRestClient client;
+    private SDKRestClient sdkRestClient;
     private SDKNamedXContentRegistry xContentRegistry;
     private DiscoveryNodeFilterer nodeFilter;
     private final TransportService transportService;
     private final ADTaskManager adTaskManager;
     private final int maxTotalEntitiesToTrack;
 
     public AnomalyDetectorProfileRunner(
-        SDKRestClient client,
+        SDKRestClient sdkRestClient,
         SDKNamedXContentRegistry xContentRegistry,
         DiscoveryNodeFilterer nodeFilter,
         long requiredSamples,
         TransportService transportService,
         ADTaskManager adTaskManager
     ) {
         super(requiredSamples);
-        this.client = client;
+        this.sdkRestClient = sdkRestClient;
         this.xContentRegistry = xContentRegistry;
         this.nodeFilter = nodeFilter;
         if (requiredSamples <= 0) {
@@ -119,7 +121,7 @@ private void calculateTotalResponsesToWait(
         ActionListener<DetectorProfile> listener
     ) {
         GetRequest getDetectorRequest = new GetRequest(ANOMALY_DETECTORS_INDEX, detectorId);
-        client.get(getDetectorRequest, ActionListener.wrap(getDetectorResponse -> {
+        sdkRestClient.get(getDetectorRequest, ActionListener.wrap(getDetectorResponse -> {
             if (getDetectorResponse != null && getDetectorResponse.isExists()) {
                 try (
                     XContentParser xContentParser = XContentType.JSON
@@ -153,7 +155,7 @@ private void prepareProfile(
     ) {
         String detectorId = detector.getDetectorId();
         GetRequest getRequest = new GetRequest(ANOMALY_DETECTOR_JOB_INDEX, detectorId);
-        client.get(getRequest, ActionListener.wrap(getResponse -> {
+        sdkRestClient.get(getRequest, ActionListener.wrap(getResponse -> {
             if (getResponse != null && getResponse.isExists()) {
                 try (
                     XContentParser parser = XContentType.JSON
@@ -298,7 +300,7 @@ private void profileEntityStats(MultiResponsesDelegateActionListener<DetectorPro
                 searchSourceBuilder.aggregation(aggBuilder);
 
                 SearchRequest request = new SearchRequest(detector.getIndices().toArray(new String[0]), searchSourceBuilder);
-                client.search(request, ActionListener.wrap(searchResponse -> {
+                sdkRestClient.search(request, ActionListener.wrap(searchResponse -> {
                     Map<String, Aggregation> aggMap = searchResponse.getAggregations().asMap();
                     InternalCardinality totalEntities = (InternalCardinality) aggMap.get(CommonName.TOTAL_ENTITIES);
                     long value = totalEntities.getValue();
@@ -321,7 +323,7 @@ private void profileEntityStats(MultiResponsesDelegateActionListener<DetectorPro
                 SearchRequest searchRequest = new SearchRequest()
                     .indices(detector.getIndices().toArray(new String[0]))
                     .source(searchSourceBuilder);
-                client.search(searchRequest, ActionListener.wrap(searchResponse -> {
+                sdkRestClient.search(searchRequest, ActionListener.wrap(searchResponse -> {
                     DetectorProfile.Builder profileBuilder = new DetectorProfile.Builder();
                     Aggregations aggs = searchResponse.getAggregations();
                     if (aggs == null) {
@@ -383,7 +385,7 @@ private void profileStateRelated(
     ) {
         if (enabled) {
             RCFPollingRequest request = new RCFPollingRequest(detector.getDetectorId());
-            // client.execute(RCFPollingAction.INSTANCE, request, onPollRCFUpdates(detector, profilesToCollect, listener));
+            sdkRestClient.execute(RCFPollingAction.INSTANCE, request, onPollRCFUpdates(detector, profilesToCollect, listener));
         } else {
             DetectorProfile.Builder builder = new DetectorProfile.Builder();
             if (profilesToCollect.contains(DetectorProfileName.STATE)) {
@@ -402,7 +404,8 @@ private void profileModels(
     ) {
         DiscoveryNode[] dataNodes = nodeFilter.getEligibleDataNodes();
         ProfileRequest profileRequest = new ProfileRequest(detector.getDetectorId(), profiles, forMultiEntityDetector, dataNodes);
-        // client.execute(ProfileAction.INSTANCE, profileRequest, onModelResponse(detector, profiles, job, listener));// get init progress
+        sdkRestClient.execute(ProfileAction.INSTANCE, profileRequest, onModelResponse(detector, profiles, job, listener));// get init
+                                                                                                                          // progress
     }
 
     private ActionListener<ProfileResponse> onModelResponse(
@@ -482,7 +485,7 @@ private void confirmMultiEntityDetectorInitStatus(
         MultiResponsesDelegateActionListener<DetectorProfile> listener
     ) {
         SearchRequest searchLatestResult = createInittedEverRequest(detector.getDetectorId(), enabledTime, detector.getResultIndex());
-        client.search(searchLatestResult, onInittedEver(enabledTime, profile, profilesToCollect, detector, totalUpdates, listener));
+        sdkRestClient.search(searchLatestResult, onInittedEver(enabledTime, profile, profilesToCollect, detector, totalUpdates, listener));
     }
 
     private ActionListener<SearchResponse> onInittedEver(

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

@@ -12,6 +12,7 @@
 package org.opensearch.ad.rest;
 
 import static org.opensearch.ad.util.RestHandlerUtils.DETECTOR_ID;
+import static org.opensearch.ad.util.RestHandlerUtils.PROFILE;
 import static org.opensearch.ad.util.RestHandlerUtils.TYPE;
 
 import java.io.IOException;
@@ -108,13 +109,6 @@ protected ExtensionRestResponse prepareRequest(RestRequest request) throws IOExc
         return getAnomalyDetectorResponse(request, response);
     }
 
-    @Override
-    public List<ReplacedRouteHandler> replacedRouteHandlers() {
-        String path = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorExtension.LEGACY_OPENDISTRO_AD_BASE_URI, DETECTOR_ID);
-        String newPath = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID);
-        return ImmutableList.of(new ReplacedRouteHandler(RestRequest.Method.GET, newPath, RestRequest.Method.GET, path, handleRequest));
-    }
-
     private Function<RestRequest, ExtensionRestResponse> handleRequest = (request) -> {
         try {
             return prepareRequest(request);
@@ -124,54 +118,58 @@ public List<ReplacedRouteHandler> replacedRouteHandlers() {
         }
     };
 
-    /*@Override
-    public List<Route> routes() {
+    @Override
+    public List<RouteHandler> routeHandlers() {
         return ImmutableList
-                .of(
-                        // Opensearch-only API. Considering users may provide entity in the search body, support POST as well.
-                        new Route(
-                                RestRequest.Method.POST,
-                                String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorPlugin.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE)
-                        ),
-                        new Route(
-                                RestRequest.Method.POST,
-                                String.format(Locale.ROOT, "%s/{%s}/%s/{%s}", AnomalyDetectorPlugin.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE, TYPE)
-                        )
-                );
-    }*/
-
-    /* @Override
-    public List<ReplacedRoute> replacedRoutes() {
-        String path = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorPlugin.LEGACY_OPENDISTRO_AD_BASE_URI, DETECTOR_ID);
-        String newPath = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorPlugin.AD_BASE_DETECTORS_URI, DETECTOR_ID);
+            .of(
+                // Opensearch-only API. Considering users may provide entity in the search body, support POST as well.
+                new RouteHandler(
+                    RestRequest.Method.POST,
+                    String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE),
+                    handleRequest
+                ),
+                new RouteHandler(
+                    RestRequest.Method.POST,
+                    String
+                        .format(Locale.ROOT, "%s/{%s}/%s/{%s}", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE, TYPE),
+                    handleRequest
+                )
+            );
+    }
+
+    @Override
+    public List<ReplacedRouteHandler> replacedRouteHandlers() {
+        String path = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorExtension.LEGACY_OPENDISTRO_AD_BASE_URI, DETECTOR_ID);
+        String newPath = String.format(Locale.ROOT, "%s/{%s}", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID);
         return ImmutableList
             .of(
-                new ReplacedRoute(RestRequest.Method.GET, newPath, RestRequest.Method.GET, path),
-                new ReplacedRoute(RestRequest.Method.HEAD, newPath, RestRequest.Method.HEAD, path),
-                new ReplacedRoute(
+                new ReplacedRouteHandler(RestRequest.Method.GET, newPath, RestRequest.Method.GET, path, handleRequest),
+                new ReplacedRouteHandler(RestRequest.Method.HEAD, newPath, RestRequest.Method.HEAD, path, handleRequest),
+                new ReplacedRouteHandler(
                     RestRequest.Method.GET,
-                    String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorPlugin.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE),
+                    String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE),
                     RestRequest.Method.GET,
-                    String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorPlugin.LEGACY_OPENDISTRO_AD_BASE_URI, DETECTOR_ID, PROFILE)
+                    String.format(Locale.ROOT, "%s/{%s}/%s", AnomalyDetectorExtension.LEGACY_OPENDISTRO_AD_BASE_URI, DETECTOR_ID, PROFILE),
+                    handleRequest
                 ),
-                // types is a profile names. See a complete list of supported profiles names in
-                // org.opensearch.ad.model.ProfileName.
-                new ReplacedRoute(
+                new ReplacedRouteHandler(
                     RestRequest.Method.GET,
-                    String.format(Locale.ROOT, "%s/{%s}/%s/{%s}", AnomalyDetectorPlugin.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE, TYPE),
+                    String
+                        .format(Locale.ROOT, "%s/{%s}/%s/{%s}", AnomalyDetectorExtension.AD_BASE_DETECTORS_URI, DETECTOR_ID, PROFILE, TYPE),
                     RestRequest.Method.GET,
                     String
                         .format(
                             Locale.ROOT,
                             "%s/{%s}/%s/{%s}",
-                            AnomalyDetectorPlugin.LEGACY_OPENDISTRO_AD_BASE_URI,
+                            AnomalyDetectorExtension.LEGACY_OPENDISTRO_AD_BASE_URI,
                             DETECTOR_ID,
                             PROFILE,
                             TYPE
-                        )
+                        ),
+                    handleRequest
                 )
             );
-    }*/
+    }
 
     private Entity buildEntity(RestRequest request, String detectorId) throws IOException {
         if (Strings.isEmpty(detectorId)) {