[ML] Ensure internal client is used where appropriate (elastic/x-pack-elasticsearch#2415)

Implementation details of ML endpoints should be performed using the
internal client, so that the end user only requires permissions for
the public ML endpoints and does not need to know how they are
implemented.  This change fixes some instances where this rule was
not adhered to.

Original commit: elastic/x-pack-elasticsearch@01c8f5172c

plugin/src/main/java/org/elasticsearch/xpack/ml/action/DeleteFilterAction.java

@@ -9,9 +9,8 @@ import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.Action;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.bulk.TransportBulkAction;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
@@ -37,6 +36,7 @@ import org.elasticsearch.xpack.ml.job.config.Detector;
 import org.elasticsearch.xpack.ml.job.config.Job;
 import org.elasticsearch.xpack.ml.job.config.MlFilter;
 import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.security.InternalClient;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -144,18 +144,18 @@ public class DeleteFilterAction extends Action<DeleteFilterAction.Request, Delet
 
     public static class TransportAction extends HandledTransportAction<Request, Response> {
 
-        private final TransportBulkAction transportAction;
+        private final InternalClient client;
         private final ClusterService clusterService;
 
         @Inject
         public TransportAction(Settings settings, ThreadPool threadPool,
                 TransportService transportService, ActionFilters actionFilters,
                 IndexNameExpressionResolver indexNameExpressionResolver,
-                ClusterService clusterService, TransportBulkAction transportAction) {
+                ClusterService clusterService, InternalClient client) {
             super(settings, NAME, threadPool, transportService, actionFilters,
                     indexNameExpressionResolver, Request::new);
             this.clusterService = clusterService;
-            this.transportAction = transportAction;
+            this.client = client;
         }
 
         @Override
@@ -181,10 +181,10 @@ public class DeleteFilterAction extends Action<DeleteFilterAction.Request, Delet
             }
 
             DeleteRequest deleteRequest = new DeleteRequest(MlMetaIndex.INDEX_NAME, MlMetaIndex.TYPE, MlFilter.documentId(filterId));
-            BulkRequest bulkRequest = new BulkRequest();
-            bulkRequest.add(deleteRequest);
-            bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
-            transportAction.execute(bulkRequest, new ActionListener<BulkResponse>() {
+            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+            bulkRequestBuilder.add(deleteRequest);
+            bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+            bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
                 @Override
                 public void onResponse(BulkResponse bulkResponse) {
                     if (bulkResponse.getItems()[0].status() == RestStatus.NOT_FOUND) {

plugin/src/main/java/org/elasticsearch/xpack/ml/action/PutFilterAction.java

@@ -11,9 +11,8 @@ import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
-import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.bulk.TransportBulkAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
@@ -37,6 +36,7 @@ import org.elasticsearch.xpack.ml.MlMetaIndex;
 import org.elasticsearch.xpack.ml.job.config.MlFilter;
 import org.elasticsearch.xpack.ml.job.messages.Messages;
 import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.security.InternalClient;
 import org.elasticsearch.xpack.watcher.watch.Payload;
 
 import java.io.IOException;
@@ -161,16 +161,16 @@ public class PutFilterAction extends Action<PutFilterAction.Request, PutFilterAc
 
     public static class TransportAction extends HandledTransportAction<Request, Response> {
 
-        private final TransportBulkAction transportBulkAction;
+        private final InternalClient client;
 
         @Inject
         public TransportAction(Settings settings, ThreadPool threadPool,
-                TransportService transportService, ActionFilters actionFilters,
-                IndexNameExpressionResolver indexNameExpressionResolver,
-                TransportBulkAction transportBulkAction) {
+                               TransportService transportService, ActionFilters actionFilters,
+                               IndexNameExpressionResolver indexNameExpressionResolver,
+                               InternalClient client) {
             super(settings, NAME, threadPool, transportService, actionFilters,
                     indexNameExpressionResolver, Request::new);
-            this.transportBulkAction = transportBulkAction;
+            this.client = client;
         }
 
         @Override
@@ -183,10 +183,11 @@ public class PutFilterAction extends Action<PutFilterAction.Request, PutFilterAc
             } catch (IOException e) {
                 throw new IllegalStateException("Failed to serialise filter with id [" + filter.getId() + "]", e);
             }
-            BulkRequest bulkRequest = new BulkRequest().add(indexRequest);
-            bulkRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+            bulkRequestBuilder.add(indexRequest);
+            bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
 
-            transportBulkAction.execute(bulkRequest, new ActionListener<BulkResponse>() {
+            bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
                 @Override
                 public void onResponse(BulkResponse indexResponse) {
                     listener.onResponse(new Response());

plugin/src/main/java/org/elasticsearch/xpack/ml/action/RevertModelSnapshotAction.java

@@ -14,7 +14,6 @@ import org.elasticsearch.action.support.master.AcknowledgedRequest;
 import org.elasticsearch.action.support.master.AcknowledgedResponse;
 import org.elasticsearch.action.support.master.MasterNodeOperationRequestBuilder;
 import org.elasticsearch.action.support.master.TransportMasterNodeAction;
-import org.elasticsearch.client.Client;
 import org.elasticsearch.client.ElasticsearchClient;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.block.ClusterBlockException;
@@ -253,7 +252,7 @@ extends Action<RevertModelSnapshotAction.Request, RevertModelSnapshotAction.Resp
 
     public static class TransportAction extends TransportMasterNodeAction<Request, Response> {
 
-        private final Client client;
+        private final InternalClient client;
         private final JobManager jobManager;
         private final JobProvider jobProvider;
         private final JobDataCountsPersister jobDataCountsPersister;

plugin/src/main/java/org/elasticsearch/xpack/ml/action/UpdateModelSnapshotAction.java

@@ -12,12 +12,12 @@ import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
-import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.bulk.BulkResponse;
-import org.elasticsearch.action.bulk.TransportBulkAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.ElasticsearchClient;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.common.ParseField;
@@ -43,6 +43,7 @@ import org.elasticsearch.xpack.ml.job.persistence.JobProvider;
 import org.elasticsearch.xpack.ml.job.process.autodetect.state.ModelSnapshot;
 import org.elasticsearch.xpack.ml.job.results.Result;
 import org.elasticsearch.xpack.ml.utils.ExceptionsHelper;
+import org.elasticsearch.xpack.security.InternalClient;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -265,14 +266,14 @@ public class UpdateModelSnapshotAction extends Action<UpdateModelSnapshotAction.
     public static class TransportAction extends HandledTransportAction<Request, Response> {
 
         private final JobProvider jobProvider;
-        private final TransportBulkAction transportBulkAction;
+        private final InternalClient client;
 
         @Inject
         public TransportAction(Settings settings, TransportService transportService, ThreadPool threadPool, ActionFilters actionFilters,
-                IndexNameExpressionResolver indexNameExpressionResolver, JobProvider jobProvider, TransportBulkAction transportBulkAction) {
+                               IndexNameExpressionResolver indexNameExpressionResolver, JobProvider jobProvider, InternalClient client) {
             super(settings, NAME, threadPool, transportService, actionFilters, indexNameExpressionResolver, Request::new);
             this.jobProvider = jobProvider;
-            this.transportBulkAction = transportBulkAction;
+            this.client = client;
         }
 
         @Override
@@ -314,8 +315,10 @@ public class UpdateModelSnapshotAction extends Action<UpdateModelSnapshotAction.
                 errorHandler.accept(e);
                 return;
             }
-            BulkRequest bulkRequest = new BulkRequest().add(indexRequest);
-            transportBulkAction.execute(bulkRequest, new ActionListener<BulkResponse>() {
+            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
+            bulkRequestBuilder.add(indexRequest);
+            bulkRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+            bulkRequestBuilder.execute(new ActionListener<BulkResponse>() {
                 @Override
                 public void onResponse(BulkResponse indexResponse) {
                     handler.accept(true);

qa/smoke-test-ml-with-security/roles.yml

@@ -15,4 +15,3 @@ minimal:
         - indices:data/read/search
         - indices:data/write/bulk
         - indices:data/write/index
-        - indices:data/write/delete