Write json processor

data-prepper-plugins/write-json-processor/build.gradle

@@ -0,0 +1,20 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java'
+}
+
+dependencies {
+    implementation project(':data-prepper-api')
+    implementation project(':data-prepper-plugins:common')
+    implementation 'io.micrometer:micrometer-core'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-ion'
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-xml'
+    implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.0'
+    testImplementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.0'
+    testImplementation project(path: ':data-prepper-test-common')
+}

data-prepper-plugins/write-json-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/write_json/WriteJsonProcessor.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.write_json;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import io.micrometer.core.instrument.Counter;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.plugin.PluginFactory;
+
+import static org.opensearch.dataprepper.logging.DataPrepperMarkers.EVENT;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.processor.AbstractProcessor;
+import org.opensearch.dataprepper.model.processor.Processor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+
+@DataPrepperPlugin(name = "write_json", pluginType = Processor.class, pluginConfigurationType = WriteJsonProcessorConfig.class)
+public class WriteJsonProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
+    private static final Logger LOG = LoggerFactory.getLogger(WriteJsonProcessor.class);
+    private static final String WRITE_JSON_FAILED_COUNTER = "writeJsonFailedCounter";
+    private final String source;
+    private final Counter writeJsonFailedCounter;
+    private String target;
+    private final ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+
+    @DataPrepperPluginConstructor
+    public WriteJsonProcessor(final WriteJsonProcessorConfig writeJsonProcessorConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory) {
+        super(pluginMetrics);
+        source = writeJsonProcessorConfig.getSource();
+        target = writeJsonProcessorConfig.getTarget();
+        if (target == null) {
+            target = source;
+        }
+        writeJsonFailedCounter = pluginMetrics.counter(WRITE_JSON_FAILED_COUNTER);
+    }
+
+    @Override
+    public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
+        for (final Record<Event> record : records) {
+            final Event event = record.getData();
+            Object value = event.get(source, Object.class);
+            if (value != null) {
+                try {
+                    event.put(target, objectMapper.writeValueAsString(value));
+                } catch (Exception e) {
+                    LOG.error(EVENT, "Failed to convert source to json string", e);
+                    writeJsonFailedCounter.increment();
+                }
+            }
+        }
+        return records;
+    }
+
+    @Override
+    public void prepareForShutdown() {
+
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+
+    }
+}

data-prepper-plugins/write-json-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/write_json/WriteJsonProcessorConfig.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.write_json;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import jakarta.validation.constraints.NotNull;
+
+public class WriteJsonProcessorConfig {
+    @JsonProperty("source")
+    @NotNull
+    private String source;
+
+    @JsonProperty("target")
+    private String target;
+
+    public String getSource() {
+        return source;
+    }
+
+    public String getTarget() {
+        return target;
+    }
+}

data-prepper-plugins/write-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/write_json/WriteJsonProcessorConfigTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.write_json;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.test.helper.ReflectivelySetField;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.util.UUID;
+
+public class WriteJsonProcessorConfigTest {
+    WriteJsonProcessorConfig writeJsonProcessorConfig;
+
+    @BeforeEach
+    void setup() {
+        writeJsonProcessorConfig = new WriteJsonProcessorConfig();
+    }
+
+    @Test
+    public void testDefaults() {
+        assertThat(writeJsonProcessorConfig.getTarget(), equalTo(null));
+    }
+
+    @Test
+    public void testParameterSets() throws Exception {
+        String sourceKey = UUID.randomUUID().toString();
+        String targetKey = UUID.randomUUID().toString();
+        ReflectivelySetField.setField(WriteJsonProcessorConfig.class, writeJsonProcessorConfig, "source", sourceKey);
+        ReflectivelySetField.setField(WriteJsonProcessorConfig.class, writeJsonProcessorConfig, "target", targetKey);
+        assertThat(writeJsonProcessorConfig.getSource(), equalTo(sourceKey));
+        assertThat(writeJsonProcessorConfig.getTarget(), equalTo(targetKey));
+    }
+}

data-prepper-plugins/write-json-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/write_json/WriteJsonProcessorTest.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor.write_json;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.plugin.PluginFactory;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.mock;
+import org.mockito.Mock;
+
+import java.util.Map;
+import java.util.List;
+import java.util.UUID;
+
+public class WriteJsonProcessorTest {
+    private final ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
+
+    @Mock
+    private WriteJsonProcessorConfig mockConfig;
+
+    @Mock
+    private PluginFactory pluginFactory;
+
+    @Mock
+    private PluginMetrics pluginMetrics;
+
+    private WriteJsonProcessor writeJsonProcessor;
+
+    private String sourceKey;
+
+    @BeforeEach
+    void setup() {
+        pluginMetrics = mock(PluginMetrics.class);
+        pluginFactory = mock(PluginFactory.class);
+        sourceKey = UUID.randomUUID().toString();
+        mockConfig = mock(WriteJsonProcessorConfig.class);
+        when(mockConfig.getSource()).thenReturn(sourceKey);
+    }
+
+    WriteJsonProcessor createObjectUnderTest() {
+        return new WriteJsonProcessor(mockConfig, pluginMetrics, pluginFactory);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {"", "targetKey"})
+    public void testBasic(String target) throws Exception {
+        String targetKey = (target.equals("")) ? sourceKey : target;
+        when(mockConfig.getTarget()).thenReturn(targetKey);
+        Map<String, Object> value1 = Map.of("stringKey", "testString", "intKey", 10);
+        Map<String, Object> value = Map.of("mapKey", value1, "boolKey", true);
+        String expectedString1 = "{\"mapKey\":{\"stringKey\":\"testString\",\"intKey\":10},\"boolKey\":true}";
+        String expectedString2 = "{\"boolKey\":true,\"mapKey\":{\"stringKey\":\"testString\",\"intKey\":10}}";
+        String expectedString3 = "{\"boolKey\":true,\"mapKey\":{\"intKey\":10,\"stringKey\":\"testString\"}}";
+        String expectedString4 = "{\"boolKey\":true,\"mapKey\":{\"intKey\":10,\"stringKey\":\"testString\"}}";
+
+
+        Map<String, Object> data = Map.of(sourceKey, value);
+        Record<Event> record = createRecord(data);
+
+        writeJsonProcessor = createObjectUnderTest();
+        final List<Record<Event>> outputRecords = (List<Record<Event>>) writeJsonProcessor.doExecute(List.of(record));
+        assertThat(outputRecords.size(), equalTo(1));
+        Event event = outputRecords.get(0).getData();
+        String expectedResult = objectMapper.writeValueAsString(value);
+        String targetInEvent = event.get(targetKey, String.class);
+        assertThat(targetInEvent, equalTo(expectedResult));
+        assertTrue(expectedString1.equals(targetInEvent)||expectedString2.equals(targetInEvent)||expectedString3.equals(targetInEvent)||expectedString4.equals(targetInEvent));
+
+    }
+
+    static Record<Event> createRecord(final Map<String, Object> data) {
+        return new Record<>(JacksonEvent.builder()
+            .withData(data)
+            .withEventType("event")
+            .build());
+    }
+
+}

settings.gradle

@@ -123,6 +123,7 @@ include 'data-prepper-plugins:geoip-processor'
 include 'data-prepper-plugins:grok-processor'
 include 'data-prepper-plugins:aggregate-processor'
 include 'data-prepper-plugins:log-generator-source'
+include 'data-prepper-plugins:write-json-processor'
 include 'data-prepper-logstash-configuration'
 include 'e2e-test'
 include 'e2e-test:trace'