Adding aws-s3 metric for approximate messages waiting

CHANGELOG.next.asciidoc

@@ -212,6 +212,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add pagination support for Salesforce module. {issue}34057[34057] {pull}34065[34065]
 - Allow users to redact sensitive data from CEL input debug logs. {pull}34302[34302]
 - Added support for HTTP destination override to Google Cloud Storage input. {pull}34413[34413]
+- Added metric `sqs_messages_waiting_gauge` for aws-s3 input. {pull}34488[34488]
 - Add support for new Rabbitmq timestamp format for logs {pull}34211[34211]
 - Allow user configuration of timezone offset in Cisco ASA and FTD modules. {pull}34436[34436]
 - Allow user configuration of timezone offset in Checkpoint module. {pull}34472[34472]

x-pack/filebeat/docs/inputs/input-aws-s3.asciidoc

@@ -497,6 +497,11 @@ In case `delete_after_backup` is set the following permission is required as wel
 s3:DeleteObject
 ----
 
+In case optional SQS metric `sqs_messages_waiting_gauge` is desired, the following permission is required:
+----
+sqs:GetQueueAttributes
+----
+
 [float]
 === S3 and SQS setup
 
@@ -732,6 +737,7 @@ observe the activity of the input.
 | `sqs_messages_inflight_gauge`             | Number of SQS messages inflight (gauge).
 | `sqs_messages_returned_total`             | Number of SQS message returned to queue (happens on errors implicitly after visibility timeout passes).
 | `sqs_messages_deleted_total`              | Number of SQS messages deleted.
+| `sqs_messages_waiting_gauge`              | Number of SQS messages waiting in the SQS queue (gauge). The value is refreshed every minute via data from GetQueueAttributes.
 | `sqs_message_processing_time`             | Histogram of the elapsed SQS processing times in nanoseconds (time of receipt to time of delete/return).
 | `sqs_lag_time`                            | Histogram of the difference between the SQS SentTimestamp attribute and the time when the SQS message was received expressed in nanoseconds.
 | `s3_objects_requested_total`              | Number of S3 objects downloaded.

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

@@ -6,13 +6,16 @@ package awss3
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/url"
 	"strings"
+	"time"
 
 	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/aws/smithy-go"
 
 	"github.com/elastic/beats/v7/filebeat/beater"
 	v2 "github.com/elastic/beats/v7/filebeat/input/v2"
@@ -24,7 +27,10 @@ import (
 	"github.com/elastic/go-concert/unison"
 )
 
-const inputName = "aws-s3"
+const (
+	inputName                = "aws-s3"
+	sqsAccessDeniedErrorCode = "AccessDeniedException"
+)
 
 func Plugin(store beater.StateStore) v2.Plugin {
 	return v2.Plugin{
@@ -123,6 +129,9 @@ func (in *s3Input) Run(inputContext v2.Context, pipeline beat.Pipeline) error {
 		}
 		defer receiver.metrics.Close()
 
+		// Poll sqs waiting metric periodically in the background.
+		go pollSqsWaitingMetric(ctx, receiver)
+
 		if err := receiver.Receive(ctx); err != nil {
 			return err
 		}
@@ -376,5 +385,30 @@ func getProviderFromDomain(endpoint string, ProviderOverride string) string {
 	return "unknown"
 }
 
+func pollSqsWaitingMetric(ctx context.Context, receiver *sqsReader) {
+	t := time.NewTicker(time.Minute)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			count, err := receiver.GetApproximateMessageCount(ctx)
+
+			var apiError smithy.APIError
+			if errors.As(err, &apiError) {
+				switch apiError.ErrorCode() {
+				case sqsAccessDeniedErrorCode:
+					// stop polling if auth error is encountered
+					receiver.metrics.setSQSMessagesWaiting(int64(count))
+					return
+				}
+			}
+
+			receiver.metrics.setSQSMessagesWaiting(int64(count))
+		}
+	}
+}
+
 // boolPtr returns a pointer to b.
 func boolPtr(b bool) *bool { return &b }

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

@@ -67,6 +67,10 @@ func (*constantSQS) ChangeMessageVisibility(ctx context.Context, msg *sqsTypes.M
 	return nil
 }
 
