Adding suggest and validate transport actions to node client (#1605)

* adding skip-changelog to get auto added to depandabot PRs

Signed-off-by: Amit Galitzky <[email protected]>

* adding suggest and validate api to node client

Signed-off-by: Amit Galitzky <[email protected]>

---------

Signed-off-by: Amit Galitzky <[email protected]>

Author: amitgalitz

.github/dependabot.yml

@@ -6,6 +6,8 @@ updates:
       interval: "weekly"
     commit-message:
       prefix: "dependabot:"
+    labels:
+      - "skip-changelog"
     ignore:
       # For all packages, ignore all major versions to minimize breaking issues
       - dependency-name: "*"
@@ -15,4 +17,6 @@ updates:
     schedule:
       interval: "weekly"
     commit-message:
-      prefix: "dependabot:"
+      prefix: "dependabot:"
+    labels:
+      - "skip-changelog"

CHANGELOG.md

@@ -7,6 +7,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 ### Features
 ### Enhancements
 - Adds capability to automatically switch to old access-control if model-group is excluded from protected resources setting ([#1569](https://github.com/opensearch-project/anomaly-detection/pull/1569))
+- Adding suggest and validate transport actions to node client ([#1605](https://github.com/opensearch-project/anomaly-detection/pull/1605))
 - Adding auto create as an optional field on detectors ([#1602](https://github.com/opensearch-project/anomaly-detection/pull/1602))
 
 ### Bug Fixes

src/main/java/org/opensearch/ad/client/AnomalyDetectionClient.java

@@ -12,6 +12,10 @@
 import org.opensearch.common.action.ActionFuture;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.timeseries.transport.GetConfigRequest;
+import org.opensearch.timeseries.transport.SuggestConfigParamRequest;
+import org.opensearch.timeseries.transport.SuggestConfigParamResponse;
+import org.opensearch.timeseries.transport.ValidateConfigRequest;
+import org.opensearch.timeseries.transport.ValidateConfigResponse;
 
 /**
  * A client to provide interfaces for anomaly detection functionality. This will be used by other plugins.
@@ -71,4 +75,39 @@ default ActionFuture<GetAnomalyDetectorResponse> getDetectorProfile(GetConfigReq
      */
     void getDetectorProfile(GetConfigRequest profileRequest, ActionListener<GetAnomalyDetectorResponse> listener);
 
+    /**
+     * Validate anomaly detector - refer to https://opensearch.org/docs/latest/observing-your-data/ad/api/#validate-detector
+     * @param validateRequest request to validate the detector configuration
+     * @return ActionFuture of ValidateConfigResponse
+     */
+    default ActionFuture<ValidateConfigResponse> validateAnomalyDetector(ValidateConfigRequest validateRequest) {
+        PlainActionFuture<ValidateConfigResponse> actionFuture = PlainActionFuture.newFuture();
+        validateAnomalyDetector(validateRequest, actionFuture);
+        return actionFuture;
+    }
+
+    /**
+     * Validate anomaly detector - refer to https://opensearch.org/docs/latest/observing-your-data/ad/api/#validate-detector
+     * @param validateRequest request to validate the detector configuration
+     * @param listener a listener to be notified of the result
+     */
+    void validateAnomalyDetector(ValidateConfigRequest validateRequest, ActionListener<ValidateConfigResponse> listener);
+
+    /**
+     * Suggest anomaly detector parameters - refer to https://opensearch.org/docs/latest/observing-your-data/ad/api/#suggest-detector
+     * @param suggestRequest request to suggest detector configuration parameters
+     * @return ActionFuture of SuggestConfigParamResponse
+     */
+    default ActionFuture<SuggestConfigParamResponse> suggestAnomalyDetector(SuggestConfigParamRequest suggestRequest) {
+        PlainActionFuture<SuggestConfigParamResponse> actionFuture = PlainActionFuture.newFuture();
+        suggestAnomalyDetector(suggestRequest, actionFuture);
+        return actionFuture;
+    }
+
+    /**
+     * Suggest anomaly detector parameters - refer to https://opensearch.org/docs/latest/observing-your-data/ad/api/#suggest-detector
+     * @param suggestRequest request to suggest detector configuration parameters
+     * @param listener a listener to be notified of the result
+     */
+    void suggestAnomalyDetector(SuggestConfigParamRequest suggestRequest, ActionListener<SuggestConfigParamResponse> listener);
 }

src/main/java/org/opensearch/ad/client/AnomalyDetectionNodeClient.java

@@ -13,10 +13,16 @@
 import org.opensearch.ad.transport.GetAnomalyDetectorResponse;
 import org.opensearch.ad.transport.SearchAnomalyDetectorAction;
 import org.opensearch.ad.transport.SearchAnomalyResultAction;
+import org.opensearch.ad.transport.SuggestAnomalyDetectorParamAction;
+import org.opensearch.ad.transport.ValidateAnomalyDetectorAction;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.action.ActionResponse;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.timeseries.transport.GetConfigRequest;
+import org.opensearch.timeseries.transport.SuggestConfigParamRequest;
+import org.opensearch.timeseries.transport.SuggestConfigParamResponse;
+import org.opensearch.timeseries.transport.ValidateConfigRequest;
+import org.opensearch.timeseries.transport.ValidateConfigResponse;
 import org.opensearch.transport.client.Client;
 
 public class AnomalyDetectionNodeClient implements AnomalyDetectionClient {
@@ -47,6 +53,16 @@ public void getDetectorProfile(GetConfigRequest profileRequest, ActionListener<G
         this.client.execute(GetAnomalyDetectorAction.INSTANCE, profileRequest, getAnomalyDetectorResponseActionListener(listener));
     }
 
+    @Override
+    public void validateAnomalyDetector(ValidateConfigRequest validateRequest, ActionListener<ValidateConfigResponse> listener) {
+        this.client.execute(ValidateAnomalyDetectorAction.INSTANCE, validateRequest, validateConfigResponseActionListener(listener));
+    }
+
+    @Override
+    public void suggestAnomalyDetector(SuggestConfigParamRequest suggestRequest, ActionListener<SuggestConfigParamResponse> listener) {
+        this.client.execute(SuggestAnomalyDetectorParamAction.INSTANCE, suggestRequest, suggestConfigResponseActionListener(listener));
+    }
+
     // We need to wrap AD-specific response type listeners around an internal listener, and re-generate the response from a generic
     // ActionResponse. This is needed to prevent classloader issues and ClassCastExceptions when executed by other plugins.
     // Additionally, we need to inject the configured NamedWriteableRegistry so NamedWriteables (present in sub-fields of
@@ -66,6 +82,31 @@ private ActionListener<GetAnomalyDetectorResponse> getAnomalyDetectorResponseAct
         return actionListener;
     }
 
+    private ActionListener<ValidateConfigResponse> validateConfigResponseActionListener(ActionListener<ValidateConfigResponse> listener) {
+        ActionListener<ValidateConfigResponse> internalListener = ActionListener.wrap(validateConfigResponse -> {
+            listener.onResponse(validateConfigResponse);
+        }, listener::onFailure);
+        ActionListener<ValidateConfigResponse> actionListener = wrapActionListener(internalListener, actionResponse -> {
+            ValidateConfigResponse response = ValidateConfigResponse.fromActionResponse(actionResponse, this.namedWriteableRegistry);
+            return response;
+        });
+        return actionListener;
+    }
+
+    private ActionListener<SuggestConfigParamResponse> suggestConfigResponseActionListener(
+        ActionListener<SuggestConfigParamResponse> listener
+    ) {
+        ActionListener<SuggestConfigParamResponse> internalListener = ActionListener.wrap(suggestConfigResponse -> {
+            listener.onResponse(suggestConfigResponse);
+        }, listener::onFailure);
+        ActionListener<SuggestConfigParamResponse> actionListener = wrapActionListener(internalListener, actionResponse -> {
+            SuggestConfigParamResponse response = SuggestConfigParamResponse
+                .fromActionResponse(actionResponse, this.namedWriteableRegistry);
+            return response;
+        });
+        return actionListener;
+    }
+
     private <T extends ActionResponse> ActionListener<T> wrapActionListener(
         final ActionListener<T> listener,
         final Function<ActionResponse, T> recreate

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

@@ -23,6 +23,7 @@
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.commons.authuser.User;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.timeseries.AnalysisType;
 import org.opensearch.timeseries.Name;
 import org.opensearch.timeseries.constant.CommonMessages;
@@ -50,7 +51,8 @@ public SuggestAnomalyDetectorParamTransportAction(
         ADIndexManagement anomalyDetectionIndices,
         ActionFilters actionFilters,
         TransportService transportService,
-        SearchFeatureDao searchFeatureDao
+        SearchFeatureDao searchFeatureDao,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(
             SuggestAnomalyDetectorParamAction.NAME,
@@ -64,7 +66,9 @@ public SuggestAnomalyDetectorParamTransportAction(
             AnalysisType.AD,
             searchFeatureDao,
             Name.getListStrs(Arrays.asList(ADSuggestName.values())),
-            AnomalyDetector.class
+            AnomalyDetector.class,
+            namedWriteableRegistry
+
         );
     }
 

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

@@ -24,6 +24,7 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.commons.authuser.User;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.rest.RestRequest;
 import org.opensearch.timeseries.feature.SearchFeatureDao;
@@ -50,7 +51,8 @@ public ValidateAnomalyDetectorTransportAction(
         ADIndexManagement anomalyDetectionIndices,
         ActionFilters actionFilters,
         TransportService transportService,
-        SearchFeatureDao searchFeatureDao
+        SearchFeatureDao searchFeatureDao,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(
             ValidateAnomalyDetectorAction.NAME,
@@ -65,7 +67,8 @@ public ValidateAnomalyDetectorTransportAction(
             searchFeatureDao,
             AD_FILTER_BY_BACKEND_ROLES,
             ValidationAspect.DETECTOR,
-            AnomalyDetector.class
+            AnomalyDetector.class,
+            namedWriteableRegistry
         );
     }
 

src/main/java/org/opensearch/forecast/transport/SuggestForecasterParamTransportAction.java

@@ -21,6 +21,7 @@
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.commons.authuser.User;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.forecast.indices.ForecastIndexManagement;
 import org.opensearch.forecast.model.Forecaster;
 import org.opensearch.timeseries.AnalysisType;
@@ -50,7 +51,8 @@ public SuggestForecasterParamTransportAction(
         ForecastIndexManagement anomalyDetectionIndices,
         ActionFilters actionFilters,
         TransportService transportService,
-        SearchFeatureDao searchFeatureDao
+        SearchFeatureDao searchFeatureDao,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(
             SuggestForecasterParamAction.NAME,
@@ -64,7 +66,8 @@ public SuggestForecasterParamTransportAction(
             AnalysisType.FORECAST,
             searchFeatureDao,
             Name.getListStrs(Arrays.asList(ForecastSuggestName.values())),
-            Forecaster.class
+            Forecaster.class,
+            namedWriteableRegistry
         );
     }
 

src/main/java/org/opensearch/forecast/transport/ValidateForecasterTransportAction.java

@@ -14,6 +14,7 @@
 import org.opensearch.common.inject.Inject;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.commons.authuser.User;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.forecast.indices.ForecastIndex;
 import org.opensearch.forecast.indices.ForecastIndexManagement;
@@ -45,7 +46,8 @@ public ValidateForecasterTransportAction(
         ForecastIndexManagement anomalyDetectionIndices,
         ActionFilters actionFilters,
         TransportService transportService,
-        SearchFeatureDao searchFeatureDao
+        SearchFeatureDao searchFeatureDao,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(
             ValidateForecasterAction.NAME,
@@ -60,7 +62,8 @@ public ValidateForecasterTransportAction(
             searchFeatureDao,
             FORECAST_FILTER_BY_BACKEND_ROLES,
             ValidationAspect.FORECASTER,
-            Forecaster.class
+            Forecaster.class,
+            namedWriteableRegistry
         );
     }
 

src/main/java/org/opensearch/timeseries/transport/BaseSuggestConfigParamTransportAction.java

@@ -19,6 +19,7 @@
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.action.ActionRequest;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.HandledTransportAction;
 import org.opensearch.cluster.service.ClusterService;
@@ -28,6 +29,7 @@
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.commons.authuser.User;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.tasks.Task;
 import org.opensearch.timeseries.AnalysisType;
 import org.opensearch.timeseries.Name;
@@ -45,12 +47,13 @@
 import org.opensearch.transport.client.Client;
 
 public abstract class BaseSuggestConfigParamTransportAction<ConfigType extends Config> extends
-    HandledTransportAction<SuggestConfigParamRequest, SuggestConfigParamResponse> {
+    HandledTransportAction<ActionRequest, SuggestConfigParamResponse> {
     public static final Logger logger = LogManager.getLogger(BaseSuggestConfigParamTransportAction.class);
 
     protected final Client client;
     protected final SecurityClientUtil clientUtil;
     protected final SearchFeatureDao searchFeatureDao;
+    protected final NamedWriteableRegistry namedWriteableRegistry;
     protected volatile Boolean filterByEnabled;
     protected Clock clock;
     protected AnalysisType context;
@@ -70,11 +73,13 @@ public BaseSuggestConfigParamTransportAction(
         AnalysisType context,
         SearchFeatureDao searchFeatureDao,
         Set<String> allSuggestParamStrs,
-        Class<ConfigType> configTypeClass
+        Class<ConfigType> configTypeClass,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(actionName, transportService, actionFilters, SuggestConfigParamRequest::new);
         this.client = client;
         this.clientUtil = clientUtil;
+        this.namedWriteableRegistry = namedWriteableRegistry;
         this.filterByEnabled = filterByBackendRoleSetting.get(settings);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(filterByBackendRoleSetting, it -> filterByEnabled = it);
         this.clock = Clock.systemUTC();
@@ -86,7 +91,8 @@ public BaseSuggestConfigParamTransportAction(
     }
 
     @Override
-    protected void doExecute(Task task, SuggestConfigParamRequest request, ActionListener<SuggestConfigParamResponse> listener) {
+    protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<SuggestConfigParamResponse> listener) {
+        SuggestConfigParamRequest request = SuggestConfigParamRequest.fromActionRequest(actionRequest, namedWriteableRegistry);
         User user = ParseUtils.getUserContext(client);
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
             String resourceType = getResourceTypeFromClassName(configTypeClass.getSimpleName());

src/main/java/org/opensearch/timeseries/transport/BaseValidateConfigTransportAction.java

@@ -14,6 +14,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.action.ActionRequest;
 import org.opensearch.action.search.SearchRequest;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.HandledTransportAction;
@@ -23,6 +24,7 @@
 import org.opensearch.common.util.concurrent.ThreadContext;
 import org.opensearch.commons.authuser.User;
 import org.opensearch.core.action.ActionListener;
+import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.index.IndexNotFoundException;
 import org.opensearch.index.query.QueryBuilders;
@@ -47,7 +49,7 @@
 import org.opensearch.transport.client.Client;
 
 public abstract class BaseValidateConfigTransportAction<IndexType extends Enum<IndexType> & TimeSeriesIndex, IndexManagementType extends IndexManagement<IndexType>, ConfigType extends Config>
-    extends HandledTransportAction<ValidateConfigRequest, ValidateConfigResponse> {
+    extends HandledTransportAction<ActionRequest, ValidateConfigResponse> {
     public static final Logger logger = LogManager.getLogger(BaseValidateConfigTransportAction.class);
 
     protected final Client client;
@@ -56,6 +58,7 @@ public abstract class BaseValidateConfigTransportAction<IndexType extends Enum<I
     protected final NamedXContentRegistry xContentRegistry;
     protected final IndexManagementType indexManagement;
     protected final SearchFeatureDao searchFeatureDao;
+    protected final NamedWriteableRegistry namedWriteableRegistry;
     protected volatile Boolean filterByEnabled;
     protected Clock clock;
     protected Settings settings;
@@ -75,14 +78,16 @@ public BaseValidateConfigTransportAction(
         SearchFeatureDao searchFeatureDao,
         Setting<Boolean> filterByBackendRoleSetting,
         ValidationAspect validationAspect,
-        Class<ConfigType> configTypeClass
+        Class<ConfigType> configTypeClass,
+        NamedWriteableRegistry namedWriteableRegistry
     ) {
         super(actionName, transportService, actionFilters, ValidateConfigRequest::new);
         this.client = client;
         this.clientUtil = clientUtil;
         this.clusterService = clusterService;
         this.xContentRegistry = xContentRegistry;
         this.indexManagement = indexManagement;
+        this.namedWriteableRegistry = namedWriteableRegistry;
         this.filterByEnabled = filterByBackendRoleSetting.get(settings);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(filterByBackendRoleSetting, it -> filterByEnabled = it);
         this.searchFeatureDao = searchFeatureDao;
@@ -93,7 +98,8 @@ public BaseValidateConfigTransportAction(
     }
 
     @Override
-    protected void doExecute(Task task, ValidateConfigRequest request, ActionListener<ValidateConfigResponse> listener) {
+    protected void doExecute(Task task, ActionRequest actionRequest, ActionListener<ValidateConfigResponse> listener) {
+        ValidateConfigRequest request = ValidateConfigRequest.fromActionRequest(actionRequest, namedWriteableRegistry);
         User user = ParseUtils.getUserContext(client);
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
             String resourceType = getResourceTypeFromClassName(configTypeClass.getSimpleName());