Allow detectors to be stopped if underlying workflow is deleted. Don'…

…t allow them to then be started/editted (#810)

* Allow detectors to be stopped if underlying workflow is deleted. Don't allow them to then be started

Signed-off-by: Chase Engelbrecht <[email protected]>

* Add copyright headers

Signed-off-by: Chase Engelbrecht <[email protected]>

---------

Signed-off-by: Chase Engelbrecht <[email protected]>

src/main/java/org/opensearch/securityanalytics/model/Detector.java

@@ -568,6 +568,10 @@ public void setAlertsHistoryIndexPattern(String alertsHistoryIndexPattern) {
  this.alertsHistoryIndexPattern = alertsHistoryIndexPattern;
  }
 
+ public void setEnabled(boolean enabled) {
+ this.enabled = enabled;
+ }
+
  public void setEnabledTime(Instant enabledTime) {
  this.enabledTime = enabledTime;
  }

src/main/java/org/opensearch/securityanalytics/transport/TransportDeleteDetectorAction.java

@@ -42,9 +42,11 @@
 import org.opensearch.securityanalytics.model.Detector;
 import org.opensearch.securityanalytics.settings.SecurityAnalyticsSettings;
 import org.opensearch.securityanalytics.util.DetectorIndices;
+import org.opensearch.securityanalytics.util.ExceptionChecker;
 import org.opensearch.securityanalytics.util.MonitorService;
 import org.opensearch.securityanalytics.util.RuleTopicIndices;
 import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
+import org.opensearch.securityanalytics.util.ThrowableCheckingPredicates;
 import org.opensearch.securityanalytics.util.WorkflowService;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
@@ -61,6 +63,11 @@
 public class TransportDeleteDetectorAction extends HandledTransportAction<DeleteDetectorRequest, DeleteDetectorResponse> {
 
  private static final Logger log = LogManager.getLogger(TransportDeleteDetectorAction.class);
+ private static final List<ThrowableCheckingPredicates> ACCEPTABLE_ENTITY_MISSING_THROWABLE_MATCHERS = List.of(
+ ThrowableCheckingPredicates.MONITOR_NOT_FOUND,
+ ThrowableCheckingPredicates.WORKFLOW_NOT_FOUND,
+ ThrowableCheckingPredicates.ALERTING_CONFIG_INDEX_NOT_FOUND
+ );
 
  private final Client client;
 
@@ -83,9 +90,13 @@ public class TransportDeleteDetectorAction extends HandledTransportAction<Delete
 
  private final DetectorIndices detectorIndices;
 
+ private final ExceptionChecker exceptionChecker;
+
  @Inject
- public TransportDeleteDetectorAction(TransportService transportService, IndexTemplateManager indexTemplateManager, Client client, ActionFilters actionFilters, NamedXContentRegistry xContentRegistry, RuleTopicIndices ruleTopicIndices, DetectorIndices detectorIndices, ClusterService clusterService,
- Settings settings) {
+ public TransportDeleteDetectorAction(TransportService transportService, IndexTemplateManager indexTemplateManager, Client client,
+ ActionFilters actionFilters, NamedXContentRegistry xContentRegistry, RuleTopicIndices ruleTopicIndices,
+ DetectorIndices detectorIndices, ClusterService clusterService, Settings settings,
+ ExceptionChecker exceptionChecker) {
  super(DeleteDetectorAction.NAME, transportService, actionFilters, DeleteDetectorRequest::new);
  this.client = client;
  this.ruleTopicIndices = ruleTopicIndices;
@@ -100,6 +111,7 @@ public TransportDeleteDetectorAction(TransportService transportService, IndexTem
 
  this.enabledWorkflowUsage = SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE.get(this.settings);
  this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE, this::setEnabledWorkflowUsage);
+ this.exceptionChecker = exceptionChecker;
  }
 
  @Override
@@ -204,7 +216,8 @@ public void onResponse(Collection<DeleteMonitorResponse> responses) {
 
  @Override
  public void onFailure(Exception e) {
- if (isOnlyWorkflowOrMonitorOrIndexMissingExceptionThrownByGroupedActionListener(e, detector.getId())) {
+ if (exceptionChecker.doesGroupedActionListenerExceptionMatch(e, ACCEPTABLE_ENTITY_MISSING_THROWABLE_MATCHERS)) {
+ logAcceptableEntityMissingException(e, detector.getId());
  deleteDetectorFromConfig(detector.getId(), request.getRefreshPolicy());
  } else {
  log.error(String.format(Locale.ROOT, "Failed to delete detector %s", detector.getId()), e);
@@ -243,7 +256,8 @@ private void deleteWorkflow(Detector detector, ActionListener<AcknowledgedRespon
 
  private void handleDeleteWorkflowFailure(final String detectorId, final Exception deleteWorkflowException,
  final ActionListener<AcknowledgedResponse> actionListener) {
- if (isOnlyWorkflowOrMonitorOrIndexMissingExceptionThrownByGroupedActionListener(deleteWorkflowException, detectorId)) {
+ if (exceptionChecker.doesGroupedActionListenerExceptionMatch(deleteWorkflowException, ACCEPTABLE_ENTITY_MISSING_THROWABLE_MATCHERS)) {
+ logAcceptableEntityMissingException(deleteWorkflowException, detectorId);
  actionListener.onResponse(new AcknowledgedResponse(true));
  } else {
  actionListener.onFailure(deleteWorkflowException);
@@ -305,39 +319,12 @@ private void finishHim(String detectorId, Exception t) {
  }
  }));
  }
-
- private boolean isOnlyWorkflowOrMonitorOrIndexMissingExceptionThrownByGroupedActionListener(
- Exception ex,
- String detectorId
- ) {
- // grouped action listener listens on mutliple listeners but throws only one exception. If multiple
- // listeners fail the other exceptions are added as suppressed exceptions to the first failure.
- int len = ex.getSuppressed().length;
- for (int i = 0; i <= len; i++) {
- Throwable e = i == len ? ex : ex.getSuppressed()[i];
- if (isMonitorNotFoundException(e) || isWorkflowNotFoundException(e) || isAlertingConfigIndexNotFoundException(e)) {
- log.error(
- String.format(Locale.ROOT, "Workflow, monitor, or jobs index already deleted." +
- " Proceeding with detector %s deletion", detectorId),
- e);
- } else {
- return false;
- }
- }
- return true;
- }
- }
-
- private boolean isMonitorNotFoundException(final Throwable e) {
- return e.getMessage().matches("(.*)Monitor(.*) is not found(.*)");
- }
-
- private boolean isWorkflowNotFoundException(final Throwable e) {
- return e.getMessage().matches("(.*)Workflow(.*) not found(.*)");
  }
 
- private boolean isAlertingConfigIndexNotFoundException(final Throwable e) {
- return e.getMessage().contains("Configured indices are not found: [.opendistro-alerting-config]");
+ private void logAcceptableEntityMissingException(final Exception e, final String detectorId) {
+ final String errorMsg = String.format(Locale.ROOT, "Workflow, monitor, or jobs index already deleted." +
+ " Proceeding with detector %s deletion", detectorId);
+ log.error(errorMsg, e);
  }
 
  private void setEnabledWorkflowUsage(boolean enabledWorkflowUsage) {

src/main/java/org/opensearch/securityanalytics/transport/TransportIndexDetectorAction.java

@@ -97,11 +97,13 @@
 import org.opensearch.securityanalytics.settings.SecurityAnalyticsSettings;
 import org.opensearch.securityanalytics.util.DetectorIndices;
 import org.opensearch.securityanalytics.util.DetectorUtils;
+import org.opensearch.securityanalytics.util.ExceptionChecker;
 import org.opensearch.securityanalytics.util.IndexUtils;
 import org.opensearch.securityanalytics.util.MonitorService;
 import org.opensearch.securityanalytics.util.RuleIndices;
 import org.opensearch.securityanalytics.util.RuleTopicIndices;
 import org.opensearch.securityanalytics.util.SecurityAnalyticsException;
+import org.opensearch.securityanalytics.util.ThrowableCheckingPredicates;
 import org.opensearch.securityanalytics.util.WorkflowService;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
@@ -155,6 +157,8 @@ public class TransportIndexDetectorAction extends HandledTransportAction<IndexDe
  private final MonitorService monitorService;
  private final IndexNameExpressionResolver indexNameExpressionResolver;
 
+ private final ExceptionChecker exceptionChecker;
+
  private final TimeValue indexTimeout;
  @Inject
  public TransportIndexDetectorAction(TransportService transportService,
@@ -169,7 +173,8 @@ public TransportIndexDetectorAction(TransportService transportService,
  Settings settings,
  NamedWriteableRegistry namedWriteableRegistry,
  LogTypeService logTypeService,
- IndexNameExpressionResolver indexNameExpressionResolver) {
+ IndexNameExpressionResolver indexNameExpressionResolver,
+ ExceptionChecker exceptionChecker) {
  super(IndexDetectorAction.NAME, transportService, actionFilters, IndexDetectorRequest::new);
  this.client = client;
  this.xContentRegistry = xContentRegistry;
@@ -191,6 +196,7 @@ public TransportIndexDetectorAction(TransportService transportService,
 
  this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.FILTER_BY_BACKEND_ROLES, this::setFilterByEnabled);
  this.clusterService.getClusterSettings().addSettingsUpdateConsumer(SecurityAnalyticsSettings.ENABLE_WORKFLOW_USAGE, this::setEnabledWorkflowUsage);
+ this.exceptionChecker = exceptionChecker;
  }
 
  @Override
@@ -616,8 +622,7 @@ public void onResponse(IndexWorkflowResponse workflowResponse) {
  }
  @Override
  public void onFailure(Exception e) {
- log.error("Failed to update the workflow");
- listener.onFailure(e);
+ handleUpsertWorkflowFailure(e, listener, detector, monitorsToBeDeleted, refreshPolicy, updatedMonitors);
  }
  });
  }
@@ -674,6 +679,25 @@ private IndexMonitorRequest createDocLevelMonitorRequest(List<Pair<String, Rule>
  return new IndexMonitorRequest(monitorId, SequenceNumbers.UNASSIGNED_SEQ_NO, SequenceNumbers.UNASSIGNED_PRIMARY_TERM, refreshPolicy, restMethod, monitor, null);
  }
 
+ private void handleUpsertWorkflowFailure(final Exception e, final ActionListener<List<IndexMonitorResponse>> listener,
+ final Detector detector, final List<String> monitorsToBeDeleted,
+ final RefreshPolicy refreshPolicy, final List<IndexMonitorResponse> updatedMonitors) {
+ if (exceptionChecker.doesGroupedActionListenerExceptionMatch(e, List.of(ThrowableCheckingPredicates.WORKFLOW_NOT_FOUND))) {
+ if (detector.getEnabled()) {
+ final String errorMessage = String.format("Underlying workflow associated with detector %s not found. " +
+ "Delete and recreate the detector to restore functionality.", detector.getName());
+ log.error(errorMessage);
+ listener.onFailure(new SecurityAnalyticsException(errorMessage, RestStatus.BAD_REQUEST, e));
+ } else {
+ log.error("Underlying workflow associated with detector {} not found. Proceeding to disable detector.", detector.getName());
+ deleteMonitorStep(monitorsToBeDeleted, refreshPolicy, updatedMonitors, listener);
+ }
+ } else {
+ log.error("Failed to update the workflow");
+ listener.onFailure(e);
+ }
+ }
+
  /**
  * Creates doc level monitor which generates per document alerts for the findings of the bucket level delegate monitors in a workflow.
  * This monitor has match all query applied to generate the alerts per each finding doc.

src/main/java/org/opensearch/securityanalytics/util/ExceptionChecker.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.securityanalytics.util;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Stream;
+
+public class ExceptionChecker {
+
+ public boolean doesGroupedActionListenerExceptionMatch(final Exception ex, final List<ThrowableCheckingPredicates> exceptionMatchers) {
+ // grouped action listener listens on multiple listeners but throws only one exception. If multiple
+ // listeners fail the other exceptions are added as suppressed exceptions to the first failure.
+ return Stream.concat(Arrays.stream(ex.getSuppressed()), Stream.of(ex))
+ .allMatch(throwable -> doesExceptionMatch(throwable, exceptionMatchers));
+ }
+
+ private boolean doesExceptionMatch(final Throwable throwable, final List<ThrowableCheckingPredicates> exceptionMatchers) {
+ return exceptionMatchers.stream()
+ .map(ThrowableCheckingPredicates::getMatcherPredicate)
+ .anyMatch(matcher -> matcher.test(throwable));
+ }
+
+}

src/main/java/org/opensearch/securityanalytics/util/ThrowableCheckingPredicates.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.securityanalytics.util;
+
+import java.util.function.Predicate;
+
+public enum ThrowableCheckingPredicates {
+ MONITOR_NOT_FOUND(ThrowableCheckingPredicates::isMonitorNotFoundException),
+ WORKFLOW_NOT_FOUND(ThrowableCheckingPredicates::isWorkflowNotFoundException),
+ ALERTING_CONFIG_INDEX_NOT_FOUND(ThrowableCheckingPredicates::isAlertingConfigIndexNotFoundException);
+
+ private final Predicate<Throwable> matcherPredicate;
+ ThrowableCheckingPredicates(final Predicate<Throwable> matcherPredicate) {
+ this.matcherPredicate = matcherPredicate;
+ }
+
+ public Predicate<Throwable> getMatcherPredicate() {
+ return this.matcherPredicate;
+ }
+
+ private static boolean isMonitorNotFoundException(final Throwable e) {
+ return e.getMessage().matches("(.*)Monitor(.*) is not found(.*)");
+ }
+
+ public static boolean isWorkflowNotFoundException(final Throwable e) {
+ return e.getMessage().matches("(.*)Workflow(.*) not found(.*)");
+ }
+
+ public static boolean isAlertingConfigIndexNotFoundException(final Throwable e) {
+ return e.getMessage().contains("Configured indices are not found: [.opendistro-alerting-config]");
+ }
+}

src/test/java/org/opensearch/securityanalytics/resthandler/DetectorRestApiIT.java

@@ -751,6 +751,50 @@ public void testUpdateADetectorWithIndexNotExists() throws IOException {
  }
  }
 
+ public void testDisableEnableADetectorWithWorkflowNotExists() throws IOException {
+ final String index = createTestIndex(randomIndex(), windowsIndexMapping());
+
+ // Execute CreateMappingsAction to add alias mapping for index
+ final Request createMappingRequest = new Request("POST", SecurityAnalyticsPlugin.MAPPER_BASE_URI);
+ // both req params and req body are supported
+ createMappingRequest.setJsonEntity(
+ "{ \"index_name\":\"" + index + "\"," +
+ " \"rule_topic\":\"" + randomDetectorType() + "\", " +
+ " \"partial\":true" +
+ "}"
+ );
+
+ final Response createMappingResponse = client().performRequest(createMappingRequest);
+ assertEquals(HttpStatus.SC_OK, createMappingResponse.getStatusLine().getStatusCode());
+
+ final Detector detector = randomDetector(getRandomPrePackagedRules());
+ final Response createResponse = makeRequest(client(), "POST", SecurityAnalyticsPlugin.DETECTOR_BASE_URI, Collections.emptyMap(), toHttpEntity(detector));
+ Assert.assertEquals("Create detector failed", RestStatus.CREATED, restStatus(createResponse));
+
+ final Map<String, Object> createResponseAsMap = asMap(createResponse);
+ final String detectorId = createResponseAsMap.get("_id").toString();
+
+ final Map<String, Object> detectorSourceAsMap = getDetectorSourceAsMap(detectorId);
+ final String workflowId = ((List<String>) detectorSourceAsMap.get("workflow_ids")).get(0);
+
+ final Response deleteWorkflowResponse = deleteAlertingWorkflow(workflowId);
+ assertEquals(200, deleteWorkflowResponse.getStatusLine().getStatusCode());
+ entityAsMap(deleteWorkflowResponse);
+
+ detector.setEnabled(false);
+ Response updateResponse = makeRequest(client(), "PUT", SecurityAnalyticsPlugin.DETECTOR_BASE_URI + "/" + detectorId, Collections.emptyMap(), toHttpEntity(detector));
+ Assert.assertEquals(200, updateResponse.getStatusLine().getStatusCode());
+
+ try {
+ detector.setEnabled(true);
+ makeRequest(client(), "PUT", SecurityAnalyticsPlugin.DETECTOR_BASE_URI + "/" + detectorId, Collections.emptyMap(), toHttpEntity(detector));
+ } catch (ResponseException ex) {
+ Assert.assertEquals(400, ex.getResponse().getStatusLine().getStatusCode());
+ Assert.assertEquals(true, ex.getMessage().contains(String.format("Underlying workflow associated with detector %s not found. " +
+ "Delete and recreate the detector to restore functionality.", detector.getName())));
+ }
+ }
+
  @SuppressWarnings("unchecked")
  public void testDeletingADetector_single_ruleTopicIndex() throws IOException {
  String index = createTestIndex(randomIndex(), windowsIndexMapping());