+func (c *constantSQS) GetQueueAttributes(ctx context.Context, attr []sqsTypes.QueueAttributeName) (map[string]string, error) {
+	return map[string]string{}, nil
+}
+
 type s3PagerConstant struct {
 	mutex        *sync.Mutex
 	objects      []s3Types.Object

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

@@ -187,6 +187,7 @@ func TestInputRunSQS(t *testing.T) {
 	assert.EqualValues(t, s3Input.metrics.sqsMessagesDeletedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.sqsMessagesReturnedTotal.Get(), 1) // Invalid JSON is returned so that it can eventually be DLQed.
 	assert.EqualValues(t, s3Input.metrics.sqsVisibilityTimeoutExtensionsTotal.Get(), 0)
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesWaiting.Get(), 0)
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsInflight.Get(), 0)
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
@@ -424,6 +425,7 @@ func TestInputRunSNS(t *testing.T) {
 	assert.EqualValues(t, s3Input.metrics.sqsMessagesDeletedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.sqsMessagesReturnedTotal.Get(), 1) // Invalid JSON is returned so that it can eventually be DLQed.
 	assert.EqualValues(t, s3Input.metrics.sqsVisibilityTimeoutExtensionsTotal.Get(), 0)
+	assert.EqualValues(t, s3Input.metrics.sqsMessagesWaiting.Get(), 0)
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsInflight.Get(), 0)
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)

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

@@ -41,6 +41,7 @@ type sqsAPI interface {
 	sqsReceiver
 	sqsDeleter
 	sqsVisibilityChanger
+	sqsAttributeGetter
 }
 
 type sqsReceiver interface {
@@ -55,6 +56,10 @@ type sqsVisibilityChanger interface {
 	ChangeMessageVisibility(ctx context.Context, msg *types.Message, timeout time.Duration) error
 }
 
+type sqsAttributeGetter interface {
+	GetQueueAttributes(ctx context.Context, attr []types.QueueAttributeName) (map[string]string, error)
+}
+
 type sqsProcessor interface {
 	// ProcessSQS processes and SQS message. It takes fully ownership of the
 	// given message and is responsible for updating the message's visibility
@@ -197,6 +202,25 @@ func (a *awsSQSAPI) ChangeMessageVisibility(ctx context.Context, msg *types.Mess
 	return nil
 }
 
+func (a *awsSQSAPI) GetQueueAttributes(ctx context.Context, attr []types.QueueAttributeName) (map[string]string, error) {
+	ctx, cancel := context.WithTimeout(ctx, a.apiTimeout)
+	defer cancel()
+
+	attributeOutput, err := a.client.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{
+		AttributeNames: attr,
+		QueueUrl:       awssdk.String(a.queueURL),
+	})
+
+	if err != nil {
+		if errors.Is(err, context.DeadlineExceeded) {
+			err = fmt.Errorf("api_timeout exceeded: %w", err)
+		}
+		return nil, fmt.Errorf("sqs GetQueueAttributes failed: %w", err)
+	}
+
+	return attributeOutput.Attributes, nil
+}
+
 // ------
 // AWS S3 implementation
 // ------

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

@@ -15,13 +15,15 @@ import (
 )
 
 type inputMetrics struct {
+	registry   *monitoring.Registry
 	unregister func()
 
 	sqsMessagesReceivedTotal            *monitoring.Uint // Number of SQS messages received (not necessarily processed fully).
 	sqsVisibilityTimeoutExtensionsTotal *monitoring.Uint // Number of SQS visibility timeout extensions.
 	sqsMessagesInflight                 *monitoring.Uint // Number of SQS messages inflight (gauge).
 	sqsMessagesReturnedTotal            *monitoring.Uint // Number of SQS message returned to queue (happens on errors implicitly after visibility timeout passes).
 	sqsMessagesDeletedTotal             *monitoring.Uint // Number of SQS messages deleted.
+	sqsMessagesWaiting                  *monitoring.Int  // Number of SQS messages waiting in the SQS queue (gauge). The value is refreshed every minute via data from GetQueueAttributes.
 	sqsMessageProcessingTime            metrics.Sample   // Histogram of the elapsed SQS processing times in nanoseconds (time of receipt to time of delete/return).
 	sqsLagTime                          metrics.Sample   // Histogram of the difference between the SQS SentTimestamp attribute and the time when the SQS message was received expressed in nanoseconds.
 
@@ -40,10 +42,21 @@ func (m *inputMetrics) Close() {
 	m.unregister()
 }
 
+func (m *inputMetrics) setSQSMessagesWaiting(count int64) {
+	if count == -1 && m.sqsMessagesWaiting == nil {
+		// if metric not initialized, and count is -1, do nothing
+		return
+	}
+
+	m.sqsMessagesWaiting = monitoring.NewInt(m.registry, "sqs_messages_waiting_gauge")
+	m.sqsMessagesWaiting.Set(count)
+}
+
 func newInputMetrics(id string, optionalParent *monitoring.Registry) *inputMetrics {
 	reg, unreg := inputmon.NewInputRegistry(inputName, id, optionalParent)
 
 	out := &inputMetrics{
+		registry:                            reg,
 		unregister:                          unreg,
 		sqsMessagesReceivedTotal:            monitoring.NewUint(reg, "sqs_messages_received_total"),
 		sqsVisibilityTimeoutExtensionsTotal: monitoring.NewUint(reg, "sqs_visibility_timeout_extensions_total"),