Skip to content

Commit c8142d9

Browse files
talevytalevy
committed
add exclude_keys option to KeyValueProcessor
and modify data-structure of `include_keys` and `exclude_keys` to be backed by a HashSet
1 parent 2a6e686 commit c8142d9

File tree

4 files changed

+65
-19
lines changed

4 files changed

+65
-19
lines changed

‎docs/reference/ingest/ingest-node.asciidoc

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1621,6 +1621,7 @@ For example, if you have a log message which contains `ip=1.2.3.4 error=REFUSED`
16211621
| `value_split` | yes | - | Regex pattern to use for splitting the key from the value within a key-value pair
16221622
| `target_field` | no | `null` | The field to insert the extracted keys into. Defaults to the root of the document
16231623
| `include_keys` | no | `null` | List of keys to filter and insert into document. Defaults to including all keys
1624+
| `exclude_keys` | no | `null` | List of keys to exclude from document
16241625
| `ignore_missing` | no | `false` | If `true` and `field` does not exist or is `null`, the processor quietly exits without modifying the document
16251626
|======
16261627

‎modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/KeyValueProcessor.java

Lines changed: 26 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -19,15 +19,18 @@
1919

2020
package org.elasticsearch.ingest.common;
2121

22+
import org.elasticsearch.common.util.set.Sets;
2223
import org.elasticsearch.ingest.AbstractProcessor;
2324
import org.elasticsearch.ingest.ConfigurationUtils;
2425
import org.elasticsearch.ingest.IngestDocument;
2526
import org.elasticsearch.ingest.Processor;
2627

2728
import java.util.Arrays;
2829
import java.util.Collections;
30+
import java.util.HashSet;
2931
import java.util.List;
3032
import java.util.Map;
33+
import java.util.Set;
3134

