Skip to content

Commit

Permalink
[Feature/Extensions] Profile detector (#882)
Browse files Browse the repository at this point in the history
  • Loading branch information
@vibrantvarun
vibrantvarun authored May 2, 2023
1 parent 0892126 commit 5dc8edb
Show file tree
Hide file tree
Showing 9 changed files with 234 additions and 246 deletions.
8 changes: 7 additions & 1 deletion build.gradle
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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',

]


Expand Down
10 changes: 8 additions & 2 deletions src/main/java/org/opensearch/ad/AnomalyDetectorExtension.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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
);
Expand Down Expand Up @@ -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)
);
}

Expand Down
23 changes: 13 additions & 10 deletions src/main/java/org/opensearch/ad/AnomalyDetectorProfileRunner.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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();
Expand All @@ -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) {
Expand Down Expand Up @@ -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)) {
Expand All @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
74 changes: 36 additions & 38 deletions src/main/java/org/opensearch/ad/rest/RestGetAnomalyDetectorAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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)) {
Expand Down
Loading

0 comments on commit 5dc8edb

Please sign in to comment.