INGEST: Move all Pipeline State into IngestService

server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

@@ -521,7 +521,7 @@ private long relativeTime() {
     void processBulkIndexIngestRequest(Task task, BulkRequest original, ActionListener<BulkResponse> listener) {
         long ingestStartTimeInNanos = System.nanoTime();
         BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(original);
-        ingestService.getPipelineExecutionService().executeBulkRequest(() -> bulkRequestModifier, (indexRequest, exception) -> {
+        ingestService.executeBulkRequest(() -> bulkRequestModifier, (indexRequest, exception) -> {
             logger.debug(() -> new ParameterizedMessage("failed to execute pipeline [{}] for document [{}/{}/{}]",
                 indexRequest.getPipeline(), indexRequest.index(), indexRequest.type(), indexRequest.id()), exception);
             bulkRequestModifier.markCurrentItemAsFailed(exception);

server/src/main/java/org/elasticsearch/action/ingest/DeletePipelineTransportAction.java

@@ -27,26 +27,23 @@
 import org.elasticsearch.cluster.block.ClusterBlockException;
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
-import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.ingest.PipelineStore;
-import org.elasticsearch.node.NodeService;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 public class DeletePipelineTransportAction extends TransportMasterNodeAction<DeletePipelineRequest, AcknowledgedResponse> {
 
-    private final PipelineStore pipelineStore;
-    private final ClusterService clusterService;
+    private final IngestService ingestService;
 
     @Inject
-    public DeletePipelineTransportAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
+    public DeletePipelineTransportAction(Settings settings, ThreadPool threadPool, IngestService ingestService,
                                          TransportService transportService, ActionFilters actionFilters,
-                                         IndexNameExpressionResolver indexNameExpressionResolver, NodeService nodeService) {
-        super(settings, DeletePipelineAction.NAME, transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver, DeletePipelineRequest::new);
-        this.clusterService = clusterService;
-        this.pipelineStore = nodeService.getIngestService().getPipelineStore();
+                                         IndexNameExpressionResolver indexNameExpressionResolver) {
+        super(settings, DeletePipelineAction.NAME, transportService, ingestService.getClusterService(),
+            threadPool, actionFilters, indexNameExpressionResolver, DeletePipelineRequest::new);
+        this.ingestService = ingestService;
     }
 
     @Override
@@ -60,8 +57,9 @@ protected AcknowledgedResponse newResponse() {
     }
 
     @Override
-    protected void masterOperation(DeletePipelineRequest request, ClusterState state, ActionListener<AcknowledgedResponse> listener) throws Exception {
-        pipelineStore.delete(clusterService, request, listener);
+    protected void masterOperation(DeletePipelineRequest request, ClusterState state,
+        ActionListener<AcknowledgedResponse> listener) throws Exception {
+        ingestService.delete(request, listener);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/ingest/GetPipelineTransportAction.java

@@ -29,21 +29,17 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.ingest.PipelineStore;
-import org.elasticsearch.node.NodeService;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 public class GetPipelineTransportAction extends TransportMasterNodeReadAction<GetPipelineRequest, GetPipelineResponse> {
-
-    private final PipelineStore pipelineStore;
-
+
     @Inject
     public GetPipelineTransportAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                       TransportService transportService, ActionFilters actionFilters,
-                                      IndexNameExpressionResolver indexNameExpressionResolver, NodeService nodeService) {
+                                      IndexNameExpressionResolver indexNameExpressionResolver) {
         super(settings, GetPipelineAction.NAME, transportService, clusterService, threadPool, actionFilters, GetPipelineRequest::new, indexNameExpressionResolver);
-        this.pipelineStore = nodeService.getIngestService().getPipelineStore();
     }
 
     @Override
@@ -58,7 +54,7 @@ protected GetPipelineResponse newResponse() {
 
     @Override
     protected void masterOperation(GetPipelineRequest request, ClusterState state, ActionListener<GetPipelineResponse> listener) throws Exception {
-        listener.onResponse(new GetPipelineResponse(pipelineStore.getPipelines(state, request.getIds())));
+        listener.onResponse(new GetPipelineResponse(IngestService.getPipelines(state, request.getIds())));
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/ingest/PutPipelineTransportAction.java

@@ -32,12 +32,10 @@
 import org.elasticsearch.cluster.block.ClusterBlockLevel;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.ingest.PipelineStore;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.IngestInfo;
-import org.elasticsearch.node.NodeService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
@@ -46,19 +44,19 @@
 
 public class PutPipelineTransportAction extends TransportMasterNodeAction<PutPipelineRequest, AcknowledgedResponse> {
 
-    private final PipelineStore pipelineStore;
-    private final ClusterService clusterService;
+    private final IngestService ingestService;
     private final NodeClient client;
 
     @Inject
-    public PutPipelineTransportAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
-                                      TransportService transportService, ActionFilters actionFilters,
-                                      IndexNameExpressionResolver indexNameExpressionResolver, NodeService nodeService,
-                                      NodeClient client) {
-        super(settings, PutPipelineAction.NAME, transportService, clusterService, threadPool, actionFilters, indexNameExpressionResolver, PutPipelineRequest::new);
-        this.clusterService = clusterService;
+    public PutPipelineTransportAction(Settings settings, ThreadPool threadPool, TransportService transportService,
+        ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
+        IngestService ingestService, NodeClient client) {
+        super(
+            settings, PutPipelineAction.NAME, transportService, ingestService.getClusterService(),
+            threadPool, actionFilters, indexNameExpressionResolver, PutPipelineRequest::new
+        );
         this.client = client;
-        this.pipelineStore = nodeService.getIngestService().getPipelineStore();
+        this.ingestService = ingestService;
     }
 
     @Override
@@ -84,7 +82,7 @@ public void onResponse(NodesInfoResponse nodeInfos) {
                     for (NodeInfo nodeInfo : nodeInfos.getNodes()) {
                         ingestInfos.put(nodeInfo.getNode(), nodeInfo.getIngest());
                     }
-                    pipelineStore.put(clusterService, ingestInfos, request, listener);
+                    ingestService.putPipeline(ingestInfos, request, listener);
                 } catch (Exception e) {
                     onFailure(e);
                 }

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequest.java

@@ -32,8 +32,8 @@
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.Pipeline;
-import org.elasticsearch.ingest.PipelineStore;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -164,24 +164,23 @@ public boolean isVerbose() {
         }
     }
 
-    private static final Pipeline.Factory PIPELINE_FACTORY = new Pipeline.Factory();
     static final String SIMULATED_PIPELINE_ID = "_simulate_pipeline";
 
-    static Parsed parseWithPipelineId(String pipelineId, Map<String, Object> config, boolean verbose, PipelineStore pipelineStore) {
+    static Parsed parseWithPipelineId(String pipelineId, Map<String, Object> config, boolean verbose, IngestService ingestService) {
         if (pipelineId == null) {
             throw new IllegalArgumentException("param [pipeline] is null");
         }
-        Pipeline pipeline = pipelineStore.get(pipelineId);
+        Pipeline pipeline = ingestService.getPipeline(pipelineId);
         if (pipeline == null) {
             throw new IllegalArgumentException("pipeline [" + pipelineId + "] does not exist");
         }
         List<IngestDocument> ingestDocumentList = parseDocs(config);
         return new Parsed(pipeline, ingestDocumentList, verbose);
     }
 
-    static Parsed parse(Map<String, Object> config, boolean verbose, PipelineStore pipelineStore) throws Exception {
+    static Parsed parse(Map<String, Object> config, boolean verbose, IngestService pipelineStore) throws Exception {
         Map<String, Object> pipelineConfig = ConfigurationUtils.readMap(null, null, config, Fields.PIPELINE);
-        Pipeline pipeline = PIPELINE_FACTORY.create(SIMULATED_PIPELINE_ID, pipelineConfig, pipelineStore.getProcessorFactories());
+        Pipeline pipeline = Pipeline.create(SIMULATED_PIPELINE_ID, pipelineConfig, pipelineStore.getProcessorFactories());
         List<IngestDocument> ingestDocumentList = parseDocs(config);
         return new Parsed(pipeline, ingestDocumentList, verbose);
     }

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineTransportAction.java

@@ -26,8 +26,7 @@
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
-import org.elasticsearch.ingest.PipelineStore;
-import org.elasticsearch.node.NodeService;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -36,15 +35,15 @@
 
 public class SimulatePipelineTransportAction extends HandledTransportAction<SimulatePipelineRequest, SimulatePipelineResponse> {
 
-    private final PipelineStore pipelineStore;
+    private final IngestService ingestService;
     private final SimulateExecutionService executionService;
 
     @Inject
     public SimulatePipelineTransportAction(Settings settings, ThreadPool threadPool, TransportService transportService,
-                                           ActionFilters actionFilters, NodeService nodeService) {
+                                           ActionFilters actionFilters, IngestService ingestService) {
         super(settings, SimulatePipelineAction.NAME, transportService, actionFilters,
             (Writeable.Reader<SimulatePipelineRequest>) SimulatePipelineRequest::new);
-        this.pipelineStore = nodeService.getIngestService().getPipelineStore();
+        this.ingestService = ingestService;
         this.executionService = new SimulateExecutionService(threadPool);
     }
 
@@ -55,9 +54,9 @@ protected void doExecute(Task task, SimulatePipelineRequest request, ActionListe
         final SimulatePipelineRequest.Parsed simulateRequest;
         try {
             if (request.getId() != null) {
-                simulateRequest = SimulatePipelineRequest.parseWithPipelineId(request.getId(), source, request.isVerbose(), pipelineStore);
+                simulateRequest = SimulatePipelineRequest.parseWithPipelineId(request.getId(), source, request.isVerbose(), ingestService);
             } else {
-                simulateRequest = SimulatePipelineRequest.parse(source, request.isVerbose(), pipelineStore);
+                simulateRequest = SimulatePipelineRequest.parse(source, request.isVerbose(), ingestService);
             }
         } catch (Exception e) {
             listener.onFailure(e);