Merge branch 'main' of https://github.com/jeffreyAaron/data-prepper

Author: jeffreyAaron

build.gradle

@@ -72,7 +72,6 @@ subprojects {
         implementation platform('com.fasterxml.jackson:jackson-bom:2.17.2')
         implementation platform('org.eclipse.jetty:jetty-bom:9.4.53.v20231009')
         implementation platform('io.micrometer:micrometer-bom:1.10.5')
-        implementation 'com.github.luben:zstd-jni:1.5.7-3'
         implementation libs.guava.core
         implementation libs.slf4j.api
         testImplementation testLibs.bundles.junit

…nk/s3/compression/CompressionEngine.java …epper/model/codec/CompressionEngine.javadata-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/compression/CompressionEngine.java renamed to data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/CompressionEngine.java data-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/compression/CompressionEngine.java renamed to data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/CompressionEngine.java

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.plugins.sink.s3.compression;
+package org.opensearch.dataprepper.model.codec;
 
 import java.io.IOException;
 import java.io.OutputStream;

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEngine.java

@@ -0,0 +1,24 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEngine {
+    /**
+     * Encrypts raw data into {@link EncryptionEnvelope}.
+     *
+     * @param data the raw data in bytes
+     * @return returns the encryption envelope
+     */
+    EncryptionEnvelope encrypt(byte[] data);
+
+    /**
+     * Decrypts the encryption envelope into raw data.
+     *
+     * @param encryptionEnvelope the encryption envelope
+     * @return returns the raw data in bytes
+     */
+    byte[] decrypt(EncryptionEnvelope encryptionEnvelope);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/EncryptionEnvelope.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+public interface EncryptionEnvelope {
+    /**
+     * The encrypted data.
+     */
+    byte[] getEncryptedData();
+
+    /**
+     * The encrypted data key.
+     */
+    String getEncryptedDataKey();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/encryption/KeyProvider.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.encryption;
+
+@FunctionalInterface
+public interface KeyProvider {
+    byte[] decryptKey(byte[] encryptedKey);
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

@@ -61,6 +61,8 @@ public class JacksonEvent implements Event {
 
     private static final Logger LOG = LoggerFactory.getLogger(JacksonEvent.class);
 
+    private static final int FILL_OUT_OF_BOUNDS_ELEMENTS_LIMIT = 0;
+
     private static final String SEPARATOR = "/";
 
     private static final ObjectMapper mapper = JsonMapper.builder()
@@ -195,8 +197,31 @@ private JsonNode getOrCreateNode(final JsonNode node, final String key) {
         JsonNode childNode = node.get(key);
         if (childNode == null) {
             childNode = mapper.createObjectNode();
-            ((ObjectNode) node).set(key, childNode);
+            if (node.isArray()) {
+                int index = Integer.parseInt(key);
+                ArrayNode arrayNode = (ArrayNode) node;
+
+                int distanceFromArrayEnd = index - arrayNode.size();
+                if (distanceFromArrayEnd >= FILL_OUT_OF_BOUNDS_ELEMENTS_LIMIT + 1) {
+                    throw new IndexOutOfBoundsException(
+                            String.format("Cannot expand array past the limit of size %s to reach index %s", arrayNode.size(), index));
+                }
+                while (arrayNode.size() <= index) {
+                    arrayNode.addNull();
+                }
+
+                JsonNode existing = arrayNode.get(index);
+                if (existing == null || !existing.isObject()) {
+                    childNode = mapper.createObjectNode();
+                    arrayNode.set(index, childNode);
+                } else {
+                    childNode = existing;
+                }
+            } else {
+                ((ObjectNode) node).set(key, childNode);
+            }
         }
+
         return childNode;
     }
 

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/JacksonEventTest.java

@@ -18,6 +18,7 @@
 
 import java.math.BigDecimal;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -147,6 +148,81 @@ void testPutAndGet_withStrings_eventKey(final String key) {
         assertThat(result, is(equalTo(value)));
     }
 
+    @Test
+    void testPutAndGet_withArrays_eventKey() {
+
+        final String key = "list-key/0/foo";
+        final String newValue = UUID.randomUUID().toString();
+
+        final List<Map<String, Object>> listValue = new ArrayList<>();
+        final Map<String, Object> mapValue = Map.of("foo", "bar", "foo-2", "bar-2");
+        listValue.add(mapValue);
+
+        final String listKey = "list-key";
+        final EventKey eventKey = new JacksonEventKey(listKey);
+        event.put(eventKey, listValue);
+
+        final Map<String, Object> expectedMap = new HashMap<>();
+        expectedMap.put(listKey, listValue);
+
+        assertThat(event.toMap(), equalTo(expectedMap));
+
+        final EventKey eventNestedKey = new JacksonEventKey(key);
+        event.put(eventNestedKey, newValue);
+
+        final List<Map<String, Object>> newlistValue = new ArrayList<>();
+        final Map<String, Object> newMapValue = Map.of("foo", newValue, "foo-2", "bar-2");
+        newlistValue.add(newMapValue);
+
+        expectedMap.put(listKey, newlistValue);
+
+        assertThat(event.toMap(), equalTo(expectedMap));
+
+
+        final List<Map<String, Object>> result = event.get(listKey, List.class);
+        assertThat(result, equalTo(newlistValue));
+
+        final String resultValue = event.get(key, String.class);
+        assertThat(resultValue, equalTo(newValue));
+    }
+
+    @Test
+    void testPutAndGet_withArrays_out_of_bounds_on_end_of_list_creates_new_element() {
+
+        final String key = "list-key/1/foo";
+        final String fooValue = UUID.randomUUID().toString();
+
+        final List<Map<String, Object>> listValue = new ArrayList<>();
+        final Map<String, Object> mapValue = Map.of("foo", "bar", "foo-2", "bar-2");
+        listValue.add(mapValue);
+
+        final String listKey = "list-key";
+        final EventKey eventKey = new JacksonEventKey(listKey);
+        event.put(eventKey, listValue);
+
+        event.put(key, fooValue);
+
+        final String resultValue = event.get(key, String.class);
+        assertThat(resultValue, equalTo(fooValue));
+    }
+
+    @Test
+    void testPutAndGet_withArrays_out_of_bounds_throws_IndexOutOfBoundsException() {
+
+        final String key = "list-key/3/foo";
+        final String fooValue = UUID.randomUUID().toString();
+
+        final List<Map<String, Object>> listValue = new ArrayList<>();
+        final Map<String, Object> mapValue = Map.of("foo", "bar", "foo-2", "bar-2");
+        listValue.add(mapValue);
+
+        final String listKey = "list-key";
+        final EventKey eventKey = new JacksonEventKey(listKey);
+        event.put(eventKey, listValue);
+
+        assertThrows(IndexOutOfBoundsException.class, () -> event.put(key, fooValue));
+    }
+
     @Test
     public void testPutKeyCannotBeEmptyString() {
         Throwable exception = assertThrows(IllegalArgumentException.class, () -> event.put("", "value"));
@@ -1117,5 +1193,4 @@ private static Stream<Arguments> getBigDecimalPutTestData() {
                 Arguments.of("1.000")
         );
     }
-
 }

data-prepper-core/build.gradle

@@ -23,6 +23,7 @@ dependencies {
     implementation project(':data-prepper-event')
     implementation project(':data-prepper-plugins:blocking-buffer')
     implementation project(':data-prepper-plugins:common')
+    implementation project(':data-prepper-plugins:encryption-plugin')
     implementation project(':data-prepper-logstash-configuration')
     implementation project(':data-prepper-pipeline-parser')
     implementation project(':data-prepper-plugin-framework')

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/pipeline/server/DataPrepperServer.java

@@ -10,6 +10,7 @@
 import com.sun.net.httpserver.HttpHandler;
 import com.sun.net.httpserver.HttpServer;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
+import org.opensearch.dataprepper.plugins.encryption.EncryptionHttpHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -32,6 +33,7 @@ public class DataPrepperServer {
     private final ListPipelinesHandler listPipelinesHandler;
     private final GetPipelinesHandler getPipelinesHandler;
     private final ShutdownHandler shutdownHandler;
+    private final EncryptionHttpHandler encryptionHttpHandler;
     private final PrometheusMeterRegistry prometheusMeterRegistry;
     private final Authenticator authenticator;
     private final ExecutorService executorService;
@@ -43,13 +45,15 @@ public DataPrepperServer(
             final ListPipelinesHandler listPipelinesHandler,
             final ShutdownHandler shutdownHandler,
             final GetPipelinesHandler getPipelinesHandler,
+            @Autowired(required = false) @Nullable final EncryptionHttpHandler encryptionHttpHandler,
             @Autowired(required = false) @Nullable final PrometheusMeterRegistry prometheusMeterRegistry,
             @Autowired(required = false) @Nullable final Authenticator authenticator
     ) {
         this.serverProvider = serverProvider;
         this.listPipelinesHandler = listPipelinesHandler;
         this.shutdownHandler = shutdownHandler;
         this.getPipelinesHandler = getPipelinesHandler;
+        this.encryptionHttpHandler = encryptionHttpHandler;
         this.prometheusMeterRegistry = prometheusMeterRegistry;
         this.authenticator = authenticator;
         executorService = Executors.newFixedThreadPool(3);
@@ -72,6 +76,10 @@ private HttpServer createServer() {
         createContext(server, shutdownHandler, authenticator, "/shutdown");
         createContext(server, getPipelinesHandler, authenticator, "/pipelines");
 
+        if (encryptionHttpHandler != null) {
+            createContext(server, encryptionHttpHandler, authenticator, "/encryption/rotate");
+        }
+
         if (prometheusMeterRegistry != null) {
             final PrometheusMetricsHandler prometheusMetricsHandler = new PrometheusMetricsHandler(prometheusMeterRegistry);
             createContext(server, prometheusMetricsHandler, authenticator, "/metrics/prometheus", "/metrics/sys");

data-prepper-core/src/test/java/org/opensearch/dataprepper/core/pipeline/server/DataPrepperServerTest.java

@@ -16,6 +16,7 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.plugins.encryption.EncryptionHttpHandler;
 
 import java.net.InetSocketAddress;
 import java.util.concurrent.ExecutorService;
@@ -51,6 +52,9 @@ public class DataPrepperServerTest {
     @Mock
     private GetPipelinesHandler getPipelinesHandler;
 
+    @Mock
+    private EncryptionHttpHandler encryptionHttpHandler;
+
     @Mock
     private PrometheusMeterRegistry prometheusMeterRegistry;
 
@@ -71,15 +75,15 @@ public void tearDown() {
 
     @Test
     public void testDataPrepperServerConstructor() {
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, null);
         assertThat(dataPrepperServer, is(notNullValue()));
     }
 
     @Test
     public void testGivenValidServerWhenStartThenShouldCallServerStart() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -88,11 +92,25 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart() {
         verify(context, times(5)).setAuthenticator(eq(authenticator));
     }
 
+    @Test
+    public void testServerStartWithEncryptionHttpHandler() {
+        mockServerStart();
+
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, authenticator, encryptionHttpHandler);
+        dataPrepperServer.start();
+
+        verifyServerStart();
+        verify(server).createContext(eq("/metrics/prometheus"), any(PrometheusMetricsHandler.class));
+        verify(server).createContext(eq("/metrics/sys"), any(PrometheusMetricsHandler.class));
+        verify(server).createContext(eq("/encryption/rotate"), any(EncryptionHttpHandler.class));
+        verify(context, times(6)).setAuthenticator(eq(authenticator));
+    }
+
     @Test
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheus() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, authenticator);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, authenticator, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -103,7 +121,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheu
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullAuthenticator() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(prometheusMeterRegistry, null, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -115,7 +133,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullAuthentic
     public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheusAndAuthenticator() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null, null);
         dataPrepperServer.start();
 
         verifyServerStart();
@@ -126,7 +144,7 @@ public void testGivenValidServerWhenStartThenShouldCallServerStart_NullPrometheu
     public void testGivenValidServerWhenStopThenShouldCallServerStopWithNoDelay() {
         mockServerStart();
 
-        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null);
+        final DataPrepperServer dataPrepperServer = createObjectUnderTest(null, null, null);
         dataPrepperServer.start();
         dataPrepperServer.stop();
 
@@ -161,7 +179,10 @@ private void verifyServerStart() {
         verify(socketAddress).getPort();
     }
 
-    private DataPrepperServer createObjectUnderTest(final PrometheusMeterRegistry prometheusMeterRegistry, final Authenticator authenticator) {
-        return new DataPrepperServer(httpServerProvider, listPipelinesHandler, shutdownHandler, getPipelinesHandler, prometheusMeterRegistry, authenticator);
+    private DataPrepperServer createObjectUnderTest(final PrometheusMeterRegistry prometheusMeterRegistry,
+                                                    final Authenticator authenticator,
+                                                    final EncryptionHttpHandler encryptionHttpHandler) {
+        return new DataPrepperServer(
+                httpServerProvider, listPipelinesHandler, shutdownHandler, getPipelinesHandler, encryptionHttpHandler, prometheusMeterRegistry, authenticator);
     }
 }