Skip to content

Commit

Permalink
INGEST: Move all Pipeline State into IngestService (#32617)
Browse files Browse the repository at this point in the history
* INGEST: Move all Pipeline State into IngestService

* Moves all pipeline state into the ingest service
   * Retains the existing pipeline store and pipeline execution service as inner classes to make the review easier, they should be flattened out in the next step
   * All tests for these classes were copied (and adapted) to the ingest service tests
* This is a refactoring step to enable a clean implementation of a pipeline processor (See #32473)
  • Loading branch information
original-brownbear authored Aug 21, 2018
1 parent 6d62d67 commit 8fc213f
Show file tree
Hide file tree
Showing 19 changed files with 1,437 additions and 1,480 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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
Expand All @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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);
}

Expand All @@ -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);
Expand Down
Loading

0 comments on commit 8fc213f

Please sign in to comment.