Adding aws-s3 metric for approximate messages waiting

x-pack/filebeat/input/awss3/sqs_test.go

@@ -24,6 +24,7 @@ import (
 const testTimeout = 10 * time.Second
 
 var errFakeConnectivityFailure = errors.New("fake connectivity failure")
+var errFakeGetAttributeFailute = errors.New("something went wrong")
 
 func TestSQSReceiver(t *testing.T) {
 	err := logp.TestingSetup()
@@ -109,6 +110,61 @@ func TestSQSReceiver(t *testing.T) {
 	})
 }
 
+func TestGetApproximateMessageCount(t *testing.T) {
+	err := logp.TestingSetup()
+	assert.Nil(t, err)
+
+	const maxMessages = 5
+	const count = 500
+	attrName := []types.QueueAttributeName{sqsApproximateNumberOfMessages}
+	attr := map[string]string{"ApproximateNumberOfMessages": "500"}
+
+	t.Run("GetApproximateMessageCount success", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+		defer cancel()
+
+		ctrl, ctx := gomock.WithContext(ctx, t)
+		defer ctrl.Finish()
+		mockAPI := NewMockSQSAPI(ctrl)
+		mockMsgHandler := NewMockSQSProcessor(ctrl)
+
+		gomock.InOrder(
+			mockAPI.EXPECT().
+				GetQueueAttributes(gomock.Any(), gomock.Eq(attrName)).
+				Times(1).
+				DoAndReturn(func(_ context.Context, _ []types.QueueAttributeName) (map[string]string, error) {
+					return attr, nil
+				}),
+		)
+
+		receiver := newSQSReader(logp.NewLogger(inputName), nil, mockAPI, maxMessages, mockMsgHandler)
+		assert.Equal(t, count, receiver.GetApproximateMessageCount(ctx))
+	})
+
+	t.Run("GetApproximateMessageCount error", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+		defer cancel()
+
+		ctrl, ctx := gomock.WithContext(ctx, t)
+		defer ctrl.Finish()
+
+		mockAPI := NewMockSQSAPI(ctrl)
+		mockMsgHandler := NewMockSQSProcessor(ctrl)
+
+		gomock.InOrder(
+			mockAPI.EXPECT().
+				GetQueueAttributes(gomock.Any(), gomock.Eq(attrName)).
+				Times(1).
+				DoAndReturn(func(_ context.Context, _ []types.QueueAttributeName) (map[string]string, error) {
+					return nil, errFakeGetAttributeFailute
+				}),
+		)
+
+		receiver := newSQSReader(logp.NewLogger(inputName), nil, mockAPI, maxMessages, mockMsgHandler)
+		assert.Equal(t, -1, receiver.GetApproximateMessageCount(ctx))
+	})
+}
+
 func newSQSMessage(events ...s3EventV2) types.Message {
 	body, err := json.Marshal(s3EventsV2{Records: events})
 	if err != nil {