Add metric for sqs receive message failures for s3-sqs source (#6102)

Signed-off-by: Taylor Gray <[email protected]>

Author: graytaylor0

data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorker.java

@@ -49,6 +49,7 @@
 public class SqsWorker implements Runnable {
     private static final Logger LOG = LoggerFactory.getLogger(SqsWorker.class);
     static final String SQS_MESSAGES_RECEIVED_METRIC_NAME = "sqsMessagesReceived";
+    static final String SQS_RECEIVE_MESSAGES_FAILED_METRIC_NAME = "sqsReceiveMessageFailed";
     static final String SQS_MESSAGES_DELETED_METRIC_NAME = "sqsMessagesDeleted";
     static final String SQS_MESSAGES_FAILED_METRIC_NAME = "sqsMessagesFailed";
     static final String SQS_MESSAGES_DELETE_FAILED_METRIC_NAME = "sqsMessagesDeleteFailed";
@@ -65,6 +66,7 @@ public class SqsWorker implements Runnable {
     private final S3EventFilter objectCreatedFilter;
     private final S3EventFilter evenBridgeObjectCreatedFilter;
     private final Counter sqsMessagesReceivedCounter;
+    private final Counter sqsReceiveMessagesFailedCounter;
     private final Counter sqsMessagesDeletedCounter;
     private final Counter sqsMessagesFailedCounter;
     private final Counter s3ObjectsEmptyCounter;
@@ -105,6 +107,7 @@ public SqsWorker(final AcknowledgementSetManager acknowledgementSetManager,
         s3ObjectsEmptyCounter = pluginMetrics.counter(S3_OBJECTS_EMPTY_METRIC_NAME);
         sqsMessagesDeleteFailedCounter = pluginMetrics.counter(SQS_MESSAGES_DELETE_FAILED_METRIC_NAME);
         sqsMessageDelayTimer = pluginMetrics.timer(SQS_MESSAGE_DELAY_METRIC_NAME);
+        sqsReceiveMessagesFailedCounter = pluginMetrics.counter(SQS_RECEIVE_MESSAGES_FAILED_METRIC_NAME);
         acknowledgementSetCallbackCounter = pluginMetrics.counter(ACKNOWLEDGEMENT_SET_CALLACK_METRIC_NAME);
         sqsVisibilityTimeoutChangedCount = pluginMetrics.counter(SQS_VISIBILITY_TIMEOUT_CHANGED_COUNT_METRIC_NAME);
         sqsVisibilityTimeoutChangeFailedCount = pluginMetrics.counter(SQS_VISIBILITY_TIMEOUT_CHANGE_FAILED_COUNT_METRIC_NAME);
@@ -161,6 +164,7 @@ private List<Message> getMessagesFromSqs() {
             return messages;
         } catch (final SqsException | StsException e) {
             LOG.error("Error reading from SQS: {}. Retrying with exponential backoff.", e.getMessage());
+            sqsReceiveMessagesFailedCounter.increment();
             applyBackoff();
             return Collections.emptyList();
         }

data-prepper-plugins/s3-source/src/test/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorkerTest.java

@@ -86,6 +86,7 @@
 import static org.opensearch.dataprepper.plugins.source.s3.SqsWorker.SQS_MESSAGES_FAILED_METRIC_NAME;
 import static org.opensearch.dataprepper.plugins.source.s3.SqsWorker.SQS_MESSAGES_RECEIVED_METRIC_NAME;
 import static org.opensearch.dataprepper.plugins.source.s3.SqsWorker.SQS_MESSAGE_DELAY_METRIC_NAME;
+import static org.opensearch.dataprepper.plugins.source.s3.SqsWorker.SQS_RECEIVE_MESSAGES_FAILED_METRIC_NAME;
 import static org.opensearch.dataprepper.plugins.source.s3.SqsWorker.SQS_VISIBILITY_TIMEOUT_CHANGED_COUNT_METRIC_NAME;
 
 @ExtendWith(MockitoExtension.class)
@@ -97,6 +98,7 @@ class SqsWorkerTest {
     private PluginMetrics pluginMetrics;
     private Backoff backoff;
     private Counter sqsMessagesReceivedCounter;
+    private Counter sqsReceiveMessageFailedCounter;
     private Counter sqsMessagesDeletedCounter;
     private Counter sqsMessagesFailedCounter;
     private Counter sqsMessagesDeleteFailedCounter;
@@ -132,6 +134,7 @@ void setUp() {
         sqsMessagesDeletedCounter = mock(Counter.class);
         sqsMessagesFailedCounter = mock(Counter.class);
         sqsMessagesDeleteFailedCounter = mock(Counter.class);
+        sqsReceiveMessageFailedCounter = mock(Counter.class);
         s3ObjectsEmptyCounter = mock(Counter.class);
         sqsMessageDelayTimer = mock(Timer.class);
         when(pluginMetrics.counter(SQS_MESSAGES_RECEIVED_METRIC_NAME)).thenReturn(sqsMessagesReceivedCounter);
@@ -141,6 +144,7 @@ void setUp() {
         when(pluginMetrics.counter(S3_OBJECTS_EMPTY_METRIC_NAME)).thenReturn(s3ObjectsEmptyCounter);
         when(pluginMetrics.timer(SQS_MESSAGE_DELAY_METRIC_NAME)).thenReturn(sqsMessageDelayTimer);
         when(pluginMetrics.counter(ACKNOWLEDGEMENT_SET_CALLACK_METRIC_NAME)).thenReturn(mock(Counter.class));
+        when(pluginMetrics.counter(SQS_RECEIVE_MESSAGES_FAILED_METRIC_NAME)).thenReturn(sqsReceiveMessageFailedCounter);
         when(pluginMetrics.counter(SQS_VISIBILITY_TIMEOUT_CHANGED_COUNT_METRIC_NAME)).thenReturn(sqsVisibilityTimeoutChangedCount);
     }
 
@@ -531,6 +535,7 @@ void processSqsMessages_should_return_zero_messages_when_a_SqsException_is_throw
         final int messagesProcessed = createObjectUnderTest().processSqsMessages();
         assertThat(messagesProcessed, equalTo(0));
         verify(sqsClient, never()).deleteMessageBatch(any(DeleteMessageBatchRequest.class));
+        verify(sqsReceiveMessageFailedCounter).increment();
     }
 
     @Test