diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java index d1ea5cfbd2..add50d64c2 100644 --- a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java +++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java @@ -88,6 +88,7 @@ public Collection> doExecute(Collection> records) { private void dissectField(Event event, String field){ Dissector dissector = dissectorMap.get(field); + boolean isDeleteSourceOnSuccessfulDissect = dissectConfig.isDeleteSourceRequested(); String text = event.get(field, String.class); if (dissector.dissectText(text)) { List dissectedFields = dissector.getDissectedFields(); @@ -96,6 +97,9 @@ private void dissectField(Event event, String field){ Object dissectFieldValue = convertTargetType(dissectFieldName,disectedField.getValue()); event.put(disectedField.getKey(), dissectFieldValue); } + if (isDeleteSourceOnSuccessfulDissect) { + event.delete(field); + } } } diff --git a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java index 84a1465cb2..265eeeafc1 100644 --- a/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java +++ b/data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorConfig.java @@ -54,6 +54,10 @@ public class DissectProcessorConfig { "For example, /some_value == \"log\".") private String dissectWhen; + @JsonProperty + @JsonPropertyDescription("If true, the configured fields in the map will be deleted if the processor was successful.") + private boolean deleteSource; + public String getDissectWhen(){ return dissectWhen; } @@ -64,6 +68,10 @@ public Map getMap() { public Map getTargetTypes() { return targetTypeMap; } + public boolean isDeleteSourceRequested() { + return deleteSource; + } + @AssertTrue(message = "target_type must be a Map. Valid options include [ 'integer', 'string', 'double', 'boolean', 'long', and 'big_decimal' ]") boolean isTargetTypeValid() { try { diff --git a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java index 9ab6dec585..8d63922ca9 100644 --- a/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java +++ b/data-prepper-plugins/dissect-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessorTest.java @@ -32,6 +32,7 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.when; @@ -267,4 +268,58 @@ private void reflectivelySetDissectorMap(DissectProcessor processor) throws NoSu reflectField.setAccessible(false); } } + + @Test + void test_delete_source_requested() throws NoSuchFieldException, IllegalAccessException { + + Field dissectedField = new NormalField("level"); + dissectedField.setValue("WARN"); + + when(dissector.dissectText(any(String.class))).thenReturn(true); + when(dissector.getDissectedFields()).thenReturn(List.of(dissectedField)); + when(dissectConfig.isDeleteSourceRequested()).thenReturn(true); + + final DissectProcessor processor = createObjectUnderTest(); + reflectivelySetDissectorMap(processor); + final Record dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log"); + final List> dissectedRecords = (List>) processor.doExecute(Collections.singletonList(dataPrepperRecord)); + + assertTrue(dissectedRecords.get(0).getData().containsKey("level")); + assertThat(dissectedRecords.get(0).getData().get("level", String.class), is("WARN")); + assertFalse(dissectedRecords.get(0).getData().containsKey("test")); + } + + @Test + void test_delete_source_not_requested() throws NoSuchFieldException, IllegalAccessException { + Field dissectedField = new NormalField("level"); + dissectedField.setValue("WARN"); + + when(dissector.dissectText(any(String.class))).thenReturn(true); + when(dissector.getDissectedFields()).thenReturn(List.of(dissectedField)); + when(dissectConfig.isDeleteSourceRequested()).thenReturn(false); + + final DissectProcessor processor = createObjectUnderTest(); + reflectivelySetDissectorMap(processor); + final Record dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log"); + final List> dissectedRecords = (List>) processor.doExecute(Collections.singletonList(dataPrepperRecord)); + + assertTrue(dissectedRecords.get(0).getData().containsKey("level")); + assertThat(dissectedRecords.get(0).getData().get("level", String.class), is("WARN")); + assertTrue(dissectedRecords.get(0).getData().containsKey("test")); + } + + @Test + void test_delete_source_requested_dissect_fail() throws NoSuchFieldException, IllegalAccessException { + + when(dissector.dissectText(any(String.class))).thenReturn(false); + when(dissectConfig.isDeleteSourceRequested()).thenReturn(true); + + final DissectProcessor processor = createObjectUnderTest(); + reflectivelySetDissectorMap(processor); + final Record dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log"); + final List> dissectedRecords = (List>) processor.doExecute(Collections.singletonList(dataPrepperRecord)); + + assertTrue(dissectedRecords.get(0).getData().containsKey("test")); + } + } \ No newline at end of file