GitHub-Issue#2778: Added CouldWatchLogsService, Tests and RetransmissionException

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/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
@@ -31,7 +31,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() {
@@ -38,13 +40,19 @@ public byte[] popEvent() {
     }
 
     @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,7 +10,6 @@
 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;
 
 /**
@@ -20,6 +19,9 @@
  * CloudWatchLogs services.
  */
 public final class CloudWatchLogsClientFactory {
+    private CloudWatchLogsClientFactory() {
+        throw new IllegalStateException("Static Factory Class!");
+    }
 
     /**
      * Generates a CloudWatchLogs Client based on STS role ARN system credentials.
@@ -38,10 +40,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,144 @@
+/*
+ * 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.opensearch.dataprepper.plugins.sink.packaging.ThreadTaskEvents;
+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.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+
+@Builder
+public class CloudWatchLogsDispatcher implements Runnable {
+    private static final long UPPER_RETRY_TIME_BOUND_MILLISECONDS = 5000;
+    private static final float EXP_TIME_SCALE = 1.5F;
+    private static final Logger LOG = LoggerFactory.getLogger(CloudWatchLogsDispatcher.class);
+    private BlockingQueue<ThreadTaskEvents> taskQueue;
+    private CloudWatchLogsClient cloudWatchLogsClient;
+    private CloudWatchLogsMetrics cloudWatchLogsMetrics;
+    private String logGroup;
+    private String logStream;
+    private int retryCount;
+    private long backOffTimeBase;
+    public CloudWatchLogsDispatcher(final BlockingQueue<ThreadTaskEvents> taskQueue,
+                                    final CloudWatchLogsClient cloudWatchLogsClient,
+                                    final CloudWatchLogsMetrics cloudWatchLogsMetrics,
+                                    final String logGroup, final String logStream,
+                                    final int retryCount, final long backOffTimeBase) {
+        this.taskQueue = taskQueue;
+        this.cloudWatchLogsClient = cloudWatchLogsClient;
+        this.cloudWatchLogsMetrics = cloudWatchLogsMetrics;
+        this.logGroup = logGroup;
+        this.logStream = logStream;
+        this.retryCount = retryCount;
+        this.backOffTimeBase = backOffTimeBase;
+    }
+
+    private List<InputLogEvent> prepareInputLogEvents(final ThreadTaskEvents eventData) {
+        List<InputLogEvent> logEventList = new ArrayList<>();
+
+        for (byte[] data: eventData.getEventMessages()) {
+            InputLogEvent tempLogEvent = InputLogEvent.builder()
+                    .message(new String(data))
+                    .timestamp(System.currentTimeMillis())
+                    .build();
+            logEventList.add(tempLogEvent);
+        }
+
+        return logEventList;
+    }
+
+    /**
+     * Flush function to handle the flushing of logs to CloudWatchLogs services;
+     * @param inputLogEvents Collection of inputLogEvents to be flushed
+     * @param eventHandles Collection of EventHandles for events
+     */
+    public void dispatchLogs(List<InputLogEvent> inputLogEvents, Collection<EventHandle> eventHandles) {
+        boolean failedPost = true;
+        int failCounter = 0;
+
+        PutLogEventsRequest putLogEventsRequest = PutLogEventsRequest.builder()
+                .logEvents(inputLogEvents)
+                .logGroupName(logGroup)
+                .logStreamName(logStream)
+                .build();
+
+        try {
+            while (failedPost && (failCounter < retryCount)) {
+                try {
+                    cloudWatchLogsClient.putLogEvents(putLogEventsRequest);
+
+                    cloudWatchLogsMetrics.increaseRequestSuccessCounter(1);
+                    failedPost = false;
+
+                    //TODO: When a log is rejected by the service, we cannot send it, can probably push to a DLQ here.
+
+                } catch (CloudWatchLogsException | SdkClientException e) {
+                    LOG.error("Service-Worker {} Failed to push logs with error: {}", Thread.currentThread().getName(), e.getMessage());
+                    cloudWatchLogsMetrics.increaseRequestFailCounter(1);
+                    Thread.sleep(calculateBackOffTime(backOffTimeBase, failCounter));
+                    LOG.warn("Service-Worker {} Trying to retransmit request... {Attempt: {} }", Thread.currentThread().getName(), (++failCounter));
+                }
+            }
+        } catch (InterruptedException e) {
+            LOG.warn("Got interrupted while waiting!");
+            //TODO: Push to DLQ.
+            Thread.currentThread().interrupt();
+        }
+
+
+        if (failedPost) {
+            cloudWatchLogsMetrics.increaseLogEventFailCounter(inputLogEvents.size());
+            releaseEventHandles(false, eventHandles);
+        } else {
+            cloudWatchLogsMetrics.increaseLogEventSuccessCounter(inputLogEvents.size());
+            releaseEventHandles(true, eventHandles);
+        }
+    }
+
+    //TODO: Can abstract this if clients want more choice.
+    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;
+    }
+
+    @Override
+    public void run() {
+        try {
+            ThreadTaskEvents taskData = taskQueue.take();
+            List<InputLogEvent> inputLogEvents = prepareInputLogEvents(taskData);
+            dispatchLogs(inputLogEvents, taskData.getEventHandles());
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            //TODO: Implement back up to taskQueue read failure.
+        }
+    }
+
+    private void releaseEventHandles(final boolean result, final Collection<EventHandle> eventHandles) {
+        if (eventHandles.isEmpty()) {
+            return;
+        }
+
+        for (EventHandle eventHandle : eventHandles) {
+            eventHandle.release(result);
+        }
+    }
+}

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

@@ -0,0 +1,48 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.client;
+
+import io.micrometer.core.instrument.Counter;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+
+/**
+ * Class is meant to abstract the metric book-keeping of
+ * CloudWatchLogs metrics so that multiple instances
+ * may refer to it.
+ */
+public class CloudWatchLogsMetrics {
+    public static final String CLOUDWATCH_LOGS_REQUESTS_SUCCEEDED = "cloudWatchLogsRequestsSucceeded";
+    public static final String CLOUDWATCH_LOGS_EVENTS_SUCCEEDED = "cloudWatchLogsEventsSucceeded";
+    public static final String CLOUDWATCH_LOGS_EVENTS_FAILED = "cloudWatchLogsEventsFailed";
+    public static final String CLOUDWATCH_LOGS_REQUESTS_FAILED = "cloudWatchLogsRequestsFailed";
+    private final Counter logEventSuccessCounter;
+    private final Counter logEventFailCounter;
+    private final Counter requestSuccessCount;
+    private final Counter requestFailCount;
+
+    public CloudWatchLogsMetrics(final PluginMetrics pluginMetrics) {
+        this.logEventSuccessCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_SUCCEEDED);
+        this.requestFailCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_FAILED);
+        this.logEventFailCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_FAILED);
+        this.requestSuccessCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_SUCCEEDED);
+    }
+
+    public void increaseLogEventSuccessCounter(int value) {
+        logEventSuccessCounter.increment(value);
+    }
+
+    public void increaseRequestSuccessCounter(int value) {
+        requestSuccessCount.increment(value);
+    }
+
+    public void increaseLogEventFailCounter(int value) {
+        logEventFailCounter.increment(value);
+    }
+
+    public void increaseRequestFailCounter(int value) {
+        requestFailCount.increment(value);
+    }
+}