added delete_source functionality to the dissect processor (#5662)

* completed feature request

Signed-off-by: Zeeshan Alam <Zeeshan.Alam+fidelity@fmr.com>

* added missing import statement

Signed-off-by: Zeeshan Alam <Zeeshan.Alam+fidelity@fmr.com>

---------

Signed-off-by: Zeeshan Alam <Zeeshan.Alam+fidelity@fmr.com>
Co-authored-by: Zeeshan Alam <Zeeshan.Alam+fidelity@fmr.com>

Author: alamzeeshan

data-prepper-plugins/dissect-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/dissect/DissectProcessor.java

@@ -90,6 +90,7 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> 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<Field> dissectedFields = dissector.getDissectedFields();
@@ -98,6 +99,9 @@ private void dissectField(Event event, String field){
                 Object dissectFieldValue = convertTargetType(dissectFieldName,disectedField.getValue());
                 event.put(disectedField.getKey(), dissectFieldValue);
             }
+            if (isDeleteSourceOnSuccessfulDissect) {
+                event.delete(field);
+            }
         }
     }
 

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, <code>/some_value == \"log\"</code>.")
     private String dissectWhen;
 
+    @JsonProperty
+    @JsonPropertyDescription("If true, the configured fields in the <code>map</code>  will be deleted if the processor was successful.")
+    private boolean deleteSource;
+
     public String getDissectWhen(){
         return dissectWhen;
     }
@@ -64,6 +68,10 @@ public Map<String, String> getMap() {
 
     public Map<String, TargetType> getTargetTypes() { return targetTypeMap; }
 
+    public boolean isDeleteSourceRequested() {
+        return deleteSource;
+    }
+
     @AssertTrue(message = "target_type must be a Map<String, TargetType>. Valid options include [ 'integer', 'string', 'double', 'boolean', 'long', and 'big_decimal' ]")
     boolean isTargetTypeValid() {
         try {

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<Event> dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) 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<Event> dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) 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<Event> dataPrepperRecord = getEvent("2025-01-28T00:00:00.000Z WARN This is a test log");
+        final List<Record<Event>> dissectedRecords = (List<Record<Event>>) processor.doExecute(Collections.singletonList(dataPrepperRecord));
+
+        assertTrue(dissectedRecords.get(0).getData().containsKey("test"));
+    }
+
 }