[ML] Job in index: delete filter action (elastic#34642)

This changes the delete filter action to search
for jobs using the filter to be deleted in the index
rather than the cluster state.

Author: dimitris-athanasiou

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/action/TransportDeleteFilterAction.java

@@ -16,24 +16,21 @@
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.client.Client;
-import org.elasticsearch.cluster.ClusterState;
-import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.core.ml.MlMetaIndex;
-import org.elasticsearch.xpack.core.ml.MlMetadata;
 import org.elasticsearch.xpack.core.ml.action.DeleteFilterAction;
 import org.elasticsearch.xpack.core.ml.job.config.Detector;
 import org.elasticsearch.xpack.core.ml.job.config.Job;
 import org.elasticsearch.xpack.core.ml.job.config.MlFilter;
 import org.elasticsearch.xpack.core.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.ml.job.persistence.JobConfigProvider;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
 import java.util.function.Supplier;
 
 import static org.elasticsearch.xpack.core.ClientHelper.ML_ORIGIN;
@@ -42,25 +39,39 @@
 public class TransportDeleteFilterAction extends HandledTransportAction<DeleteFilterAction.Request, AcknowledgedResponse> {
 
     private final Client client;
-    private final ClusterService clusterService;
+    private final JobConfigProvider jobConfigProvider;
 
     @Inject
     public TransportDeleteFilterAction(Settings settings, TransportService transportService,
-                                       ActionFilters actionFilters, ClusterService clusterService, Client client) {
+                                       ActionFilters actionFilters, Client client,
+                                       JobConfigProvider jobConfigProvider) {
         super(settings, DeleteFilterAction.NAME, transportService, actionFilters,
             (Supplier<DeleteFilterAction.Request>) DeleteFilterAction.Request::new);
-        this.clusterService = clusterService;
         this.client = client;
+        this.jobConfigProvider = jobConfigProvider;
     }
 
     @Override
     protected void doExecute(Task task, DeleteFilterAction.Request request, ActionListener<AcknowledgedResponse> listener) {
-
         final String filterId = request.getFilterId();
-        ClusterState state = clusterService.state();
-        Map<String, Job> jobs = MlMetadata.getMlMetadata(state).getJobs();
+        jobConfigProvider.findJobsWithCustomRules(ActionListener.wrap(
+                jobs-> {
+                    List<String> currentlyUsedBy = findJobsUsingFilter(jobs, filterId);
+                    if (!currentlyUsedBy.isEmpty()) {
+                        listener.onFailure(ExceptionsHelper.conflictStatusException(
+                            "Cannot delete filter, currently used by jobs: " + currentlyUsedBy));
+                    } else {
+                        deleteFilter(filterId, listener);
+                    }
+                },
+                listener::onFailure
+            )
+        );
+    }
+
+    private static List<String> findJobsUsingFilter(List<Job> jobs, String filterId) {
         List<String> currentlyUsedBy = new ArrayList<>();
-        for (Job job : jobs.values()) {
+        for (Job job : jobs) {
             List<Detector> detectors = job.getAnalysisConfig().getDetectors();
             for (Detector detector : detectors) {
                 if (detector.extractReferencedFilters().contains(filterId)) {
@@ -69,31 +80,31 @@ protected void doExecute(Task task, DeleteFilterAction.Request request, ActionLi
                 }
             }
         }
-        if (!currentlyUsedBy.isEmpty()) {
-            throw ExceptionsHelper.conflictStatusException("Cannot delete filter, currently used by jobs: "
-                    + currentlyUsedBy);
-        }
+        return currentlyUsedBy;
+    }
 
-        DeleteRequest deleteRequest = new DeleteRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE, MlFilter.documentId(filterId));
+    private void deleteFilter(String filterId, ActionListener<AcknowledgedResponse> listener) {
+        DeleteRequest deleteRequest = new DeleteRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE,
+            MlFilter.documentId(filterId));
         BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
         bulkRequestBuilder.add(deleteRequest);
         bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
         executeAsyncWithOrigin(client, ML_ORIGIN, BulkAction.INSTANCE, bulkRequestBuilder.request(),
-                new ActionListener<BulkResponse>() {
-                    @Override
-                    public void onResponse(BulkResponse bulkResponse) {
-                        if (bulkResponse.getItems()[0].status() == RestStatus.NOT_FOUND) {
-                            listener.onFailure(new ResourceNotFoundException("Could not delete filter with ID [" + filterId
-                                    + "] because it does not exist"));
-                        } else {
-                            listener.onResponse(new AcknowledgedResponse(true));
-                        }
+            new ActionListener<BulkResponse>() {
+                @Override
+                public void onResponse(BulkResponse bulkResponse) {
+                    if (bulkResponse.getItems()[0].status() == RestStatus.NOT_FOUND) {
+                        listener.onFailure(new ResourceNotFoundException("Could not delete filter with ID [" + filterId
+                            + "] because it does not exist"));
+                    } else {
+                        listener.onResponse(new AcknowledgedResponse(true));
                     }
+                }
 
-                    @Override
-                    public void onFailure(Exception e) {
-                        listener.onFailure(ExceptionsHelper.serverError("Could not delete filter with ID [" + filterId + "]", e));
-                    }
-        });
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onFailure(ExceptionsHelper.serverError("Could not delete filter with ID [" + filterId + "]", e));
+                }
+            });
     }
 }

x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/job/persistence/JobConfigProvider.java

@@ -107,6 +107,7 @@ public void putJob(Job job, ActionListener<IndexResponse> listener) {
                     ElasticsearchMappings.DOC_TYPE, Job.documentId(job.getId()))
                     .setSource(source)
                     .setOpType(DocWriteRequest.OpType.CREATE)
+                    .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                     .request();
 
             executeAsyncWithOrigin(client, ML_ORIGIN, IndexAction.INSTANCE, indexRequest, ActionListener.wrap(