Skip to content

Commit a681971

Browse files
niketan16niketan16
authored
Add support for multiple entries of 'with_keys' with 'delete_when' co… (#5356)
Add support for multiple entries of 'with_keys' with 'delete_when' condition in delete_entries processor Signed-off-by: Niketan Chandarana <niketanc@amazon.com>
1 parent 5762fe2 commit a681971

4 files changed

Lines changed: 183 additions & 20 deletions

File tree

‎data-prepper-plugins/mutate-event-processors/build.gradle‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,4 +24,5 @@ dependencies {
2424
implementation 'com.fasterxml.jackson.core:jackson-databind'
2525
testImplementation project(':data-prepper-test-event')
2626
testImplementation testLibs.slf4j.simple
27+
testImplementation testLibs.spring.test
2728
}

‎data-prepper-plugins/mutate-event-processors/src/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/DeleteEntryProcessor.java‎

Lines changed: 33 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -28,23 +28,42 @@
2828
public class DeleteEntryProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
2929

3030
private static final Logger LOG = LoggerFactory.getLogger(DeleteEntryProcessor.class);
31-
private final List<EventKey> entries;
31+
private final List<EventKey> withKeys;
3232
private final String deleteWhen;
33+
private final List<DeleteEntryProcessorConfig.Entry> entries;
3334

3435
private final ExpressionEvaluator expressionEvaluator;
3536

3637
@DataPrepperPluginConstructor
3738
public DeleteEntryProcessor(final PluginMetrics pluginMetrics, final DeleteEntryProcessorConfig config, final ExpressionEvaluator expressionEvaluator) {
3839
super(pluginMetrics);
39-
this.entries = config.getWithKeys();
40+
this.withKeys = config.getWithKeys();
4041
this.deleteWhen = config.getDeleteWhen();
4142
this.expressionEvaluator = expressionEvaluator;
4243

4344
if (deleteWhen != null
44-
&& !expressionEvaluator.isValidExpressionStatement(deleteWhen)) {
45-
throw new InvalidPluginConfigurationException(
46-
String.format("delete_when %s is not a valid expression statement. See https://opensearch.org/docs/latest/data-prepper/pipelines/expression-syntax/ for valid expression syntax", deleteWhen));
45+
&& !expressionEvaluator.isValidExpressionStatement(deleteWhen)) {
46+
throw new InvalidPluginConfigurationException(
47+
String.format("delete_when %s is not a valid expression statement. See https://opensearch" +
48+
".org/docs/latest/data-prepper/pipelines/expression-syntax/ for valid expression syntax", deleteWhen));
49+
}
50+
51+
if (this.withKeys != null && !this.withKeys.isEmpty()) {
52+
DeleteEntryProcessorConfig.Entry entry = new DeleteEntryProcessorConfig.Entry(this.withKeys, this.deleteWhen);
53+
this.entries = List.of(entry);
54+
} else {
55+
this.entries = config.getEntries();
4756
}
57+
58+
this.entries.forEach(entry -> {
59+
if (entry.getDeleteWhen() != null
60+
&& !expressionEvaluator.isValidExpressionStatement(entry.getDeleteWhen())) {
61+
throw new InvalidPluginConfigurationException(
62+
String.format("delete_when %s is not a valid expression statement. See https://opensearch" +
63+
".org/docs/latest/data-prepper/pipelines/expression-syntax/ for valid expression syntax",
64+
entry.getDeleteWhen()));
65+
}
66+
});
4867
}
4968

5069
@Override
@@ -53,13 +72,15 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
5372
final Event recordEvent = record.getData();
5473

5574
try {
56-
if (Objects.nonNull(deleteWhen) && !expressionEvaluator.evaluateConditional(deleteWhen, recordEvent)) {
57-
continue;
58-
}
59-
60-
61-
for (final EventKey entry : entries) {
62-
recordEvent.delete(entry);
75+
for (final DeleteEntryProcessorConfig.Entry entry : entries) {
76+
if (Objects.nonNull(entry.getDeleteWhen()) && !expressionEvaluator.evaluateConditional(entry.getDeleteWhen(),
77+
recordEvent)) {
78+
continue;
79+
}
80+
81+
for (final EventKey key : entry.getWithKeys()) {
82+
recordEvent.delete(key);
83+
}
6384
}
6485
} catch (final Exception e) {
6586
LOG.atError()

‎data-prepper-plugins/mutate-event-processors/src/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/DeleteEntryProcessorConfig.java‎

Lines changed: 73 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -9,8 +9,14 @@
99
import com.fasterxml.jackson.annotation.JsonProperty;
1010
import com.fasterxml.jackson.annotation.JsonPropertyDescription;
1111
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
12+
import jakarta.validation.Valid;
13+
import jakarta.validation.constraints.AssertFalse;
14+
import jakarta.validation.constraints.AssertTrue;
1215
import jakarta.validation.constraints.NotEmpty;
1316
import jakarta.validation.constraints.NotNull;
17+
import org.opensearch.dataprepper.model.annotations.ConditionalRequired;
18+
import org.opensearch.dataprepper.model.annotations.ConditionalRequired.IfThenElse;
19+
import org.opensearch.dataprepper.model.annotations.ConditionalRequired.SchemaProperty;
1420
import org.opensearch.dataprepper.model.annotations.ExampleValues;
1521
import org.opensearch.dataprepper.model.annotations.ExampleValues.Example;
1622
import org.opensearch.dataprepper.model.event.EventKey;
@@ -19,31 +25,90 @@
1925

2026
import java.util.List;
2127

28+
@ConditionalRequired(value = {
29+
@IfThenElse(
30+
ifFulfilled = {@SchemaProperty(field = "entries", value = "null")},
31+
thenExpect = {@SchemaProperty(field = "with_keys")}
32+
),
33+
@IfThenElse(
34+
ifFulfilled = {@SchemaProperty(field = "with_keys", value = "null")},
35+
thenExpect = {@SchemaProperty(field = "entries")}
36+
)
37+
})
2238
@JsonPropertyOrder
2339
@JsonClassDescription("The <code>delete_entries</code> processor deletes fields from events. " +
2440
"You can define the keys you want to delete in the <code>with_keys</code> configuration. " +
2541
"Those keys and their values are deleted from events.")
2642
public class DeleteEntryProcessorConfig {
27-
@NotEmpty
28-
@NotNull
43+
44+
@JsonPropertyOrder
45+
public static class Entry {
46+
@NotEmpty
47+
@NotNull
48+
@JsonProperty("with_keys")
49+
@EventKeyConfiguration(EventKeyFactory.EventAction.DELETE)
50+
@JsonPropertyDescription("A list of keys to be deleted.")
51+
private List<@NotNull @NotEmpty EventKey> withKeys;
52+
53+
@JsonProperty("delete_when")
54+
@JsonPropertyDescription("Specifies under what condition the deletion should be performed. " +
55+
"By default, keys are always deleted. Example: <code>/mykey == \"---\"</code>")
56+
@ExampleValues({
57+
@Example(value = "/some_key == null", description = "Only runs the deletion if the key some_key is null or does not exist.")
58+
})
59+
private String deleteWhen;
60+
61+
public List<EventKey> getWithKeys() {
62+
return withKeys;
63+
}
64+
65+
public String getDeleteWhen() {
66+
return deleteWhen;
67+
}
68+
69+
public Entry(final List<EventKey> withKeys, final String deleteWhen) {
70+
this.withKeys = withKeys;
71+
this.deleteWhen = deleteWhen;
72+
}
73+
74+
public Entry() {
75+
76+
}
77+
}
78+
2979
@JsonProperty("with_keys")
3080
@EventKeyConfiguration(EventKeyFactory.EventAction.DELETE)
31-
@JsonPropertyDescription("A list of keys to be deleted.")
81+
@JsonPropertyDescription("A list of keys to be deleted. May not be used with entries.")
3282
private List<@NotNull @NotEmpty EventKey> withKeys;
3383

3484
@JsonProperty("delete_when")
35-
@JsonPropertyDescription("Specifies under what condition the <code>delete_entries</code> processor should perform deletion. " +
36-
"By default, keys are always deleted. Example: <code>/mykey == \"---\"</code>")
37-
@ExampleValues({
38-
@Example(value = "/some_key == null", description = "Only runs the delete_entries processor on the Event if the key some_key is null or does not exist.")
39-
})
85+
@JsonPropertyDescription("Specifies under what condition the deletion should be performed.")
4086
private String deleteWhen;
4187

88+
@Valid
89+
@JsonProperty("entries")
90+
@JsonPropertyDescription("A list of entries to delete from the event.")
91+
private List<Entry> entries;
92+
93+
@AssertTrue(message = "Either 'entries' or 'with_keys' must be specified, but neither was found")
94+
boolean isConfigurationPresent() {
95+
return entries != null || withKeys != null;
96+
}
97+
98+
@AssertFalse(message = "Either use 'entries' OR 'with_keys' with 'delete_when' configuration, but not both")
99+
boolean hasBothConfigurations() {
100+
return entries != null && withKeys != null;
101+
}
102+
42103
public List<EventKey> getWithKeys() {
43104
return withKeys;
44105
}
45106

46107
public String getDeleteWhen() {
47108
return deleteWhen;
48109
}
110+
111+
public List<Entry> getEntries() {
112+
return entries;
113+
}
49114
}

‎data-prepper-plugins/mutate-event-processors/src/test/java/org/opensearch/dataprepper/plugins/processor/mutateevent/DeleteEntryProcessorTests.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
import org.opensearch.dataprepper.model.event.JacksonEvent;
1818
import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
1919
import org.opensearch.dataprepper.model.record.Record;
20+
import org.springframework.test.util.ReflectionTestUtils;
2021

2122
import java.util.Collections;
2223
import java.util.HashMap;
@@ -134,6 +135,81 @@ public void testNestedDeleteProcessorTest() {
134135
assertThat(editedRecords.get(0).getData().containsKey("message"), is(true));
135136
}
136137

138+
@Test
139+
public void test_multiple_entries_with_different_delete_when_conditions() {
140+
final DeleteEntryProcessorConfig.Entry entry1 = new DeleteEntryProcessorConfig.Entry(List.of(eventKeyFactory.createEventKey("key1"
141+
, EventKeyFactory.EventAction.DELETE)), "condition1");
142+
final DeleteEntryProcessorConfig.Entry entry2 = new DeleteEntryProcessorConfig.Entry(List.of(eventKeyFactory.createEventKey("key2"
143+
, EventKeyFactory.EventAction.DELETE)), "condition2");
144+
145+
when(mockConfig.getEntries()).thenReturn(List.of(entry1, entry2));
146+
when(expressionEvaluator.isValidExpressionStatement("condition1")).thenReturn(true);
147+
when(expressionEvaluator.isValidExpressionStatement("condition2")).thenReturn(true);
148+
149+
final DeleteEntryProcessor processor = createObjectUnderTest();
150+
final Record<Event> record = getEvent("test");
151+
record.getData().put("key1", "value1");
152+
record.getData().put("key2", "value2");
153+
154+
when(expressionEvaluator.evaluateConditional("condition1", record.getData())).thenReturn(true);
155+
when(expressionEvaluator.evaluateConditional("condition2", record.getData())).thenReturn(false);
156+
157+
final List<Record<Event>> editedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
158+
159+
assertThat(editedRecords.get(0).getData().containsKey("key1"), is(false));
160+
assertThat(editedRecords.get(0).getData().containsKey("key2"), is(true));
161+
}
162+
163+
@Test
164+
public void test_legacy_format_conversion_to_entries_format() {
165+
when(mockConfig.getWithKeys()).thenReturn(List.of(eventKeyFactory.createEventKey("message", EventKeyFactory.EventAction.DELETE)));
166+
when(mockConfig.getDeleteWhen()).thenReturn("condition");
167+
when(expressionEvaluator.isValidExpressionStatement("condition")).thenReturn(true);
168+
169+
final DeleteEntryProcessor processor = createObjectUnderTest();
170+
final Record<Event> record = getEvent("test");
171+
172+
when(expressionEvaluator.evaluateConditional("condition", record.getData())).thenReturn(true);
173+
174+
final List<Record<Event>> editedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(record));
175+
176+
assertThat(editedRecords.get(0).getData().containsKey("message"), is(false));
177+
}
178+
179+
@Test
180+
public void invalid_delete_when_with_entries_format_throws_InvalidPluginConfigurationException() {
181+
DeleteEntryProcessorConfig.Entry entry = new DeleteEntryProcessorConfig.Entry(List.of(eventKeyFactory.createEventKey("key1",
182+
EventKeyFactory.EventAction.DELETE)), "invalid_condition");
183+
184+
when(mockConfig.getEntries()).thenReturn(List.of(entry));
185+
when(expressionEvaluator.isValidExpressionStatement("invalid_condition")).thenReturn(false);
186+
187+
assertThrows(InvalidPluginConfigurationException.class, this::createObjectUnderTest);
188+
}
189+
190+
@Test
191+
public void test_both_configurations_used_together() {
192+
final DeleteEntryProcessorConfig configObjectUnderTest = new DeleteEntryProcessorConfig();
193+
final DeleteEntryProcessorConfig.Entry entry = new DeleteEntryProcessorConfig.Entry(List.of(eventKeyFactory.createEventKey("key1"
194+
, EventKeyFactory.EventAction.DELETE)), "condition");
195+
196+
ReflectionTestUtils.setField(configObjectUnderTest, "withKeys", List.of(eventKeyFactory.createEventKey("message",
197+
EventKeyFactory.EventAction.DELETE)));
198+
ReflectionTestUtils.setField(configObjectUnderTest, "entries", List.of(entry));
199+
200+
assertThat(configObjectUnderTest.hasBothConfigurations(), is(true));
201+
}
202+
203+
@Test
204+
public void test_no_configuration_used() {
205+
final DeleteEntryProcessorConfig configObjectUnderTest = new DeleteEntryProcessorConfig();
206+
207+
ReflectionTestUtils.setField(configObjectUnderTest, "withKeys", null);
208+
ReflectionTestUtils.setField(configObjectUnderTest, "entries", null);
209+
210+
assertThat(configObjectUnderTest.isConfigurationPresent(), is(false));
211+
}
212+
137213
private DeleteEntryProcessor createObjectUnderTest() {
138214
return new DeleteEntryProcessor(pluginMetrics, mockConfig, expressionEvaluator);
139215
}

0 commit comments

Comments
 (0)