Adding aws-s3 metric for sqs worker utilization

CHANGELOG.next.asciidoc

@@ -232,6 +232,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add Basic Authentication support on constructed requests to CEL input {issue}34609[34609] {pull}34689[34689]
 - Add string manipulation extensions to CEL input {issue}34610[34610] {pull}34689[34689]
 - Add unix socket log parsing for nginx ingress_controller {pull}34732[34732]
+- Added metric `sqs_worker_utilization` for aws-s3 input. {pull}34793[34793]
 
 *Auditbeat*
 

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

@@ -738,6 +738,7 @@ observe the activity of the input.
 | `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_worker_utilization`                  | Rate of SQS worker utilization over previous 5 seconds. 0 indicates idle, 1 indicates all workers utilized.
 | `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

@@ -129,10 +129,13 @@ func (in *s3Input) Run(inputContext v2.Context, pipeline beat.Pipeline) error {
 		}
 		defer receiver.metrics.Close()
 
-		// Poll sqs waiting metric periodically in the background.
+		// Poll metrics periodically in the background
 		go pollSqsWaitingMetric(ctx, receiver)
+		metricReporterChan := make(chan int64)
+		defer close(metricReporterChan)
+		go pollSqsUtilizationMetric(ctx, receiver, metricReporterChan)
 
-		if err := receiver.Receive(ctx); err != nil {
+		if err := receiver.Receive(ctx, metricReporterChan); err != nil {
 			return err
 		}
 	}
@@ -410,5 +413,31 @@ func pollSqsWaitingMetric(ctx context.Context, receiver *sqsReader) {
 	}
 }
 
+func pollSqsUtilizationMetric(ctx context.Context, receiver *sqsReader, metricReporterChan chan int64) {
+	pollDur := 5 * time.Second
+	t := time.NewTicker(pollDur)
+	defer t.Stop()
+
+	utilizedNanos := new(int64)
+	go func() {
+		for elem := range metricReporterChan {
+			*utilizedNanos += elem
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			denom := pollDur.Nanoseconds() * int64(receiver.maxMessagesInflight)
+			utilizedRate := float64(*utilizedNanos) / float64(denom)
+			receiver.metrics.sqsWorkerUtilization.Set(utilizedRate)
+			// reset for the next polling duration
+			*utilizedNanos = 0
+		}
+	}
+}
+
 // boolPtr returns a pointer to b.
 func boolPtr(b bool) *bool { return &b }

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

@@ -233,7 +233,7 @@ func benchmarkInputSQS(t *testing.T, maxMessagesInflight int) testing.BenchmarkR
 
 		b.ResetTimer()
 		start := time.Now()
-		if err := sqsReader.Receive(ctx); err != nil {
+		if err := sqsReader.Receive(ctx, nil); err != nil {
 			if !errors.Is(err, context.DeadlineExceeded) {
 				t.Fatal(err)
 			}

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

@@ -192,6 +192,7 @@ func TestInputRunSQS(t *testing.T) {
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
 	assert.Greater(t, s3Input.metrics.sqsLagTime.Mean(), 0.0)
+	assert.Greater(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0)
 }
 
 func TestInputRunS3(t *testing.T) {
@@ -430,4 +431,5 @@ func TestInputRunSNS(t *testing.T) {
 	assert.EqualValues(t, s3Input.metrics.s3ObjectsRequestedTotal.Get(), 7)
 	assert.EqualValues(t, s3Input.metrics.s3EventsCreatedTotal.Get(), 12)
 	assert.Greater(t, s3Input.metrics.sqsLagTime.Mean(), 0.0)
+	assert.Greater(t, s3Input.metrics.sqsWorkerUtilization.Get(), 0.0)
 }

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

@@ -18,14 +18,15 @@ 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.
+	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.
+	sqsWorkerUtilization                *monitoring.Float // Rate of SQS worker utilization over previous 5 seconds. 0 indicates idle, 1 indicates all workers utilized.
+	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.
 
 	s3ObjectsRequestedTotal *monitoring.Uint // Number of S3 objects downloaded.
 	s3ObjectsAckedTotal     *monitoring.Uint // Number of S3 objects processed that were fully ACKed.
@@ -65,6 +66,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"),
+		sqsWorkerUtilization:                monitoring.NewFloat(reg, "sqs_worker_utilization"),
 		sqsMessageProcessingTime:            metrics.NewUniformSample(1024),
 		sqsLagTime:                          metrics.NewUniformSample(1024),
 		s3ObjectsRequestedTotal:             monitoring.NewUint(reg, "s3_objects_requested_total"),

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

@@ -47,7 +47,7 @@ func newSQSReader(log *logp.Logger, metrics *inputMetrics, sqs sqsAPI, maxMessag
 	}
 }
 
-func (r *sqsReader) Receive(ctx context.Context) error {
+func (r *sqsReader) Receive(ctx context.Context, metricReporterChan chan int64) error {
 	// This loop tries to keep the workers busy as much as possible while
 	// honoring the max message cap as opposed to a simpler loop that receives
 	// N messages, waits for them all to finish, then requests N more messages.
@@ -61,6 +61,7 @@ func (r *sqsReader) Receive(ctx context.Context) error {
 
 		// Receive (at most) as many SQS messages as there are workers.
 		msgs, err := r.sqs.ReceiveMessage(ctx, workers)
+		r.log.Warnw("Messages received", "msg count", len(msgs))
 		if err != nil {
 			r.workerSem.Release(workers)
 
@@ -85,7 +86,9 @@ func (r *sqsReader) Receive(ctx context.Context) error {
 			go func(msg types.Message, start time.Time) {
 				defer func() {
 					r.metrics.sqsMessagesInflight.Dec()
-					r.metrics.sqsMessageProcessingTime.Update(time.Since(start).Nanoseconds())
+					processingTimeNanos := time.Since(start).Nanoseconds()
+					r.metrics.sqsMessageProcessingTime.Update(processingTimeNanos)
+					metricReporterChan <- processingTimeNanos
 					workerWg.Done()
 					r.workerSem.Release(1)
 				}()

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

@@ -74,7 +74,7 @@ func TestSQSReceiver(t *testing.T) {
 
 		// Execute sqsReader and verify calls/state.
 		receiver := newSQSReader(logp.NewLogger(inputName), nil, mockAPI, maxMessages, mockMsgHandler)
-		require.NoError(t, receiver.Receive(ctx))
+		require.NoError(t, receiver.Receive(ctx, nil))
 		assert.Equal(t, maxMessages, receiver.workerSem.Available())
 	})
 
@@ -107,7 +107,7 @@ func TestSQSReceiver(t *testing.T) {
 
 		// Execute SQSReceiver and verify calls/state.
 		receiver := newSQSReader(logp.NewLogger(inputName), nil, mockAPI, maxMessages, mockMsgHandler)
-		require.NoError(t, receiver.Receive(ctx))
+		require.NoError(t, receiver.Receive(ctx, nil))
 		assert.Equal(t, maxMessages, receiver.workerSem.Available())
 	})
 }