Skip to content

Commit 5ad4885

Browse files
kkondakakkondaka
authored
CloudWatchLogs Sink: Update max event size and drop error events if DLQ is not configured (#6154)
* CloudWatchLogs Sink: Update max event size and drop error events if DLQ is not configured Signed-off-by: Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Kondaka <krishkdk@amazon.com> * Added distribution summary metrics for log size and request size Signed-off-by: Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Kondaka <krishkdk@amazon.com>
1 parent 2b37573 commit 5ad4885

12 files changed

Lines changed: 170 additions & 35 deletions

File tree

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

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package org.opensearch.dataprepper.plugins.sink.cloudwatch_logs;
77

88
import io.micrometer.core.instrument.Counter;
9+
import io.micrometer.core.instrument.DistributionSummary;
910
import com.fasterxml.jackson.databind.ObjectMapper;
1011
import org.apache.commons.lang3.RandomStringUtils;
1112
import org.junit.jupiter.api.extension.ExtendWith;
@@ -57,6 +58,7 @@
5758
import static org.mockito.Mockito.when;
5859
import static org.mockito.Mockito.lenient;
5960
import static org.hamcrest.CoreMatchers.equalTo;
61+
import static org.hamcrest.CoreMatchers.notNullValue;
6062
import static org.hamcrest.MatcherAssert.assertThat;
6163

6264
import java.time.Duration;
@@ -97,13 +99,18 @@ public class CloudWatchLogsIT {
9799
@Mock
98100
private CloudWatchLogsSinkConfig cloudWatchLogsSinkConfig;
99101

102+
@Mock
103+
private DistributionSummary summary;
104+
100105
@Mock
101106
private Counter eventsSuccessCounter;
102107
@Mock
103108
private Counter requestsSuccessCounter;
104109
@Mock
105110
private Counter eventsFailedCounter;
106111
@Mock
112+
private Counter largeEventsDroppedCounter;
113+
@Mock
107114
private Counter requestsFailedCounter;
108115
@Mock
109116
private Counter dlqSuccessCounter;
@@ -119,6 +126,7 @@ public class CloudWatchLogsIT {
119126
private AtomicInteger eventsFailedCount;
120127
private AtomicInteger requestsFailedCount;
121128
private AtomicInteger dlqSuccessCount;
129+
private AtomicInteger largeEventsDroppedCount;
122130
private CloudWatchLogsClient cloudWatchLogsClient;
123131
private ObjectMapper objectMapper;
124132
private AwsCredentialsProvider awsCredentialsProvider;
@@ -130,6 +138,7 @@ void setUp() {
130138
eventsSuccessCount = new AtomicInteger(0);
131139
requestsSuccessCount = new AtomicInteger(0);
132140
eventsFailedCount = new AtomicInteger(0);
141+
largeEventsDroppedCount = new AtomicInteger(0);
133142
requestsFailedCount = new AtomicInteger(0);
134143
dlqSuccessCount = new AtomicInteger(0);
135144
objectMapper = new ObjectMapper();
@@ -151,9 +160,11 @@ void setUp() {
151160
logGroupName = System.getProperty("tests.cloudwatch.log_group");
152161
logStreamName = createLogStream(logGroupName);
153162
pluginMetrics = mock(PluginMetrics.class);
163+
summary = mock(DistributionSummary.class);
154164
eventsSuccessCounter = mock(Counter.class);
155165
requestsSuccessCounter = mock(Counter.class);
156166
eventsFailedCounter = mock(Counter.class);
167+
largeEventsDroppedCounter = mock(Counter.class);
157168
requestsFailedCounter = mock(Counter.class);
158169
dlqSuccessCounter = mock(Counter.class);
159170
lenient().doAnswer((a)-> {
@@ -166,6 +177,11 @@ void setUp() {
166177
eventsFailedCount.addAndGet(v);
167178
return null;
168179
}).when(eventsFailedCounter).increment(any(Double.class));
180+
lenient().doAnswer((a)-> {
181+
int v = (int)(double)(a.getArgument(0));
182+
largeEventsDroppedCount.addAndGet(v);
183+
return null;
184+
}).when(largeEventsDroppedCounter).increment(any(Double.class));
169185
lenient().doAnswer((a)-> {
170186
requestsSuccessCount.addAndGet(1);
171187
return null;
@@ -199,11 +215,15 @@ void setUp() {
199215
if (s.equals(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_FAILED)) {
200216
return eventsFailedCounter;
201217
}
218+
if (s.equals(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED)) {
219+
return largeEventsDroppedCounter;
220+
}
202221
if (s.contains("NumDlqSuccess")) {
203222
return dlqSuccessCounter;
204223
}
205224
return null;
206225
}).when(pluginMetrics).counter(anyString());
226+
when(pluginMetrics.summary(anyString())).thenReturn(summary);
207227
cloudWatchLogsSinkConfig = mock(CloudWatchLogsSinkConfig.class);
208228
when(cloudWatchLogsSinkConfig.getLogGroup()).thenReturn(logGroupName);
209229
when(cloudWatchLogsSinkConfig.getDlq()).thenReturn(null);
@@ -437,6 +457,52 @@ void testWithLargeSingleMessagesSentToDLQ() {
437457

438458
}
439459

460+
@Test
461+
void testWithLargeSingleMessagesWhenDLQNotConfigured() throws Exception {
462+
long startTime = Instant.now().toEpochMilli();
463+
when(thresholdConfig.getBatchSize()).thenReturn(NUM_RECORDS);
464+
when(thresholdConfig.getMaxEventSizeBytes()).thenReturn(200L);
465+
when(thresholdConfig.getMaxRequestSizeBytes()).thenReturn(1000L);
466+
when(cloudWatchLogsSinkConfig.getDlq()).thenReturn(null);
467+
468+
sink = createObjectUnderTest();
469+
Collection<Record<Event>> records = getRecordList(NUM_RECORDS);
470+
Record<Event> largeRecord = getLargeRecord(200);
471+
records.add(largeRecord);
472+
473+
sink.doOutput(records);
474+
List<OutputLogEvent>[] foundEvents = new List[1];
475+
await().atMost(Duration.ofSeconds(30))
476+
.untilAsserted(() -> {
477+
long endTime = Instant.now().toEpochMilli();
478+
GetLogEventsRequest getRequest = GetLogEventsRequest
479+
.builder()
480+
.logGroupName(logGroupName)
481+
.logStreamName(logStreamName)
482+
.startTime(startTime)
483+
.endTime(endTime)
484+
.build();
485+
GetLogEventsResponse response = cloudWatchLogsClient.getLogEvents(getRequest);
486+
List<OutputLogEvent> events = response.events();
487+
foundEvents[0] = events;
488+
assertThat(events.size(), equalTo(NUM_RECORDS));
489+
});
490+
List<OutputLogEvent> events = foundEvents[0];
491+
assertThat(events, notNullValue());
492+
for (int i = 0; i < events.size(); i++) {
493+
String message = events.get(i).message();
494+
Map<String, Object> event = objectMapper.readValue(message, Map.class);
495+
assertThat(event.get("name"), equalTo("Person"+i));
496+
assertThat(event.get("age"), equalTo(Integer.toString(i)));
497+
}
498+
assertThat(eventsSuccessCount.get(), equalTo(NUM_RECORDS));
499+
assertThat(requestsSuccessCount.get(), equalTo(1));
500+
assertThat(largeEventsDroppedCount.get(), equalTo(1));
501+
assertThat(dlqSuccessCount.get(), equalTo(0));
502+
verify(eventHandle, times(NUM_RECORDS+1)).release(true);
503+
504+
}
505+
440506
@Test
441507
void testWithBadCredentials_AllEventsToDLQ() {
442508
s3Client = S3Client.builder()

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,7 @@ public CloudWatchLogsSink(final PluginSetting pluginSetting,
9191
.cloudWatchLogsClient(cloudWatchLogsClient)
9292
.cloudWatchLogsMetrics(cloudWatchLogsMetrics)
9393
.dlqPushHandler(dlqPushHandler)
94+
.dropIfDlqNotConfigured(true)
9495
.logGroup(cloudWatchLogsSinkConfig.getLogGroup())
9596
.logStream(cloudWatchLogsSinkConfig.getLogStream())
9697
.retryCount(cloudWatchLogsSinkConfig.getMaxRetries())
@@ -104,7 +105,7 @@ public CloudWatchLogsSink(final PluginSetting pluginSetting,
104105
throw new InvalidBufferTypeException("Error loading buffer!");
105106
}
106107

107-
cloudWatchLogsService = new CloudWatchLogsService(buffer, cloudWatchLogsLimits, cloudWatchLogsDispatcher, dlqPushHandler);
108+
cloudWatchLogsService = new CloudWatchLogsService(buffer, cloudWatchLogsMetrics, cloudWatchLogsLimits, cloudWatchLogsDispatcher, dlqPushHandler, true);
108109
}
109110

110111
@Override

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

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -35,13 +35,15 @@ public class CloudWatchLogsDispatcher {
3535
private CloudWatchLogsClient cloudWatchLogsClient;
3636
private CloudWatchLogsMetrics cloudWatchLogsMetrics;
3737
private DlqPushHandler dlqPushHandler;
38+
private boolean dropIfDlqNotConfigured;
3839
private Executor executor;
3940
private String logGroup;
4041
private String logStream;
4142
private int retryCount;
4243
public CloudWatchLogsDispatcher(final CloudWatchLogsClient cloudWatchLogsClient,
4344
final CloudWatchLogsMetrics cloudWatchLogsMetrics,
4445
final DlqPushHandler dlqPushHandler,
46+
final boolean dropIfDlqNotConfigured,
4547
final Executor executor,
4648
final String logGroup,
4749
final String logStream,
@@ -52,6 +54,7 @@ public CloudWatchLogsDispatcher(final CloudWatchLogsClient cloudWatchLogsClient,
5254
this.logStream = logStream;
5355
this.retryCount = retryCount;
5456
this.dlqPushHandler = dlqPushHandler;
57+
this.dropIfDlqNotConfigured = dropIfDlqNotConfigured;
5558

5659
this.executor = executor;
5760
}
@@ -95,6 +98,7 @@ public void dispatchLogs(List<InputLogEvent> inputLogEvents, List<EventHandle> e
9598
.dlqPushHandler(dlqPushHandler)
9699
.putLogEventsRequest(putLogEventsRequest)
97100
.eventHandles(eventHandles)
101+
.dropIfDlqNotConfigured(dropIfDlqNotConfigured)
98102
.totalEventCount(inputLogEvents.size())
99103
.retryCount(retryCount)
100104
.build());
@@ -111,6 +115,7 @@ protected static class Uploader implements Runnable {
111115
private final List<EventHandle> eventHandles;
112116
private final int totalEventCount;
113117
private final int retryCount;
118+
private boolean dropIfDlqNotConfigured;
114119

115120
@Override
116121
public void run() {
@@ -154,7 +159,7 @@ public void upload() {
154159
cloudWatchLogsMetrics.increaseLogEventFailCounter(totalEventCount);
155160
List<InputLogEvent> logEvents = putLogEventsRequest.logEvents();
156161
for (int i = 0; i < logEvents.size(); i++) {
157-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvents.get(i).message(), failureMessage, dlqPushHandler);
162+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvents.get(i).message(), failureMessage, dlqPushHandler, dropIfDlqNotConfigured);
158163
if (dlqObject != null) {
159164
dlqObjects.add(dlqObject);
160165
}
@@ -179,7 +184,7 @@ List<DlqObject> getDlqObjectsFromResponse(PutLogEventsResponse putLogEventsRespo
179184
if (endIndex != null) {
180185
int i = 0;
181186
for (InputLogEvent logEvent : logEvents.subList(0, endIndex)) {
182-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvent.message(), "Too old log event", dlqPushHandler);
187+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(i), logEvent.message(), "Too old log event", dlqPushHandler, dropIfDlqNotConfigured);
183188
if (dlqObject != null) {
184189
dlqObjects.add(dlqObject);
185190
}
@@ -190,7 +195,7 @@ List<DlqObject> getDlqObjectsFromResponse(PutLogEventsResponse putLogEventsRespo
190195
if (startIndex != null) {
191196
int i = 0;
192197
for (InputLogEvent logEvent : logEvents.subList(startIndex, logEvents.size())) {
193-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(startIndex + i), logEvent.message(), "Too old log event", dlqPushHandler);
198+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, eventHandles.get(startIndex + i), logEvent.message(), "Too old log event", dlqPushHandler, dropIfDlqNotConfigured);
194199
if (dlqObject != null) {
195200
dlqObjects.add(dlqObject);
196201
}

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

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
package org.opensearch.dataprepper.plugins.sink.cloudwatch_logs.client;
77

88
import io.micrometer.core.instrument.Counter;
9+
import io.micrometer.core.instrument.DistributionSummary;
910
import org.opensearch.dataprepper.metrics.PluginMetrics;
1011

1112
/**
@@ -18,16 +19,25 @@ public class CloudWatchLogsMetrics {
1819
public static final String CLOUDWATCH_LOGS_EVENTS_SUCCEEDED = "cloudWatchLogsEventsSucceeded";
1920
public static final String CLOUDWATCH_LOGS_EVENTS_FAILED = "cloudWatchLogsEventsFailed";
2021
public static final String CLOUDWATCH_LOGS_REQUESTS_FAILED = "cloudWatchLogsRequestsFailed";
22+
public static final String CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED = "cloudWatchLogsLargeEventsDropped";
23+
public static final String CLOUDWATCH_LOGS_LOG_SIZE = "cloudWatchLogsLogSize";
24+
public static final String CLOUDWATCH_LOGS_REQUEST_SIZE = "cloudWatchLogsRequestSize";
2125
private final Counter logEventSuccessCounter;
2226
private final Counter logEventFailCounter;
2327
private final Counter requestSuccessCount;
2428
private final Counter requestFailCount;
29+
private final Counter logLargeEventsDroppedCounter;
30+
private final DistributionSummary logSizeMetric;
31+
private final DistributionSummary requestSizeMetric;
2532

2633
public CloudWatchLogsMetrics(final PluginMetrics pluginMetrics) {
2734
this.logEventSuccessCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_SUCCEEDED);
2835
this.requestFailCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_FAILED);
2936
this.logEventFailCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_EVENTS_FAILED);
3037
this.requestSuccessCount = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_REQUESTS_SUCCEEDED);
38+
this.logLargeEventsDroppedCounter = pluginMetrics.counter(CloudWatchLogsMetrics.CLOUDWATCH_LOGS_LARGE_EVENTS_DROPPED);
39+
this.logSizeMetric = pluginMetrics.summary(CLOUDWATCH_LOGS_LOG_SIZE);
40+
this.requestSizeMetric = pluginMetrics.summary(CLOUDWATCH_LOGS_REQUEST_SIZE);
3141
}
3242

3343
public void increaseLogEventSuccessCounter(int value) {
@@ -45,4 +55,16 @@ public void increaseLogEventFailCounter(int value) {
4555
public void increaseRequestFailCounter(int value) {
4656
requestFailCount.increment(value);
4757
}
58+
59+
public void increaseLogLargeEventsDroppedCounter(int value) {
60+
logLargeEventsDroppedCounter.increment(value);
61+
}
62+
63+
public void recordLogSize(int value) {
64+
logSizeMetric.record(value);
65+
}
66+
67+
public void recordRequestSize(int value) {
68+
requestSizeMetric.record(value);
69+
}
4870
}

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

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -36,22 +36,28 @@ public class CloudWatchLogsService {
3636
private final CloudWatchLogsDispatcher cloudWatchLogsDispatcher;
3737
private final Buffer buffer;
3838
private final CloudWatchLogsLimits cloudWatchLogsLimits;
39+
private CloudWatchLogsMetrics cloudWatchLogsMetrics;
3940
private final SinkStopWatch sinkStopWatch;
4041
private final ReentrantLock processLock;
4142
private final DlqPushHandler dlqPushHandler;
43+
private final boolean dropIfDlqNotConfigured;
4244
public CloudWatchLogsService(final Buffer buffer,
45+
final CloudWatchLogsMetrics cloudWatchLogsMetrics,
4346
final CloudWatchLogsLimits cloudWatchLogsLimits,
4447
final CloudWatchLogsDispatcher cloudWatchLogsDispatcher,
45-
final DlqPushHandler dlqPushHandler) {
48+
final DlqPushHandler dlqPushHandler,
49+
final boolean dropIfDlqNotConfigured) {
4650

4751
this.buffer = buffer;
4852
this.cloudWatchLogsLimits = cloudWatchLogsLimits;
4953

54+
this.cloudWatchLogsMetrics = cloudWatchLogsMetrics;
5055
processLock = new ReentrantLock();
5156
sinkStopWatch = new SinkStopWatch();
5257

5358
this.cloudWatchLogsDispatcher = cloudWatchLogsDispatcher;
5459
this.dlqPushHandler = dlqPushHandler;
60+
this.dropIfDlqNotConfigured = dropIfDlqNotConfigured;
5561
}
5662

5763
/**
@@ -77,11 +83,14 @@ public void processLogEvents(final Collection<Record<Event>> logs) {
7783
String logString = log.getData().toJsonString();
7884
int logLength = logString.length();
7985

86+
cloudWatchLogsMetrics.recordLogSize(logLength);
8087
if (cloudWatchLogsLimits.isGreaterThanMaxEventSize(logLength)) {
8188
final String failureMessage = String.format("Event blocked due to Max Size restriction! Event Size : %s", (logLength + CloudWatchLogsLimits.APPROXIMATE_LOG_EVENT_OVERHEAD_SIZE));
82-
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, log.getData().getEventHandle(), logString, failureMessage, dlqPushHandler);
89+
DlqObject dlqObject = CloudWatchLogsSinkUtils.createDlqObject(0, log.getData().getEventHandle(), logString, failureMessage, dlqPushHandler, dropIfDlqNotConfigured);
8390
if (dlqObject != null) {
8491
dlqObjects.add(dlqObject);
92+
} else if (dropIfDlqNotConfigured) {
93+
cloudWatchLogsMetrics.increaseLogLargeEventsDroppedCounter(1);
8594
}
8695
continue;
8796
}
@@ -113,6 +122,7 @@ private void stageLogEvents() {
113122

114123
List<InputLogEvent> inputLogEvents = cloudWatchLogsDispatcher.prepareInputLogEvents(buffer.getBufferedData());
115124
cloudWatchLogsDispatcher.dispatchLogs(inputLogEvents, buffer.getEventHandles());
125+
cloudWatchLogsMetrics.recordRequestSize(buffer.getBufferSize());
116126

117127
buffer.resetBuffer();
118128
}

‎data-prepper-plugins/cloudwatch-logs/src/main/java/org/opensearch/dataprepper/plugins/sink/cloudwatch_logs/config/CloudWatchLogsSinkConfig.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ public class CloudWatchLogsSinkConfig {
3030
private PluginModel dlq;
3131

3232
@JsonProperty("threshold")
33+
@Valid
3334
private ThresholdConfig thresholdConfig = new ThresholdConfig();
3435

3536
@JsonProperty("log_group")

0 commit comments

Comments
 (0)