3235
/**
3336
* The KeyValueProcessor parses and extracts messages of the `key=value` variety into fields with values of the keys.
@@ -39,18 +42,20 @@ public final class KeyValueProcessor extends AbstractProcessor {
3942
private final String field;
4043
private final String fieldSplit;
4144
private final String valueSplit;
42-
private final List<String> includeKeys;
45+
private final Set<String> includeKeys;
46+
private final Set<String> excludeKeys;
4347
private final String targetField;
4448
private final boolean ignoreMissing;
4549

46-
KeyValueProcessor(String tag, String field, String fieldSplit, String valueSplit, List<String> includeKeys,
47-
String targetField, boolean ignoreMissing) {
50+
KeyValueProcessor(String tag, String field, String fieldSplit, String valueSplit, Set<String> includeKeys,
51+
Set<String> excludeKeys, String targetField, boolean ignoreMissing) {
4852
super(tag);
4953
this.field = field;
5054
this.targetField = targetField;
5155
this.fieldSplit = fieldSplit;
5256
this.valueSplit = valueSplit;
5357
this.includeKeys = includeKeys;
58+
this.excludeKeys = excludeKeys;
5459
this.ignoreMissing = ignoreMissing;
5560
}
5661

@@ -66,10 +71,14 @@ String getValueSplit() {
6671
return valueSplit;
6772
}
6873

69-
List<String> getIncludeKeys() {
74+
Set<String> getIncludeKeys() {
7075
return includeKeys;
7176
}
7277

78+
Set<String> getExcludeKeys() {
79+
return excludeKeys;
80+
}
81+
7382
String getTargetField() {
7483
return targetField;
7584
}
@@ -105,7 +114,9 @@ public void execute(IngestDocument document) {
105114
}
106115
return kv;
107116
})
108-
.filter((p) -> includeKeys == null || includeKeys.contains(p[0]))
117+
.filter((p) ->
118+
(includeKeys == null || includeKeys.contains(p[0])) &&
119+
(excludeKeys == null || excludeKeys.contains(p[0]) == false))
109120
.forEach((p) -> append(document, fieldPathPrefix + p[0], p[1]));
110121
}
111122

@@ -122,12 +133,18 @@ public KeyValueProcessor create(Map<String, Processor.Factory> registry, String
122133
String targetField = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "target_field");
123134
String fieldSplit = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field_split");
124135
String valueSplit = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "value_split");
125-
List<String> includeKeys = ConfigurationUtils.readOptionalList(TYPE, processorTag, config, "include_keys");
126-
if (includeKeys != null) {
127-
includeKeys = Collections.unmodifiableList(includeKeys);
136+
Set<String> includeKeys = null;
137+
Set<String> excludeKeys = null;
138+
List<String> includeKeysList = ConfigurationUtils.readOptionalList(TYPE, processorTag, config, "include_keys");
139+
if (includeKeysList != null) {
140+
includeKeys = Collections.unmodifiableSet(Sets.newHashSet(includeKeysList));
141+
}
142+
List<String> excludeKeysList = ConfigurationUtils.readOptionalList(TYPE, processorTag, config, "exclude_keys");
143+
if (excludeKeysList != null) {
144+
excludeKeys = Collections.unmodifiableSet(Sets.newHashSet(excludeKeysList));
128145
}
129146
boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_missing", false);
130-
return new KeyValueProcessor(processorTag, field, fieldSplit, valueSplit, includeKeys, targetField, ignoreMissing);
147+
return new KeyValueProcessor(processorTag, field, fieldSplit, valueSplit, includeKeys, excludeKeys, targetField, ignoreMissing);
131148
}
132149
}
133150
}

‎modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/KeyValueProcessorFactoryTests.java

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,11 @@
2121

2222
import org.elasticsearch.ElasticsearchException;
2323
import org.elasticsearch.ElasticsearchParseException;
24+
import org.elasticsearch.common.util.set.Sets;
2425
import org.elasticsearch.test.ESTestCase;
2526

2627
import java.util.Arrays;
28+
import java.util.Collections;
2729
import java.util.HashMap;
2830
import java.util.Map;
2931

@@ -58,14 +60,16 @@ public void testCreateWithAllFieldsSet() throws Exception {
5860
config.put("value_split", "=");
5961
config.put("target_field", "target");
6062
config.put("include_keys", Arrays.asList("a", "b"));
63+
config.put("exclude_keys", Collections.emptyList());
6164
config.put("ignore_missing", true);
6265
String processorTag = randomAlphaOfLength(10);
6366
KeyValueProcessor processor = factory.create(null, processorTag, config);
6467
assertThat(processor.getTag(), equalTo(processorTag));
6568
assertThat(processor.getField(), equalTo("field1"));
6669
assertThat(processor.getFieldSplit(), equalTo("&"));
6770
assertThat(processor.getValueSplit(), equalTo("="));
68-
assertThat(processor.getIncludeKeys(), equalTo(Arrays.asList("a", "b")));
71+
assertThat(processor.getIncludeKeys(), equalTo(Sets.newHashSet("a", "b")));
72+
assertThat(processor.getExcludeKeys(), equalTo(Collections.emptySet()));
6973
assertThat(processor.getTargetField(), equalTo("target"));
7074
assertTrue(processor.isIgnoreMissing());
7175
}

‎modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/KeyValueProcessorTests.java

Lines changed: 33 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.elasticsearch.ingest.common;
2121

22+
import org.elasticsearch.common.util.set.Sets;
2223
import org.elasticsearch.ingest.IngestDocument;
2324
import org.elasticsearch.ingest.Processor;
2425
import org.elasticsearch.ingest.RandomDocumentPicks;
@@ -36,7 +37,7 @@ public class KeyValueProcessorTests extends ESTestCase {
3637
public void test() throws Exception {
3738
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
3839
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "first=hello&second=world&second=universe");
39-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=", null, "target", false);
40+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=", null, null, "target", false);
4041
processor.execute(ingestDocument);
4142
assertThat(ingestDocument.getFieldValue("target.first", String.class), equalTo("hello"));
4243
assertThat(ingestDocument.getFieldValue("target.second", List.class), equalTo(Arrays.asList("world", "universe")));
@@ -45,7 +46,7 @@ public void test() throws Exception {
4546
public void testRootTarget() throws Exception {
4647
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.emptyMap());
4748
ingestDocument.setFieldValue("myField", "first=hello&second=world&second=universe");
48-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "myField", "&", "=", null, null, false);
49+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "myField", "&", "=", null, null,null, false);
4950
processor.execute(ingestDocument);
5051
assertThat(ingestDocument.getFieldValue("first", String.class), equalTo("hello"));
5152
assertThat(ingestDocument.getFieldValue("second", List.class), equalTo(Arrays.asList("world", "universe")));
@@ -54,7 +55,7 @@ public void testRootTarget() throws Exception {
5455
public void testKeySameAsSourceField() throws Exception {
5556
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.emptyMap());
5657
ingestDocument.setFieldValue("first", "first=hello");
57-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "first", "&", "=", null, null, false);
58+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "first", "&", "=", null, null,null, false);
5859
processor.execute(ingestDocument);
5960
assertThat(ingestDocument.getFieldValue("first", List.class), equalTo(Arrays.asList("first=hello", "hello")));
6061
}
@@ -63,15 +64,38 @@ public void testIncludeKeys() throws Exception {
6364
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
6465
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "first=hello&second=world&second=universe");
6566
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=",
66-
Collections.singletonList("first"), "target", false);
67+
Sets.newHashSet("first"), null, "target", false);
6768
processor.execute(ingestDocument);
6869
assertThat(ingestDocument.getFieldValue("target.first", String.class), equalTo("hello"));
6970
assertFalse(ingestDocument.hasField("target.second"));
7071
}
7172

73+
public void testExcludeKeys() throws Exception {
74+
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
75+
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "first=hello&second=world&second=universe");
76+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=",
77+
null, Sets.newHashSet("second"), "target", false);
78+
processor.execute(ingestDocument);
79+
assertThat(ingestDocument.getFieldValue("target.first", String.class), equalTo("hello"));
80+
assertFalse(ingestDocument.hasField("target.second"));
81+
}
82+
83+
public void testIncludeAndExcludeKeys() throws Exception {
84+
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
85+
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument,
86+
"first=hello&second=world&second=universe&third=bar");
87+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=",
88+
Sets.newHashSet("first", "second"), Sets.newHashSet("first", "second"), "target", false);
89+
processor.execute(ingestDocument);
90+
assertFalse(ingestDocument.hasField("target.first"));
91+
assertFalse(ingestDocument.hasField("target.second"));
92+
assertFalse(ingestDocument.hasField("target.third"));
93+
}
94+
7295
public void testMissingField() {
7396
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.emptyMap());
74-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "unknown", "&", "=", null, "target", false);
97+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "unknown", "&",
98+
"=", null, null, "target", false);
7599
IllegalArgumentException exception = expectThrows(IllegalArgumentException.class, () -> processor.execute(ingestDocument));
76100
assertThat(exception.getMessage(), equalTo("field [unknown] not present as part of path [unknown]"));
77101
}
@@ -81,31 +105,31 @@ public void testNullValueWithIgnoreMissing() throws Exception {
81105
IngestDocument originalIngestDocument = RandomDocumentPicks.randomIngestDocument(random(),
82106
Collections.singletonMap(fieldName, null));
83107
IngestDocument ingestDocument = new IngestDocument(originalIngestDocument);
84-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "", "", null, "target", true);
108+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "", "", null, null, "target", true);
85109
processor.execute(ingestDocument);
86110
assertIngestDocument(originalIngestDocument, ingestDocument);
87111
}
88112

89113
public void testNonExistentWithIgnoreMissing() throws Exception {
90114
IngestDocument originalIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.emptyMap());
91115
IngestDocument ingestDocument = new IngestDocument(originalIngestDocument);
92-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "unknown", "", "", null, "target", true);
116+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "unknown", "", "", null, null, "target", true);
93117
processor.execute(ingestDocument);
94118
assertIngestDocument(originalIngestDocument, ingestDocument);
95119
}
96120

97121
public void testFailFieldSplitMatch() throws Exception {
98122
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
99123
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, "first=hello|second=world|second=universe");
100-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=", null, "target", false);
124+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), fieldName, "&", "=", null, null, "target", false);
101125
processor.execute(ingestDocument);
102126
assertThat(ingestDocument.getFieldValue("target.first", String.class), equalTo("hello|second=world|second=universe"));
103127
assertFalse(ingestDocument.hasField("target.second"));
104128
}
105129

106130
public void testFailValueSplitMatch() throws Exception {
107131
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.singletonMap("foo", "bar"));
108-
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "foo", "&", "=", null, "target", false);
132+
Processor processor = new KeyValueProcessor(randomAlphaOfLength(10), "foo", "&", "=", null, null, "target", false);
109133
Exception exception = expectThrows(IllegalArgumentException.class, () -> processor.execute(ingestDocument));
110134
assertThat(exception.getMessage(), equalTo("field [foo] does not contain value_split [=]"));
111135
}

0 commit comments

Comments
 (0)