Adding aws-s3 metric for approximate messages waiting

CHANGELOG.next.asciidoc

@@ -196,6 +196,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 - 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) (optional).
 | `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

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"net/url"
 	"strings"
+	"time"
 
 	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -123,6 +124,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 +380,23 @@ 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 := receiver.GetApproximateMessageCount(ctx)
+			if count == -1 {
+				// stop polling if error is encountered
+				return
+			}
+			receiver.metrics.sqsMessagesWaiting.Set(uint64(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

@@ -22,6 +22,7 @@ type inputMetrics struct {
 	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.Uint // Number of SQS messages waiting in the SQS Queue (gauge) (optional).
 	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.
 
@@ -50,6 +51,7 @@ func newInputMetrics(id string, optionalParent *monitoring.Registry) *inputMetri
 		sqsMessagesInflight:                 monitoring.NewUint(reg, "sqs_messages_inflight_gauge"),
 		sqsMessagesReturnedTotal:            monitoring.NewUint(reg, "sqs_messages_returned_total"),
 		sqsMessagesDeletedTotal:             monitoring.NewUint(reg, "sqs_messages_deleted_total"),
+		sqsMessagesWaiting:                  monitoring.NewUint(reg, "sqs_messages_waiting_gauge"),
 		sqsMessageProcessingTime:            metrics.NewUniformSample(1024),
 		sqsLagTime:                          metrics.NewUniformSample(1024),
 		s3ObjectsRequestedTotal:             monitoring.NewUint(reg, "s3_objects_requested_total"),