Added distribution summary metrics for log size and request size

Signed-off-by: Kondaka <krishkdk@amazon.com>

Author: kkondaka

data-prepper-plugins/cloudwatch-logs/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsIT.java

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.plugins.sink.cloudwatch_logs;
 
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -98,6 +99,9 @@ public class CloudWatchLogsIT {
     @Mock
     private CloudWatchLogsSinkConfig cloudWatchLogsSinkConfig;
 
+    @Mock
+    private DistributionSummary summary;
+
     @Mock
     private Counter eventsSuccessCounter;
     @Mock
@@ -156,6 +160,7 @@ void setUp() {
         logGroupName = System.getProperty("tests.cloudwatch.log_group");
         logStreamName = createLogStream(logGroupName);
         pluginMetrics = mock(PluginMetrics.class);
+        summary = mock(DistributionSummary.class);
         eventsSuccessCounter = mock(Counter.class);
         requestsSuccessCounter = mock(Counter.class);
         eventsFailedCounter = mock(Counter.class);
@@ -218,6 +223,7 @@ void setUp() {
             }
             return null;
         }).when(pluginMetrics).counter(anyString());
+        when(pluginMetrics.summary(anyString())).thenReturn(summary);
         cloudWatchLogsSinkConfig = mock(CloudWatchLogsSinkConfig.class);
         when(cloudWatchLogsSinkConfig.getLogGroup()).thenReturn(logGroupName);
         when(cloudWatchLogsSinkConfig.getDlq()).thenReturn(null);

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

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.plugins.sink.cloudwatch_logs.client;
 
 import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.DistributionSummary;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 
 /**
@@ -19,18 +20,24 @@ public class CloudWatchLogsMetrics {
     public static final String CLOUDWATCH_LOGS_EVENTS_FAILED = "cloudWatchLogsEventsFailed";
     public static final String CLOUDWATCH_LOGS_REQUESTS_FAILED = "cloudWatchLogsRequestsFailed";
     public static final String CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED = "cloudWatchLogsLargeEventsDropped";
+    public static final String CLOUDWATCH_LOGS_LOG_SIZE = "cloudWatchLogsLogSize";
+    public static final String CLOUDWATCH_LOGS_REQUEST_SIZE = "cloudWatchLogsRequestSize";
     private final Counter logEventSuccessCounter;
     private final Counter logEventFailCounter;
     private final Counter requestSuccessCount;
     private final Counter requestFailCount;
     private final Counter logLargeEventsDroppedCounter;
+    private final DistributionSummary logSizeMetric;
+    private final DistributionSummary requestSizeMetric;
 
     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);
         this.logLargeEventsDroppedCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED);
+        this.logSizeMetric = pluginMetrics.summary(CLOUDWATCH_LOGS_LOG_SIZE);
+        this.requestSizeMetric = pluginMetrics.summary(CLOUDWATCH_LOGS_REQUEST_SIZE);
     }
 
     public void increaseLogEventSuccessCounter(int value) {
@@ -52,4 +59,12 @@ public void increaseRequestFailCounter(int value) {
     public void increaseLogLargeEventsDroppedCounter(int value) {
         logLargeEventsDroppedCounter.increment(value);
     }
+
+    public void recordLogSize(int value) {
+        logSizeMetric.record(value);
+    }
+
+    public void recordRequestSize(int value) {
+        requestSizeMetric.record(value);
+    }
 }

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

@@ -83,6 +83,7 @@ public void processLogEvents(final Collection<Record<Event>> logs) {
             String logString = log.getData().toJsonString();
             int logLength = logString.length();
 
+            cloudWatchLogsMetrics.recordLogSize(logLength);
             if (cloudWatchLogsLimits.isGreaterThanMaxEventSize(logLength)) {
                 final String failureMessage = String.format("Event blocked due to Max Size restriction! Event Size : %s", (logLength + CloudWatchLogsLimits.APPROXIMATE_LOG_EVENT_OVERHEAD_SIZE));
                 DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, log.getData().getEventHandle(), logString, failureMessage, dlqPushHandler, dropIfDlqNotConfigured);
@@ -121,6 +122,7 @@ private void stageLogEvents() {
 
         List<InputLogEvent> inputLogEvents = cloudWatchLogsDispatcher.prepareInputLogEvents(buffer.getBufferedData());
         cloudWatchLogsDispatcher.dispatchLogs(inputLogEvents, buffer.getEventHandles());
+        cloudWatchLogsMetrics.recordRequestSize(buffer.getBufferSize());
 
         buffer.resetBuffer();
     }

data-prepper-plugins/cloudwatch-logs/src/test/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/CloudWatchLogsSinkTest.java

@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper.plugins.sink.cloudwatch_logs;
 
+import io.micrometer.core.instrument.DistributionSummary;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.MockedStatic;
@@ -30,6 +31,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.atLeast;
 import static org.mockito.Mockito.mock;
@@ -68,6 +70,8 @@ void setUp() {
         mockHeaderOverrides.put("X-Test-Header", "test-value");
         mockCloudWatchLogsMetrics = mock(CloudWatchLogsMetrics.class);
         mockClient = mock(CloudWatchLogsClient.class);
+        DistributionSummary summary = mock(DistributionSummary.class);
+        when(mockPluginMetrics.summary(anyString())).thenReturn(summary);
 
         when(mockCloudWatchLogsSinkConfig.getAwsConfig()).thenReturn(mockAwsConfig);
         when(mockCloudWatchLogsSinkConfig.getThresholdConfig()).thenReturn(thresholdConfig);
@@ -213,4 +217,4 @@ void WHEN_sink_initialization_with_header_overrides_THEN_sink_is_ready() {
             assertTrue(testCloudWatchSink.isReady());
         }
     }
-}
+}