Skip to content

Commit 2d28f71

Browse files
jakelandisjakelandis
authored
Per processor description for verbose simulate (#58207)
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 #57906
1 parent f269738 commit 2d28f71

File tree

5 files changed

+98
-35
lines changed

5 files changed

+98
-35
lines changed

‎modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/90_simulate.yml‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -813,7 +813,7 @@ teardown:
813813

814814

815815
---
816-
"Test simulate with provided pipeline definition with description in processors":
816+
"Test simulate with provided pipeline definition with tag and description in processors":
817817
- do:
818818
ingest.simulate:
819819
verbose: true
@@ -825,6 +825,7 @@ teardown:
825825
{
826826
"set" : {
827827
"description": "processor_description",
828+
"tag": "processor_tag",
828829
"field" : "field2",
829830
"value" : "_value"
830831
}
@@ -844,3 +845,5 @@ teardown:
844845
- length: { docs: 1 }
845846
- length: { docs.0.processor_results: 1 }
846847
- match: { docs.0.processor_results.0.doc._source.field2: "_value" }
848+
- match: { docs.0.processor_results.0.description: "processor_description" }
849+
- match: { docs.0.processor_results.0.tag: "processor_tag" }

‎server/src/main/java/org/elasticsearch/action/ingest/SimulateProcessorResult.java‎

Lines changed: 34 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.elasticsearch.action.ingest;
2020

2121
import org.elasticsearch.ElasticsearchException;
22+
import org.elasticsearch.Version;
2223
import org.elasticsearch.common.ParseField;
2324
import org.elasticsearch.common.io.stream.StreamInput;
2425
import org.elasticsearch.common.io.stream.StreamOutput;
@@ -39,6 +40,7 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {
3940

4041
private static final String IGNORED_ERROR_FIELD = "ignored_error";
4142
private final String processorTag;
43+
private final String description;
4244
private final WriteableIngestDocument ingestDocument;
4345
private final Exception failure;
4446

@@ -62,18 +64,20 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {
6264
true,
6365
a -> {
6466
String processorTag = a[0] == null ? null : (String)a[0];
65-
IngestDocument document = a[1] == null ? null : ((WriteableIngestDocument)a[1]).getIngestDocument();
67+
String description = a[1] == null ? null : (String)a[1];
68+
IngestDocument document = a[2] == null ? null : ((WriteableIngestDocument)a[2]).getIngestDocument();
6669
Exception failure = null;
67-
if (a[2] != null) {
68-
failure = (ElasticsearchException)a[2];
69-
} else if (a[3] != null) {
70+
if (a[3] != null) {
7071
failure = (ElasticsearchException)a[3];
72+
} else if (a[4] != null) {
73+
failure = (ElasticsearchException)a[4];
7174
}
72-
return new SimulateProcessorResult(processorTag, document, failure);
75+
return new SimulateProcessorResult(processorTag, description, document, failure);
7376
}
7477
);
7578
static {
7679
PARSER.declareString(optionalConstructorArg(), new ParseField(ConfigurationUtils.TAG_KEY));
80+
PARSER.declareString(optionalConstructorArg(), new ParseField(ConfigurationUtils.DESCRIPTION_KEY));
7781
PARSER.declareObject(
7882
optionalConstructorArg(),
7983
WriteableIngestDocument.INGEST_DOC_PARSER,
@@ -91,22 +95,24 @@ public class SimulateProcessorResult implements Writeable, ToXContentObject {
9195
);
9296
}
9397

94-
public SimulateProcessorResult(String processorTag, IngestDocument ingestDocument, Exception failure) {
98+
public SimulateProcessorResult(String processorTag, String description, IngestDocument ingestDocument,
99+
Exception failure) {
95100
this.processorTag = processorTag;
101+
this.description = description;
96102
this.ingestDocument = (ingestDocument == null) ? null : new WriteableIngestDocument(ingestDocument);
97103
this.failure = failure;
98104
}
99105

100-
public SimulateProcessorResult(String processorTag, IngestDocument ingestDocument) {
101-
this(processorTag, ingestDocument, null);
106+
public SimulateProcessorResult(String processorTag, String description, IngestDocument ingestDocument) {
107+
this(processorTag, description, ingestDocument, null);
102108
}
103109

104-
public SimulateProcessorResult(String processorTag, Exception failure) {
105-
this(processorTag, null, failure);
110+
public SimulateProcessorResult(String processorTag, String description, Exception failure) {
111+
this(processorTag, description, null, failure);
106112
}
107113

108-
public SimulateProcessorResult(String processorTag) {
109-
this(processorTag, null, null);
114+
public SimulateProcessorResult(String processorTag, String description) {
115+
this(processorTag, description, null, null);
110116
}
111117

112118
/**
@@ -116,13 +122,21 @@ public SimulateProcessorResult(String processorTag) {
116122
this.processorTag = in.readString();
117123
this.ingestDocument = in.readOptionalWriteable(WriteableIngestDocument::new);
118124
this.failure = in.readException();
125+
if (in.getVersion().onOrAfter(Version.V_7_9_0)) {
126+
this.description = in.readOptionalString();
127+
} else {
128+
this.description = null;
129+
}
119130
}
120131

121132
@Override
122133
public void writeTo(StreamOutput out) throws IOException {
123134
out.writeString(processorTag);
124135
out.writeOptionalWriteable(ingestDocument);
125136
out.writeException(failure);
137+
if (out.getVersion().onOrAfter(Version.V_7_9_0)) {
138+
out.writeOptionalString(description);
139+
}
126140
}
127141

128142
public IngestDocument getIngestDocument() {
@@ -140,6 +154,10 @@ public Exception getFailure() {
140154
return failure;
141155
}
142156

157+
public String getDescription() {
158+
return description;
159+
}
160+
143161
@Override
144162
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
145163
if (processorTag == null && failure == null && ingestDocument == null) {
@@ -153,6 +171,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
153171
builder.field(ConfigurationUtils.TAG_KEY, processorTag);
154172
}
155173

174+
if (description != null) {
175+
builder.field(ConfigurationUtils.DESCRIPTION_KEY, description);
176+
}
177+
156178
if (failure != null && ingestDocument != null) {
157179
builder.startObject(IGNORED_ERROR_FIELD);
158180
ElasticsearchException.generateFailureXContent(builder, params, failure, true);

‎server/src/main/java/org/elasticsearch/ingest/TrackingResultProcessor.java‎

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -66,9 +66,10 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
6666
if (elasticsearchException.getCause() instanceof IllegalStateException) {
6767
if (ignoreFailure) {
6868
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
69-
new IngestDocument(ingestDocument), e));
69+
pipelineProcessor.getDescription(), new IngestDocument(ingestDocument), e));
7070
} else {
71-
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(), e));
71+
processorResultList.add(new SimulateProcessorResult(pipelineProcessor.getTag(),
72+
pipelineProcessor.getDescription(), e));
7273
}
7374
handler.accept(null, elasticsearchException);
7475
}
@@ -86,17 +87,21 @@ public void execute(IngestDocument ingestDocument, BiConsumer<IngestDocument, Ex
8687
actualProcessor.execute(ingestDocument, (result, e) -> {
8788
if (e != null) {
8889
if (ignoreFailure) {
89-
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument), e));
90+
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
91+
actualProcessor.getDescription(), new IngestDocument(ingestDocument), e));
9092
} else {
91-
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), e));
93+
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
94+
actualProcessor.getDescription(), e));
9295
}
9396
handler.accept(null, e);
9497
} else {
9598
if (result != null) {
96-
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(), new IngestDocument(ingestDocument)));
99+
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
100+
actualProcessor.getDescription(), new IngestDocument(ingestDocument)));
97101
handler.accept(result, null);
98102
} else {
99-
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag()));
103+
processorResultList.add(new SimulateProcessorResult(actualProcessor.getTag(),
104+
actualProcessor.getDescription()));
100105
handler.accept(null, null);
101106
}
102107
}

‎server/src/test/java/org/elasticsearch/action/ingest/SimulateProcessorResultTests.java‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,11 +19,13 @@
1919

2020
package org.elasticsearch.action.ingest;
2121

22+
import org.elasticsearch.Version;
2223
import org.elasticsearch.common.io.stream.BytesStreamOutput;
2324
import org.elasticsearch.common.io.stream.StreamInput;
2425
import org.elasticsearch.common.xcontent.XContentParser;
2526
import org.elasticsearch.ingest.IngestDocument;
2627
import org.elasticsearch.test.AbstractXContentTestCase;
28+
import org.elasticsearch.test.VersionUtils;
2729

2830
import java.io.IOException;
2931
import java.util.StringJoiner;
@@ -50,6 +52,7 @@ public void testSerialization() throws IOException {
5052
StreamInput streamInput = out.bytes().streamInput();
5153
SimulateProcessorResult otherSimulateProcessorResult = new SimulateProcessorResult(streamInput);
5254
assertThat(otherSimulateProcessorResult.getProcessorTag(), equalTo(simulateProcessorResult.getProcessorTag()));
55+
assertThat(otherSimulateProcessorResult.getDescription(), equalTo(simulateProcessorResult.getDescription()));
5356
if (isSuccessful) {
5457
assertIngestDocument(otherSimulateProcessorResult.getIngestDocument(), simulateProcessorResult.getIngestDocument());
5558
if (isIgnoredException) {
@@ -67,19 +70,36 @@ public void testSerialization() throws IOException {
6770
}
6871
}
6972

73+
public void testBWCDescription() throws IOException {
74+
boolean isSuccessful = randomBoolean();
75+
boolean isIgnoredException = randomBoolean();
76+
SimulateProcessorResult simulateProcessorResult = createTestInstance(isSuccessful, isIgnoredException);
77+
78+
BytesStreamOutput out = new BytesStreamOutput();
79+
out.setVersion(VersionUtils.getPreviousVersion(Version.V_7_9_0));
80+
simulateProcessorResult.writeTo(out);
81+
StreamInput in = out.bytes().streamInput();
82+
in.setVersion(VersionUtils.getPreviousVersion(Version.V_7_9_0));
83+
SimulateProcessorResult otherSimulateProcessorResult = new SimulateProcessorResult(in);
84+
assertNull(otherSimulateProcessorResult.getDescription());
85+
}
86+
7087
static SimulateProcessorResult createTestInstance(boolean isSuccessful,
7188
boolean isIgnoredException) {
7289
String processorTag = randomAlphaOfLengthBetween(1, 10);
90+
String description = randomAlphaOfLengthBetween(1, 10);
7391
SimulateProcessorResult simulateProcessorResult;
7492
if (isSuccessful) {
7593
IngestDocument ingestDocument = createRandomIngestDoc();
7694
if (isIgnoredException) {
77-
simulateProcessorResult = new SimulateProcessorResult(processorTag, ingestDocument, new IllegalArgumentException("test"));
95+
simulateProcessorResult = new SimulateProcessorResult(processorTag, description, ingestDocument,
96+
new IllegalArgumentException("test"));
7897
} else {
79-
simulateProcessorResult = new SimulateProcessorResult(processorTag, ingestDocument);
98+
simulateProcessorResult = new SimulateProcessorResult(processorTag, description, ingestDocument);
8099
}
81100
} else {
82-
simulateProcessorResult = new SimulateProcessorResult(processorTag, new IllegalArgumentException("test"));
101+
simulateProcessorResult = new SimulateProcessorResult(processorTag, description,
102+
new IllegalArgumentException("test"));
83103
}
84104
return simulateProcessorResult;
85105
}

‎server/src/test/java/org/elasticsearch/ingest/TrackingResultProcessorTests.java‎

Lines changed: 26 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,8 @@ public void testActualProcessor() throws Exception {
6767
TrackingResultProcessor trackingProcessor = new TrackingResultProcessor(false, actualProcessor, null, resultList);
6868
trackingProcessor.execute(ingestDocument, (result, e) -> {});
6969

70-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
70+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
71+
actualProcessor.getDescription(), ingestDocument);
7172

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

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

110-
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(), ingestDocument);
111-
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(), ingestDocument);
112+
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(),
113+
failProcessor.getDescription(), ingestDocument);
114+
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(),
115+
failProcessor.getDescription(), ingestDocument);
112116

113117
assertThat(failProcessor.getInvokedCounter(), equalTo(2));
114118
assertThat(onFailureProcessor.getInvokedCounter(), equalTo(2));
@@ -159,8 +163,10 @@ public void testActualCompoundProcessorWithOnFailureAndTrueCondition() throws Ex
159163
trackingProcessor.execute(ingestDocument, (result, e) -> {
160164
});
161165

162-
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(), ingestDocument);
163-
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(), ingestDocument);
166+
SimulateProcessorResult expectedFailResult = new SimulateProcessorResult(failProcessor.getTag(),
167+
failProcessor.getDescription(), ingestDocument);
168+
SimulateProcessorResult expectedSuccessResult = new SimulateProcessorResult(onFailureProcessor.getTag(),
169+
onFailureProcessor.getDescription(), ingestDocument);
164170

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

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

191-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(testProcessor.getTag(), ingestDocument);
197+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(testProcessor.getTag(),
198+
testProcessor.getDescription(), ingestDocument);
192199
assertThat(testProcessor.getInvokedCounter(), equalTo(1));
193200
assertThat(resultList.size(), equalTo(1));
194201
assertThat(resultList.get(0).getIngestDocument(), equalTo(expectedResult.getIngestDocument()));
@@ -218,7 +225,8 @@ public void testActualCompoundProcessorWithFalseConditional() throws Exception {
218225

219226
CompoundProcessor trackingProcessor = decorate(compoundProcessor, null, resultList);
220227
trackingProcessor.execute(ingestDocument, (result, e) -> {});
221-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(compoundProcessor.getTag(), ingestDocument);
228+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(compoundProcessor.getTag(),
229+
compoundProcessor.getDescription(), ingestDocument);
222230

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

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

265-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
273+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
274+
actualProcessor.getDescription(), ingestDocument);
266275
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);
267276

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

333342

334-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
343+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
344+
actualProcessor.getDescription(), ingestDocument);
335345
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId1);
336346

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

403413

404-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
414+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
415+
actualProcessor.getDescription(), ingestDocument);
405416
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId1);
406417

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

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

456-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
467+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
468+
actualProcessor.getDescription(), ingestDocument);
457469
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);
458470

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

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

533-
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(), ingestDocument);
545+
SimulateProcessorResult expectedResult = new SimulateProcessorResult(actualProcessor.getTag(),
546+
actualProcessor.getDescription(), ingestDocument);
534547
expectedResult.getIngestDocument().getIngestMetadata().put("pipeline", pipelineId);
535548

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

0 commit comments

Comments
 (0)