GitHub-Issue#2778: Added CloudWatchLogsSink

data-prepper-plugins/cloudwatch-logs/build.gradle

@@ -17,9 +17,12 @@ dependencies {
     implementation 'software.amazon.awssdk:cloudwatch'
     implementation 'software.amazon.awssdk:cloudwatchlogs'
     implementation 'org.apache.commons:commons-lang3:3.12.0'
+    implementation 'org.projectlombok:lombok:1.18.26'
     testImplementation project(path: ':data-prepper-test-common')
     testImplementation testLibs.mockito.inline
     testImplementation 'org.junit.jupiter:junit-jupiter'
+    compileOnly 'org.projectlombok:lombok:1.18.24'
+    annotationProcessor 'org.projectlombok:lombok:1.18.24'
 }
 
 jacocoTestCoverageVerification {

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/CloudWatchLogsSink.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink;
+
+import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.configuration.PluginSetting;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+import org.opensearch.dataprepper.model.sink.AbstractSink;
+import org.opensearch.dataprepper.model.sink.Sink;
+import org.opensearch.dataprepper.plugins.sink.buffer.Buffer;
+import org.opensearch.dataprepper.plugins.sink.buffer.BufferFactory;
+import org.opensearch.dataprepper.plugins.sink.buffer.InMemoryBufferFactory;
+import org.opensearch.dataprepper.plugins.sink.client.CloudWatchLogsDispatcher;
+import org.opensearch.dataprepper.plugins.sink.client.CloudWatchLogsMetrics;
+import org.opensearch.dataprepper.plugins.sink.client.CloudWatchLogsService;
+import org.opensearch.dataprepper.plugins.sink.client.CloudWatchLogsClientFactory;
+import org.opensearch.dataprepper.plugins.sink.config.AwsConfig;
+import org.opensearch.dataprepper.plugins.sink.config.CloudWatchLogsSinkConfig;
+import org.opensearch.dataprepper.plugins.sink.config.ThresholdConfig;
+import org.opensearch.dataprepper.plugins.sink.utils.CloudWatchLogsLimits;
+import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
+
+import java.util.Collection;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+@DataPrepperPlugin(name = "cloudwatchlogs-sink", pluginType = Sink.class, pluginConfigurationType = CloudWatchLogsSinkConfig.class)
+public class CloudWatchLogsSink extends AbstractSink<Record<Event>> {
+    private final CloudWatchLogsService cloudWatchLogsService;
+    private boolean isInitialized;
+    @DataPrepperPluginConstructor
+    public CloudWatchLogsSink(final PluginSetting pluginSetting,
+                              final PluginMetrics pluginMetrics,
+                              final CloudWatchLogsSinkConfig cloudWatchLogsSinkConfig,
+                              final AwsCredentialsSupplier awsCredentialsSupplier) {
+        super(pluginSetting);
+
+        AwsConfig awsConfig = cloudWatchLogsSinkConfig.getAwsConfig();
+        ThresholdConfig thresholdConfig = cloudWatchLogsSinkConfig.getThresholdConfig();
+
+        CloudWatchLogsMetrics cloudWatchLogsMetrics = new CloudWatchLogsMetrics(pluginMetrics);
+        CloudWatchLogsLimits cloudWatchLogsLimits = new CloudWatchLogsLimits(thresholdConfig.getBatchSize(),
+                thresholdConfig.getMaxEventSizeBytes(),
+                thresholdConfig.getMaxRequestSize(),thresholdConfig.getLogSendInterval());
+
+        CloudWatchLogsClient cloudWatchLogsClient = CloudWatchLogsClientFactory.createCwlClient(awsConfig, awsCredentialsSupplier);
+        BufferFactory bufferFactory = new InMemoryBufferFactory();
+
+        Executor executor = Executors.newCachedThreadPool();
+
+        CloudWatchLogsDispatcher cloudWatchLogsDispatcher = CloudWatchLogsDispatcher.builder()
+                .cloudWatchLogsClient(cloudWatchLogsClient)
+                .cloudWatchLogsMetrics(cloudWatchLogsMetrics)
+                .logGroup(cloudWatchLogsSinkConfig.getLogGroup())
+                .logStream(cloudWatchLogsSinkConfig.getLogStream())
+                .backOffTimeBase(thresholdConfig.getBackOffTime())
+                .retryCount(thresholdConfig.getRetryCount())
+                .executor(executor)
+                .build();
+
+        Buffer buffer = bufferFactory.getBuffer();
+
+        cloudWatchLogsService = new CloudWatchLogsService(buffer, cloudWatchLogsLimits, cloudWatchLogsDispatcher);
+    }
+
+    @Override
+    public void doInitialize() {
+        isInitialized = true;
+    }
+
+    @Override
+    public void doOutput(Collection<Record<Event>> records) {
+        if (records.isEmpty()) {
+            return;
+        }
+
+        cloudWatchLogsService.processLogEvents(records);
+    }
+
+    @Override
+    public boolean isReady() {
+        return isInitialized;
+    }
+}

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/buffer/Buffer.java

@@ -5,7 +5,7 @@
 
 package org.opensearch.dataprepper.plugins.sink.buffer;
 
-import java.util.ArrayList;
+import java.util.List;
 
 /**
  * Buffer that handles the temporary storage of
@@ -14,6 +14,13 @@
  * 2. Transforms to Byte type.
  * 3. Returns a Byte type.
  */
+
+/*
+    TODO:
+     Need to add PriorityQueue for extracting timestamp, this will need the timestamp and the actual string message itself.
+     Can refactor the buffer to contain
+ */
+
 public interface Buffer {
     /**
      * Size of buffer in events.
@@ -31,7 +38,9 @@ public interface Buffer {
 
     byte[] popEvent();
 
-    ArrayList<byte[]> getBufferedData();
+    List<byte[]> getBufferedData();
 
     void clearBuffer();
+
+    void resetBuffer();
 }

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/buffer/InMemoryBuffer.java

@@ -6,9 +6,11 @@
 package org.opensearch.dataprepper.plugins.sink.buffer;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 public class InMemoryBuffer implements Buffer {
-    private final ArrayList<byte[]> eventsBuffered;
+    private List<byte[]> eventsBuffered;
     private int bufferSize = 0;
 
     InMemoryBuffer() {
@@ -33,18 +35,27 @@ public void writeEvent(final byte[] event) {
 
     @Override
     public byte[] popEvent() {
+        if (eventsBuffered.isEmpty()) {
+            return new byte[0];
+        }
         bufferSize -= eventsBuffered.get(0).length;
         return eventsBuffered.remove(0);
     }
 
     @Override
-    public ArrayList<byte[]> getBufferedData() {
-        return eventsBuffered;
+    public List<byte[]> getBufferedData() {
+        return Collections.unmodifiableList(eventsBuffered);
     }
 
     @Override
     public void clearBuffer() {
         bufferSize = 0;
         eventsBuffered.clear();
     }
+
+    @Override
+    public void resetBuffer() {
+        bufferSize = 0;
+        eventsBuffered = new ArrayList<>();
+    }
 }

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/client/CloudWatchLogsClientFactory.java

@@ -10,22 +10,21 @@
 import org.opensearch.dataprepper.plugins.sink.config.AwsConfig;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
-import software.amazon.awssdk.core.retry.RetryPolicy;
 import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
 
 /**
- * CwlClientFactory is in charge of reading in
- * aws config parameters to return a working
- * client for interfacing with
- * CloudWatchLogs services.
+ * CwlClientFactory is in charge of reading in aws config parameters to return a working
+ * client for interfacing with CloudWatchLogs services.
  */
 public final class CloudWatchLogsClientFactory {
+    private CloudWatchLogsClientFactory() {
+    }
 
     /**
      * Generates a CloudWatchLogs Client based on STS role ARN system credentials.
-     * @param awsConfig - AwsConfig specifying region, roles, and header overrides.
-     * @param awsCredentialsSupplier - AwsCredentialsSupplier Interface for which to create CredentialsProvider for Client config.
-     * @return CloudWatchLogsClient - used to interact with CloudWatch Logs services.
+     * @param awsConfig AwsConfig specifying region, roles, and header overrides.
+     * @param awsCredentialsSupplier AwsCredentialsSupplier Interface for which to create CredentialsProvider for Client config.
+     * @return CloudWatchLogsClient used to interact with CloudWatch Logs services.
      */
     public static CloudWatchLogsClient createCwlClient(final AwsConfig awsConfig, final AwsCredentialsSupplier awsCredentialsSupplier) {
         final AwsCredentialsOptions awsCredentialsOptions = convertToCredentialOptions(awsConfig);
@@ -38,10 +37,8 @@ public static CloudWatchLogsClient createCwlClient(final AwsConfig awsConfig, fi
     }
 
     private static ClientOverrideConfiguration createOverrideConfiguration() {
-        final RetryPolicy retryPolicy = RetryPolicy.builder().numRetries(AwsConfig.DEFAULT_CONNECTION_ATTEMPTS).build();
-
         return ClientOverrideConfiguration.builder()
-                .retryPolicy(retryPolicy)
+                .retryPolicy(r -> r.numRetries(AwsConfig.DEFAULT_CONNECTION_ATTEMPTS))
                 .build();
     }
 

data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/client/CloudWatchLogsDispatcher.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.client;
+
+import lombok.Builder;
+import org.opensearch.dataprepper.model.event.EventHandle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CloudWatchLogsException;
+import software.amazon.awssdk.services.cloudwatchlogs.model.InputLogEvent;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutLogEventsRequest;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+@Builder
+public class CloudWatchLogsDispatcher {
+    private static final long UPPER_RETRY_TIME_BOUND_MILLISECONDS = 2000;
+    private static final float EXP_TIME_SCALE = 1.25F;
+    private static final Logger LOG = LoggerFactory.getLogger(CloudWatchLogsDispatcher.class);
+    private CloudWatchLogsClient cloudWatchLogsClient;
+    private CloudWatchLogsMetrics cloudWatchLogsMetrics;
+    private Executor executor;
+    private String logGroup;
+    private String logStream;
+    private int retryCount;
+    private long backOffTimeBase;
+    public CloudWatchLogsDispatcher(final CloudWatchLogsClient cloudWatchLogsClient,
+                                    final CloudWatchLogsMetrics cloudWatchLogsMetrics,
+                                    final Executor executor,
+                                    final String logGroup, final String logStream,
+                                    final int retryCount, final long backOffTimeBase) {
+        this.cloudWatchLogsClient = cloudWatchLogsClient;
+        this.cloudWatchLogsMetrics = cloudWatchLogsMetrics;
+        this.logGroup = logGroup;
+        this.logStream = logStream;
+        this.retryCount = retryCount;
+        this.backOffTimeBase = backOffTimeBase;
+
+        this.executor = executor;
+    }
+
+    /**
+     * Will read in a collection of log messages in byte form and transform them into a collection of InputLogEvents.
+     * @param eventMessageBytes Collection of byte arrays holding event messages.
+     * @return List of InputLogEvents holding the wrapped event messages.
+     */
+    public List<InputLogEvent> prepareInputLogEvents(final Collection<byte[]> eventMessageBytes) {
+        List<InputLogEvent> logEventList = new ArrayList<>();
+
+        /**
+         * In the current implementation, the timestamp is generated during transmission.
+         * To properly extract timestamp we need to order the InputLogEvents. Can be done by
+         * refactoring buffer class with timestamp param, or adding a sorting algorithm in between
+         * making the PLE object (in prepareInputLogEvents).
+         */
+
+        for (byte[] data : eventMessageBytes) {
+            InputLogEvent tempLogEvent = InputLogEvent.builder()
+                    .message(new String(data, StandardCharsets.UTF_8))
+                    .timestamp(System.currentTimeMillis())
+                    .build();
+            logEventList.add(tempLogEvent);
+        }
+
+        return logEventList;
+    }
+
+    public void dispatchLogs(List<InputLogEvent> inputLogEvents, Collection<EventHandle> eventHandles) {
+        PutLogEventsRequest putLogEventsRequest = PutLogEventsRequest.builder()
+                .logEvents(inputLogEvents)
+                .logGroupName(logGroup)
+                .logStreamName(logStream)
+                .build();
+
+        executor.execute(Uploader.builder()
+                .cloudWatchLogsClient(cloudWatchLogsClient)
+                .cloudWatchLogsMetrics(cloudWatchLogsMetrics)
+                .putLogEventsRequest(putLogEventsRequest)
+                .eventHandles(eventHandles)
+                .backOffTimeBase(backOffTimeBase)
+                .retryCount(retryCount)
+                .build());
+    }
+
+    @Builder
+    protected static class Uploader implements Runnable {
+        private final CloudWatchLogsClient cloudWatchLogsClient;
+        private final CloudWatchLogsMetrics cloudWatchLogsMetrics;
+        private final PutLogEventsRequest putLogEventsRequest;
+        private final Collection<EventHandle> eventHandles;
+        private final int retryCount;
+        private final long backOffTimeBase;
+
+        @Override
+        public void run() {
+            upload();
+        }
+
+        public void upload() {
+            boolean failedToTransmit = true;
+            int failCount = 0;
+
+            try {
+                while (failedToTransmit && (failCount < retryCount)) {
+                    try {
+                        cloudWatchLogsClient.putLogEvents(putLogEventsRequest);
+
+                        cloudWatchLogsMetrics.increaseRequestSuccessCounter(1);
+                        failedToTransmit = false;
+
+                    } catch (CloudWatchLogsException | SdkClientException e) {
+                        LOG.error("Failed to push logs with error: {}", e.getMessage());
+                        cloudWatchLogsMetrics.increaseRequestFailCounter(1);
+                        Thread.sleep(calculateBackOffTime(backOffTimeBase, failCount));
+                        failCount++;
+                    }
+                }
+            } catch (InterruptedException e) {
+                LOG.warn("Uploader Thread got interrupted during retransmission with exception: {}", e.getMessage());
+                //TODO: Push to DLQ.
+                Thread.currentThread().interrupt();
+            }
+
+
+            if (failedToTransmit) {
+                cloudWatchLogsMetrics.increaseLogEventFailCounter(eventHandles.size());
+                releaseEventHandles(false, eventHandles);
+            } else {
+                cloudWatchLogsMetrics.increaseLogEventSuccessCounter(eventHandles.size());
+                releaseEventHandles(true, eventHandles);
+            }
+        }
+
+        private long calculateBackOffTime(final long backOffTimeBase, final int failCounter) {
+            long scale = (long)Math.pow(EXP_TIME_SCALE, failCounter);
+
+            if (scale >= UPPER_RETRY_TIME_BOUND_MILLISECONDS) {
+                return UPPER_RETRY_TIME_BOUND_MILLISECONDS;
+            }
+
+            return scale * backOffTimeBase;
+        }
+
+        private void releaseEventHandles(final boolean result, final Collection<EventHandle> eventHandles) {
+            if (eventHandles.isEmpty()) {
+                return;
+            }
+
+            for (EventHandle eventHandle : eventHandles) {
+                eventHandle.release(result);
+            }
+        }
+    }
+}