Skip to content

IngestService code cleanups #94593

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 22, 2023
Merged

IngestService code cleanups #94593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
84246d3
Silence some warnings from intellij
joegallo Mar 21, 2023
5b268d2
Rename and cleanup this method
joegallo Mar 21, 2023
738ebce
Tidy up these test assertions a bit
joegallo Mar 21, 2023
de68949
Make this class into a record
joegallo Mar 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 12 additions & 17 deletions server/src/main/java/org/elasticsearch/ingest/IngestService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -198,7 +198,7 @@ private static Map<String, Processor.Factory> processorFactories(List<IngestPlug
/**
* Resolves the potential pipelines (default and final) from the requests or templates associated to the index and then **mutates**
* the {@link org.elasticsearch.action.index.IndexRequest} passed object with the pipeline information.
*
* <p>
* Also, this method marks the request as `isPipelinesResolved = true`: Due to the request could be rerouted from a coordinating node
* to an ingest node, we have to be able to avoid double resolving the pipelines and also able to distinguish that either the pipeline
* comes as part of the request or resolved from this method. All this is made to later be able to reject the request in case the
Expand Down Expand Up @@ -476,10 +476,9 @@ Map<String, PipelineHolder> pipelines() {
* 'on_failure', so we report metrics for the set processor, not an on_failure processor.
*
* @param compoundProcessor The compound processor to start walking the non-failure processors
* @param processorMetrics The list of {@link Processor} {@link IngestMetric} tuples.
* @return the processorMetrics for all non-failure processor that belong to the original compoundProcessor
* @param processorMetrics The list to populate with {@link Processor} {@link IngestMetric} tuples.
*/
private static List<Tuple<Processor, IngestMetric>> getProcessorMetrics(
private static void collectProcessorMetrics(
CompoundProcessor compoundProcessor,
List<Tuple<Processor, IngestMetric>> processorMetrics
) {
Expand All @@ -505,12 +504,11 @@ private static List<Tuple<Processor, IngestMetric>> getProcessorMetrics(
} while (unwrapped);

if (processor instanceof CompoundProcessor cp) {
getProcessorMetrics(cp, processorMetrics);
collectProcessorMetrics(cp, processorMetrics);
} else {
processorMetrics.add(new Tuple<>(processor, metric));
}
}
return processorMetrics;
}

/**
Expand Down Expand Up @@ -840,7 +838,7 @@ public IngestStats stats() {
CompoundProcessor rootProcessor = pipeline.getCompoundProcessor();
statsBuilder.addPipelineMetrics(id, pipeline.getMetrics());
List<Tuple<Processor, IngestMetric>> processorMetrics = new ArrayList<>();
getProcessorMetrics(rootProcessor, processorMetrics);
collectProcessorMetrics(rootProcessor, processorMetrics);
processorMetrics.forEach(t -> {
Processor processor = t.v1();
IngestMetric processorMetric = t.v2();
Expand Down Expand Up @@ -1008,8 +1006,8 @@ synchronized void innerUpdatePipelines(IngestMetadata newIngestMetadata) {
newPipeline.getMetrics().add(oldPipeline.getMetrics());
List<Tuple<Processor, IngestMetric>> oldPerProcessMetrics = new ArrayList<>();
List<Tuple<Processor, IngestMetric>> newPerProcessMetrics = new ArrayList<>();
getProcessorMetrics(oldPipeline.getCompoundProcessor(), oldPerProcessMetrics);
getProcessorMetrics(newPipeline.getCompoundProcessor(), newPerProcessMetrics);
collectProcessorMetrics(oldPipeline.getCompoundProcessor(), oldPerProcessMetrics);
collectProcessorMetrics(newPipeline.getCompoundProcessor(), newPerProcessMetrics);
// Best attempt to populate new processor metrics using a parallel array of the old metrics. This is not ideal since
// the per processor metrics may get reset when the arrays don't match. However, to get to an ideal model, unique and
// consistent id's per processor and/or semantic equals for each processor will be needed.
Expand Down Expand Up @@ -1144,14 +1142,11 @@ public String getType() {
return new Pipeline(id, description, null, null, new CompoundProcessor(failureProcessor));
}

static class PipelineHolder {

final PipelineConfiguration configuration;
final Pipeline pipeline;
record PipelineHolder(PipelineConfiguration configuration, Pipeline pipeline) {

PipelineHolder(PipelineConfiguration configuration, Pipeline pipeline) {
this.configuration = Objects.requireNonNull(configuration);
this.pipeline = Objects.requireNonNull(pipeline);
public PipelineHolder {
Objects.requireNonNull(configuration);
Objects.requireNonNull(pipeline);
}
}

Expand Down Expand Up @@ -1236,7 +1231,7 @@ private static Optional<Pipelines> resolvePipelinesFromIndexTemplates(IndexReque

/**
* Checks whether an IndexRequest has at least one pipeline defined.
*
* <p>
* This method assumes that the pipelines are beforehand resolved.
*/
public static boolean hasPipeline(IndexRequest indexRequest) {
Expand Down
72 changes: 47 additions & 25 deletions server/src/test/java/org/elasticsearch/ingest/IngestServiceTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,10 +209,12 @@ public void testUpdatePipelines() {
.build();
ingestService.applyClusterState(new ClusterChangedEvent("", clusterState, previousClusterState));
assertThat(ingestService.pipelines().size(), is(1));
assertThat(ingestService.pipelines().get("_id").pipeline.getId(), equalTo("_id"));
assertThat(ingestService.pipelines().get("_id").pipeline.getDescription(), nullValue());
assertThat(ingestService.pipelines().get("_id").pipeline.getProcessors().size(), equalTo(1));
assertThat(ingestService.pipelines().get("_id").pipeline.getProcessors().get(0).getType(), equalTo("set"));

Pipeline p = ingestService.getPipeline("_id");
assertThat(p.getId(), equalTo("_id"));
assertThat(p.getDescription(), nullValue());
assertThat(p.getProcessors().size(), equalTo(1));
assertThat(p.getProcessors().get(0).getType(), equalTo("set"));
}

public void testInnerUpdatePipelines() {
Expand All @@ -224,51 +226,71 @@ public void testInnerUpdatePipelines() {

ingestService.innerUpdatePipelines(ingestMetadata);
assertThat(ingestService.pipelines().size(), is(1));
assertThat(ingestService.pipelines().get("_id1").pipeline.getId(), equalTo("_id1"));
assertThat(ingestService.pipelines().get("_id1").pipeline.getProcessors().size(), equalTo(0));
{
Pipeline p1 = ingestService.getPipeline("_id1");
assertThat(p1.getId(), equalTo("_id1"));
assertThat(p1.getProcessors().size(), equalTo(0));
}

PipelineConfiguration pipeline2 = new PipelineConfiguration("_id2", new BytesArray("{\"processors\": []}"), XContentType.JSON);
ingestMetadata = new IngestMetadata(Map.of("_id1", pipeline1, "_id2", pipeline2));

ingestService.innerUpdatePipelines(ingestMetadata);
assertThat(ingestService.pipelines().size(), is(2));
assertThat(ingestService.pipelines().get("_id1").pipeline.getId(), equalTo("_id1"));
assertThat(ingestService.pipelines().get("_id1").pipeline.getProcessors().size(), equalTo(0));
assertThat(ingestService.pipelines().get("_id2").pipeline.getId(), equalTo("_id2"));
assertThat(ingestService.pipelines().get("_id2").pipeline.getProcessors().size(), equalTo(0));
{
Pipeline p1 = ingestService.getPipeline("_id1");
assertThat(p1.getId(), equalTo("_id1"));
assertThat(p1.getProcessors().size(), equalTo(0));
Pipeline p2 = ingestService.getPipeline("_id2");
assertThat(p2.getId(), equalTo("_id2"));
assertThat(p2.getProcessors().size(), equalTo(0));
}

PipelineConfiguration pipeline3 = new PipelineConfiguration("_id3", new BytesArray("{\"processors\": []}"), XContentType.JSON);
ingestMetadata = new IngestMetadata(Map.of("_id1", pipeline1, "_id2", pipeline2, "_id3", pipeline3));

ingestService.innerUpdatePipelines(ingestMetadata);
assertThat(ingestService.pipelines().size(), is(3));
assertThat(ingestService.pipelines().get("_id1").pipeline.getId(), equalTo("_id1"));
assertThat(ingestService.pipelines().get("_id1").pipeline.getProcessors().size(), equalTo(0));
assertThat(ingestService.pipelines().get("_id2").pipeline.getId(), equalTo("_id2"));
assertThat(ingestService.pipelines().get("_id2").pipeline.getProcessors().size(), equalTo(0));
assertThat(ingestService.pipelines().get("_id3").pipeline.getId(), equalTo("_id3"));
assertThat(ingestService.pipelines().get("_id3").pipeline.getProcessors().size(), equalTo(0));
{
Pipeline p1 = ingestService.getPipeline("_id1");
assertThat(p1.getId(), equalTo("_id1"));
assertThat(p1.getProcessors().size(), equalTo(0));
Pipeline p2 = ingestService.getPipeline("_id2");
assertThat(p2.getId(), equalTo("_id2"));
assertThat(p2.getProcessors().size(), equalTo(0));
Pipeline p3 = ingestService.getPipeline("_id3");
assertThat(p3.getId(), equalTo("_id3"));
assertThat(p3.getProcessors().size(), equalTo(0));
}

ingestMetadata = new IngestMetadata(Map.of("_id1", pipeline1, "_id3", pipeline3));

ingestService.innerUpdatePipelines(ingestMetadata);
assertThat(ingestService.pipelines().size(), is(2));
assertThat(ingestService.pipelines().get("_id1").pipeline.getId(), equalTo("_id1"));
assertThat(ingestService.pipelines().get("_id1").pipeline.getProcessors().size(), equalTo(0));
assertThat(ingestService.pipelines().get("_id3").pipeline.getId(), equalTo("_id3"));
assertThat(ingestService.pipelines().get("_id3").pipeline.getProcessors().size(), equalTo(0));
{
Pipeline p1 = ingestService.getPipeline("_id1");
assertThat(p1.getId(), equalTo("_id1"));
assertThat(p1.getProcessors().size(), equalTo(0));
Pipeline p3 = ingestService.getPipeline("_id3");
assertThat(p3.getId(), equalTo("_id3"));
assertThat(p3.getProcessors().size(), equalTo(0));
}

pipeline3 = new PipelineConfiguration("_id3", new BytesArray("""
{"processors": [{"set" : {"field": "_field", "value": "_value"}}]}"""), XContentType.JSON);
ingestMetadata = new IngestMetadata(Map.of("_id1", pipeline1, "_id3", pipeline3));

ingestService.innerUpdatePipelines(ingestMetadata);
assertThat(ingestService.pipelines().size(), is(2));
assertThat(ingestService.pipelines().get("_id1").pipeline.getId(), equalTo("_id1"));
assertThat(ingestService.pipelines().get("_id1").pipeline.getProcessors().size(), equalTo(0));
assertThat(ingestService.pipelines().get("_id3").pipeline.getId(), equalTo("_id3"));
assertThat(ingestService.pipelines().get("_id3").pipeline.getProcessors().size(), equalTo(1));
assertThat(ingestService.pipelines().get("_id3").pipeline.getProcessors().get(0).getType(), equalTo("set"));
{
Pipeline p1 = ingestService.getPipeline("_id1");
assertThat(p1.getId(), equalTo("_id1"));
assertThat(p1.getProcessors().size(), equalTo(0));
Pipeline p3 = ingestService.getPipeline("_id3");
assertThat(p3.getId(), equalTo("_id3"));
assertThat(p3.getProcessors().size(), equalTo(1));
assertThat(p3.getProcessors().get(0).getType(), equalTo("set"));
}

// Perform an update with no changes:
Map<String, IngestService.PipelineHolder> pipelines = ingestService.pipelines();
Expand Down