Skip to content

Commit

Permalink
Per processor description for verbose simulate
Browse files Browse the repository at this point in the history 
For ingest node processors a per processor description
was recently added. This commit displays that description
in the verbose output of the pipeline simulation.

related elastic#57906
  • Loading branch information
@jakelandis
jakelandis committed Jun 16, 2020
1 parent 9b7d2a6 commit a8a9d87
Show file tree
Hide file tree
Showing 5 changed files with 77 additions and 35 deletions.
5 changes: 4 additions & 1 deletion modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/90_simulate.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -813,7 +813,7 @@ teardown:


---
"Test simulate with provided pipeline definition with description in processors":
"Test simulate with provided pipeline definition with tag and description in processors":
- do:
ingest.simulate:
verbose: true
Expand All @@ -825,6 +825,7 @@ teardown:
{
"set" : {
"description": "processor_description",
"tag": "processor_tag",
"field" : "field2",
"value" : "_value"
}
Expand All @@ -844,3 +845,5 @@ teardown:
- length: { docs: 1 }
- length: { docs.0.processor_results: 1 }
- match: { docs.0.processor_results.0.doc._source.field2: "_value" }
- match: { docs.0.processor_results.0.description: "processor_description" }
- match: { docs.0.processor_results.0.tag: "processor_tag" }
42 changes: 30 additions & 12 deletions server/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.elasticsearch.action.ingest;

import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.Version;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
Expand All @@ -39,6 +40,7 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {

private static final String IGNORED_ERROR_FIELD = "ignored_error";
private final String processorTag;
private final String description;
private final WriteableIngestDocument ingestDocument;
private final Exception failure;

Expand All @@ -62,18 +64,20 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {
true,
a -> {
String processorTag = a[0] == null ? null : (String)a[0];
IngestDocument document = a[1] == null ? null : ((WriteableIngestDocument)a[1]).getIngestDocument();
String description = a[1] == null ? null : (String)a[1];
IngestDocument document = a[2] == null ? null : ((WriteableIngestDocument)a[2]).getIngestDocument();
Exception failure = null;
if (a[2] != null) {
failure = (ElasticsearchException)a[2];
} else if (a[3] != null) {
if (a[3] != null) {
failure = (ElasticsearchException)a[3];
} else if (a[4] != null) {
failure = (ElasticsearchException)a[4];
}
return new SimulateProcessorResult(processorTag, document, failure);
return new SimulateProcessorResult(processorTag, description, document, failure);
}
);
static {
PARSER.declareString(optionalConstructorArg(), new ParseField(ConfigurationUtils.TAG_KEY));
PARSER.declareString(optionalConstructorArg(), new ParseField(ConfigurationUtils.DESCRIPTION_KEY));
PARSER.declareObject(
optionalConstructorArg(),
WriteableIngestDocument.INGEST_DOC_PARSER,
Expand All @@ -91,22 +95,24 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {
);
}

public SimulateProcessorResult(String processorTag, IngestDocument ingestDocument, Exception failure) {
public SimulateProcessorResult(String processorTag, String description, IngestDocument ingestDocument,
Exception failure) {
this.processorTag = processorTag;
this.description = description;
this.ingestDocument = (ingestDocument == null) ? null : new WriteableIngestDocument(ingestDocument);
this.failure = failure;
}

public SimulateProcessorResult(String processorTag, IngestDocument ingestDocument) {
this(processorTag, ingestDocument, null);
public SimulateProcessorResult(String processorTag, String description, IngestDocument ingestDocument) {
this(processorTag, description, ingestDocument, null);
}

public SimulateProcessorResult(String processorTag, Exception failure) {
this(processorTag, null, failure);
public SimulateProcessorResult(String processorTag, String description, Exception failure) {
this(processorTag, description, null, failure);
}

public SimulateProcessorResult(String processorTag) {
this(processorTag, null, null);
public SimulateProcessorResult(String processorTag, String description) {
this(processorTag, description, null, null);
}

/**
Expand All @@ -116,13 +122,21 @@ public SimulateProcessorResult(String processorTag) {
this.processorTag = in.readString();
this.ingestDocument = in.readOptionalWriteable(WriteableIngestDocument::new);
this.failure = in.readException();
if (in.getVersion().onOrAfter(Version.V_7_9_0)) {
this.description = in.readOptionalString();
} else {
this.description = null;
}
}

@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeString(processorTag);
out.writeOptionalWriteable(ingestDocument);
out.writeException(failure);
if (out.getVersion().onOrAfter(Version.V_7_9_0)) {
out.writeOptionalString(description);
}
}

public IngestDocument getIngestDocument() {
Expand Down Expand Up @@ -153,6 +167,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
builder.field(ConfigurationUtils.TAG_KEY, processorTag);
}

if (description != null) {
builder.field(ConfigurationUtils.DESCRIPTION_KEY, description);
}

if (failure != null && ingestDocument != null) {
builder.startObject(IGNORED_ERROR_FIELD);
ElasticsearchException.generateFailureXContent(builder, params, failure, true);
Expand Down
17 changes: 11 additions & 6 deletions server/src/main/java/org/elasticsearch/ingest/TrackingResultProcessor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,9 +66,10 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
if (elasticsearchException.getCause() instanceof IllegalStateException) {
if (ignoreFailure) {
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
new IngestDocument(ingestDocument), e));
pipelineProcessor.getDescription(), new IngestDocument(ingestDocument), e));
} else {
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(), e));
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
pipelineProcessor.getDescription(), e));
}
handler.accept(null, elasticsearchException);
}
Expand All @@ -86,17 +87,21 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
actualProcessor.execute(ingestDocument, (result, e) -> {
if (e != null) {
if (ignoreFailure) {
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument), e));
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), new IngestDocument(ingestDocument), e));
} else {
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), e));
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), e));
}
handler.accept(null, e);
} else {
if (result != null) {
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument)));
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), new IngestDocument(ingestDocument)));
handler.accept(result, null);
} else {
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag()));
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription()));
handler.accept(null, null);
}
}
Expand Down
9 changes: 6 additions & 3 deletions server/src/test/java/org/elasticsearch/action/ingest/SimulateProcessorResultTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,16 +70,19 @@ public void testSerialization() throws IOException {
static SimulateProcessorResult createTestInstance(boolean isSuccessful,
boolean isIgnoredException) {
String processorTag = randomAlphaOfLengthBetween(1, 10);
String description = randomAlphaOfLengthBetween(1, 10);
SimulateProcessorResult simulateProcessorResult;
if (isSuccessful) {
IngestDocument ingestDocument = createRandomIngestDoc();
if (isIgnoredException) {
simulateProcessorResult = new SimulateProcessorResult(processorTag, ingestDocument, new IllegalArgumentException("test"));
simulateProcessorResult = new SimulateProcessorResult(processorTag, description, ingestDocument,
new IllegalArgumentException("test"));
} else {
simulateProcessorResult = new SimulateProcessorResult(processorTag, ingestDocument);
simulateProcessorResult = new SimulateProcessorResult(processorTag, description, ingestDocument);
}
} else {
simulateProcessorResult = new SimulateProcessorResult(processorTag, new IllegalArgumentException("test"));
simulateProcessorResult = new SimulateProcessorResult(processorTag, description,
new IllegalArgumentException("test"));
}
return simulateProcessorResult;
}
Expand Down
39 changes: 26 additions & 13 deletions server/src/test/java/org/elasticsearch/ingest/TrackingResultProcessorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,8 @@ public void testActualProcessor() throws Exception {
TrackingResultProcessor trackingProcessor = new TrackingResultProcessor(false, actualProcessor, null, resultList);
trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);

assertThat(actualProcessor.getInvokedCounter(), equalTo(1));
assertThat(resultList.size(), equalTo(1));
Expand All @@ -87,7 +88,8 @@ public void testActualCompoundProcessorWithoutOnFailure() throws Exception {
trackingProcessor.execute(ingestDocument, (result, e) -> holder[0] = e);
assertThat(((IngestProcessorException) holder[0]).getRootCause().getMessage(), equalTo(exception.getMessage()));

SimulateProcessorResult expectedFirstResult = new SimulateProcessorResult(testProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedFirstResult = new SimulateProcessorResult(testProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
assertThat(testProcessor.getInvokedCounter(), equalTo(1));
assertThat(resultList.size(), equalTo(1));
assertThat(resultList.get(0).getIngestDocument(), nullValue());
Expand All @@ -107,8 +109,10 @@ public void testActualCompoundProcessorWithOnFailure() throws Exception {
CompoundProcessor trackingProcessor = decorate(actualProcessor, null, resultList);
trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(),
failProcessor.getDescription(), ingestDocument);
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(),
failProcessor.getDescription(), ingestDocument);

assertThat(failProcessor.getInvokedCounter(), equalTo(2));
assertThat(onFailureProcessor.getInvokedCounter(), equalTo(2));
Expand Down Expand Up @@ -159,8 +163,10 @@ public void testActualCompoundProcessorWithOnFailureAndTrueCondition() throws Ex
trackingProcessor.execute(ingestDocument, (result, e) -> {
});

SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(),
failProcessor.getDescription(), ingestDocument);
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(),
onFailureProcessor.getDescription(), ingestDocument);

assertThat(failProcessor.getInvokedCounter(), equalTo(1));
assertThat(onFailureProcessor.getInvokedCounter(), equalTo(1));
Expand Down Expand Up @@ -188,7 +194,8 @@ public void testActualCompoundProcessorWithIgnoreFailure() throws Exception {

trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedResult = new SimulateProcessorResult(testProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(testProcessor.getTag(),
testProcessor.getDescription(), ingestDocument);
assertThat(testProcessor.getInvokedCounter(), equalTo(1));
assertThat(resultList.size(), equalTo(1));
assertThat(resultList.get(0).getIngestDocument(), equalTo(expectedResult.getIngestDocument()));
Expand Down Expand Up @@ -218,7 +225,8 @@ public void testActualCompoundProcessorWithFalseConditional() throws Exception {

CompoundProcessor trackingProcessor = decorate(compoundProcessor, null, resultList);
trackingProcessor.execute(ingestDocument, (result, e) -> {});
SimulateProcessorResult expectedResult = new SimulateProcessorResult(compoundProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(compoundProcessor.getTag(),
compoundProcessor.getDescription(), ingestDocument);

//the step for key 2 is never executed due to conditional and thus not part of the result set
assertThat(resultList.size(), equalTo(2));
Expand Down Expand Up @@ -262,7 +270,8 @@ pipelineId, null, null, new CompoundProcessor(

trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);

verify(ingestService, Mockito.atLeast(1)).getPipeline(pipelineId);
Expand Down Expand Up @@ -331,7 +340,8 @@ pipelineId2, null, null, new CompoundProcessor(
trackingProcessor.execute(ingestDocument, (result, e) -> {});


SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId1);

verify(ingestService, Mockito.atLeast(1)).getPipeline(pipelineId1);
Expand Down Expand Up @@ -401,7 +411,8 @@ pipelineId2, null, null, new CompoundProcessor(
trackingProcessor.execute(ingestDocument, (result, e) -> {});


SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId1);

verify(ingestService, Mockito.atLeast(1)).getPipeline(pipelineId1);
Expand Down Expand Up @@ -453,7 +464,8 @@ pipelineId, null, null, new CompoundProcessor(

trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);

verify(ingestService, Mockito.atLeast(2)).getPipeline(pipelineId);
Expand Down Expand Up @@ -530,7 +542,8 @@ pipelineId, null, null, new CompoundProcessor(

trackingProcessor.execute(ingestDocument, (result, e) -> {});

SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
actualProcessor.getDescription(), ingestDocument);
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);

verify(ingestService, Mockito.atLeast(2)).getPipeline(pipelineId);
Expand Down

0 comments on commit a8a9d87

Please sign in to comment.