Adding aws-s3 metric for sqs worker utilization

CHANGELOG.next.asciidoc

@@ -238,6 +238,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]
 - Improve CEL input documentation {pull}34831[34831]
 - Add metrics documentation for CEL and AWS CloudWatch inputs. {issue}34887[34887] {pull}34889[34889]
 - Register MIME handlers for CSV types in CEL input. {pull}34934[34934]

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,7 +129,7 @@ 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)
 
 		if err := receiver.Receive(ctx); err != nil {
@@ -208,9 +208,9 @@ func (in *s3Input) createSQSReceiver(ctx v2.Context, pipeline beat.Pipeline) (*s
 	if err != nil {
 		return nil, err
 	}
-	in.metrics = newInputMetrics(ctx.ID, nil)
-	s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), in.metrics, s3API, fileSelectors, in.config.BackupConfig)
-	sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), in.metrics, sqsAPI, script, in.config.VisibilityTimeout, in.config.SQSMaxReceiveCount, pipeline, s3EventHandlerFactory)
+	in.metrics = newInputMetrics(ctx.ID, nil, in.config.MaxNumberOfMessages)
+	s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), in.metrics, s3API, fileSelectors, in.config.BackupConfig, in.config.MaxNumberOfMessages)
+	sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), in.metrics, sqsAPI, script, in.config.VisibilityTimeout, in.config.SQSMaxReceiveCount, pipeline, s3EventHandlerFactory, in.config.MaxNumberOfMessages)
 	sqsReader := newSQSReader(log.Named("sqs"), in.metrics, sqsAPI, in.config.MaxNumberOfMessages, sqsMessageHandler)
 
 	return sqsReader, nil
@@ -281,8 +281,8 @@ func (in *s3Input) createS3Lister(ctx v2.Context, cancelCtx context.Context, cli
 	if len(in.config.FileSelectors) == 0 {
 		fileSelectors = []fileSelectorConfig{{ReaderConfig: in.config.ReaderConfig}}
 	}
-	in.metrics = newInputMetrics(ctx.ID, nil)
-	s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), in.metrics, s3API, fileSelectors, in.config.BackupConfig)
+	in.metrics = newInputMetrics(ctx.ID, nil, in.config.MaxNumberOfMessages)
+	s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), in.metrics, s3API, fileSelectors, in.config.BackupConfig, in.config.MaxNumberOfMessages)
 	s3Poller := newS3Poller(log.Named("s3_poller"),
 		in.metrics,
 		s3API,

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

@@ -211,14 +211,14 @@ func benchmarkInputSQS(t *testing.T, maxMessagesInflight int) testing.BenchmarkR
 	return testing.Benchmark(func(b *testing.B) {
 		log := logp.NewLogger(inputName)
 		metricRegistry := monitoring.NewRegistry()
-		metrics := newInputMetrics("test_id", metricRegistry)
+		metrics := newInputMetrics("test_id", metricRegistry, maxMessagesInflight)
 		sqsAPI := newConstantSQS()
 		s3API := newConstantS3(t)
 		pipeline := &fakePipeline{}
 		conf := makeBenchmarkConfig(t)
 
-		s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), metrics, s3API, conf.FileSelectors, backupConfig{})
-		sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), metrics, sqsAPI, nil, time.Minute, 5, pipeline, s3EventHandlerFactory)
+		s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), metrics, s3API, conf.FileSelectors, backupConfig{}, maxMessagesInflight)
+		sqsMessageHandler := newSQSS3EventProcessor(log.Named("sqs_s3_event"), metrics, sqsAPI, nil, time.Minute, 5, pipeline, s3EventHandlerFactory, maxMessagesInflight)
 		sqsReader := newSQSReader(log.Named("sqs"), metrics, sqsAPI, maxMessagesInflight, sqsMessageHandler)
 
 		ctx, cancel := context.WithCancel(context.Background())
@@ -302,7 +302,7 @@ func benchmarkInputS3(t *testing.T, numberOfWorkers int) testing.BenchmarkResult
 		log.Infof("benchmark with %d number of workers", numberOfWorkers)
 
 		metricRegistry := monitoring.NewRegistry()
-		metrics := newInputMetrics("test_id", metricRegistry)
+		metrics := newInputMetrics("test_id", metricRegistry, numberOfWorkers)
 
 		client := pubtest.NewChanClientWithCallback(100, func(event beat.Event) {
 			event.Private.(*awscommon.EventACKTracker).ACK()
@@ -348,7 +348,7 @@ func benchmarkInputS3(t *testing.T, numberOfWorkers int) testing.BenchmarkResult
 					return
 				}
 
-				s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), metrics, s3API, config.FileSelectors, backupConfig{})
+				s3EventHandlerFactory := newS3ObjectProcessorFactory(log.Named("s3"), metrics, s3API, config.FileSelectors, backupConfig{}, numberOfWorkers)
 				s3Poller := newS3Poller(logp.NewLogger(inputName), metrics, s3API, client, s3EventHandlerFactory, newStates(inputCtx), store, "bucket", listPrefix, "region", "provider", numberOfWorkers, time.Second)
 
 				if err := s3Poller.Poll(ctx); err != nil {

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

@@ -5,7 +5,10 @@
 package awss3
 
 import (
+	"context"
 	"io"
+	"sync/atomic"
+	"time"
 
 	"github.com/rcrowley/go-metrics"
 
@@ -17,15 +20,20 @@ import (
 type inputMetrics struct {
 	registry   *monitoring.Registry
 	unregister func()
+	ctx        context.Context
+	cancel     context.CancelFunc
 
-	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.
+	utilizationNanos int64
+
+	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.
@@ -37,11 +45,17 @@ type inputMetrics struct {
 	s3ObjectProcessingTime  metrics.Sample   // Histogram of the elapsed S3 object processing times in nanoseconds (start of download to completion of parsing).
 }
 
-// Close removes the metrics from the registry.
+// Close cancels the context and removes the metrics from the registry.
 func (m *inputMetrics) Close() {
+	m.cancel()
 	m.unregister()
 }
 
+func (m *inputMetrics) updateSQSProcessingTime(d time.Duration) {
+	m.sqsMessageProcessingTime.Update(d.Nanoseconds())
+	atomic.AddInt64(&m.utilizationNanos, d.Nanoseconds())
+}
+
 func (m *inputMetrics) setSQSMessagesWaiting(count int64) {
 	if m.sqsMessagesWaiting == nil {
 		// if metric not initialized, and count is -1, do nothing
@@ -54,17 +68,58 @@ func (m *inputMetrics) setSQSMessagesWaiting(count int64) {
 	m.sqsMessagesWaiting.Set(count)
 }
 
-func newInputMetrics(id string, optionalParent *monitoring.Registry) *inputMetrics {
+func calculateUtilizationAndReset(d time.Duration, maxMessagesInflight int, m *inputMetrics) float64 {
+	maxUtilization := float64(d) * float64(maxMessagesInflight)
+	utilizedRate := float64(atomic.SwapInt64(&m.utilizationNanos, 0)) / maxUtilization
+
+	if utilizedRate == 0 {
+		// Falling back to inflight stats when the workers are long running
+		inflight := m.sqsMessagesInflight.Get()
+		if inflight > 0 {
+			return float64(inflight) / float64(maxMessagesInflight)
+		}
+	} else if utilizedRate > 1 {
+		// Normalizing the utilization after long running workers
+		return 1
+	}
+
+	return utilizedRate
+}
+
+func (m *inputMetrics) pollSQSUtilizationMetric(ctx context.Context, maxMessagesInflight int) {
+	t := time.NewTicker(5 * time.Second)
+	defer t.Stop()
+
+	lastTick := time.Now()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case tick := <-t.C:
+			duration := tick.Sub(lastTick)
+			utilizedRate := calculateUtilizationAndReset(duration, maxMessagesInflight, m)
+			m.sqsWorkerUtilization.Set(utilizedRate)
+			// reset for the next polling duration
+			lastTick = tick
+		}
+	}
+}
+
+func newInputMetrics(id string, optionalParent *monitoring.Registry, maxWorkers int) *inputMetrics {
 	reg, unreg := inputmon.NewInputRegistry(inputName, id, optionalParent)
+	ctx, cancelInputCtx := context.WithCancel(context.Background())
 
 	out := &inputMetrics{
 		registry:                            reg,
 		unregister:                          unreg,
+		ctx:                                 ctx,
+		cancel:                              cancelInputCtx,
 		sqsMessagesReceivedTotal:            monitoring.NewUint(reg, "sqs_messages_received_total"),
 		sqsVisibilityTimeoutExtensionsTotal: monitoring.NewUint(reg, "sqs_visibility_timeout_extensions_total"),
 		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"),
@@ -82,6 +137,9 @@ func newInputMetrics(id string, optionalParent *monitoring.Registry) *inputMetri
 		Register("histogram", metrics.NewHistogram(out.sqsLagTime)) //nolint:errcheck // A unique namespace is used so name collisions are impossible.
 	adapter.NewGoMetrics(reg, "s3_object_processing_time", adapter.Accept).
 		Register("histogram", metrics.NewHistogram(out.s3ObjectProcessingTime)) //nolint:errcheck // A unique namespace is used so name collisions are impossible.
+
+	go out.pollSQSUtilizationMetric(ctx, maxWorkers)
+
 	return out
 }
 

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

@@ -6,6 +6,9 @@ package awss3
 
 import (
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
 
 	"github.com/elastic/elastic-agent-libs/monitoring"
 )
@@ -17,10 +20,44 @@ import (
 func TestInputMetricsClose(t *testing.T) {
 	reg := monitoring.NewRegistry()
 
-	metrics := newInputMetrics("aws-s3-aws.cloudfront_logs-8b312b5f-9f99-492c-b035-3dff354a1f01", reg)
+	metrics := newInputMetrics("aws-s3-aws.cloudfront_logs-8b312b5f-9f99-492c-b035-3dff354a1f01", reg, 1)
 	metrics.Close()
 
 	reg.Do(monitoring.Full, func(s string, _ interface{}) {
 		t.Errorf("registry should be empty, but found %v", s)
 	})
 }
+
+func TestInputMetricsSQSWorkerUtilization(t *testing.T) {
+	reg := monitoring.NewRegistry()
+	const maxWorkers = 1
+	metrics := newInputMetrics("test", reg, maxWorkers)
+
+	metrics.utilizationNanos = time.Second.Nanoseconds()
+
+	utilization := calculateUtilizationAndReset(time.Second, maxWorkers, metrics)
+	assert.Greater(t, utilization, 0.95)
+}
+
+func TestInputMetricsSQSWorkerUtilization_LongRunningWorkers(t *testing.T) {
+	reg := monitoring.NewRegistry()
+	const maxWorkers = 1
+	metrics := newInputMetrics("test", reg, maxWorkers)
+
+	metrics.utilizationNanos = time.Minute.Nanoseconds()
+
+	utilization := calculateUtilizationAndReset(time.Second, maxWorkers, metrics)
+	assert.Equal(t, utilization, 1.0)
+}
+
+func TestInputMetricsSQSWorkerUtilization_InFlightWorkers(t *testing.T) {
+	reg := monitoring.NewRegistry()
+	const maxWorkers = 1
+	metrics := newInputMetrics("test", reg, maxWorkers)
+
+	metrics.utilizationNanos = 0
+	metrics.sqsMessagesInflight.Set(maxWorkers)
+
+	utilization := calculateUtilizationAndReset(time.Second, maxWorkers, metrics)
+	assert.Equal(t, utilization, 1.0)
+}

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

@@ -77,7 +77,7 @@ func newS3Poller(log *logp.Logger,
 	bucketPollInterval time.Duration,
 ) *s3Poller {
 	if metrics == nil {
-		metrics = newInputMetrics("", monitoring.NewRegistry())
+		metrics = newInputMetrics("", monitoring.NewRegistry(), numberOfWorkers)
 	}
 	return &s3Poller{
 		numberOfWorkers:      numberOfWorkers,

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

@@ -45,9 +45,9 @@ type s3ObjectProcessorFactory struct {
 	backupConfig  backupConfig
 }
 
-func newS3ObjectProcessorFactory(log *logp.Logger, metrics *inputMetrics, s3 s3API, sel []fileSelectorConfig, backupConfig backupConfig) *s3ObjectProcessorFactory {
+func newS3ObjectProcessorFactory(log *logp.Logger, metrics *inputMetrics, s3 s3API, sel []fileSelectorConfig, backupConfig backupConfig, maxWorkers int) *s3ObjectProcessorFactory {
 	if metrics == nil {
-		metrics = newInputMetrics("", monitoring.NewRegistry())
+		metrics = newInputMetrics("", monitoring.NewRegistry(), maxWorkers)
 	}
 	if len(sel) == 0 {
 		sel = []fileSelectorConfig{

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

@@ -144,7 +144,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 			GetObject(gomock.Any(), gomock.Eq(s3Event.S3.Bucket.Name), gomock.Eq(s3Event.S3.Object.Key)).
 			Return(nil, errFakeConnectivityFailure)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{})
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{}, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).ProcessS3Object()
 		require.Error(t, err)
@@ -166,7 +166,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 			GetObject(gomock.Any(), gomock.Eq(s3Event.S3.Bucket.Name), gomock.Eq(s3Event.S3.Object.Key)).
 			Return(nil, nil)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{})
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{}, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).ProcessS3Object()
 		require.Error(t, err)
@@ -193,7 +193,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 				Times(2),
 		)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{})
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupConfig{}, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).ProcessS3Object()
 		require.NoError(t, err)
@@ -219,7 +219,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 				Return(nil, nil),
 		)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg)
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).FinalizeS3Object()
 		require.NoError(t, err)
@@ -249,7 +249,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 				Return(nil, nil),
 		)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg)
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).FinalizeS3Object()
 		require.NoError(t, err)
@@ -276,7 +276,7 @@ func TestS3ObjectProcessor(t *testing.T) {
 				Return(nil, nil),
 		)
 
-		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg)
+		s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, nil, backupCfg, 1)
 		ack := awscommon.NewEventACKTracker(ctx)
 		err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).FinalizeS3Object()
 		require.NoError(t, err)
@@ -322,7 +322,7 @@ func _testProcessS3Object(t testing.TB, file, contentType string, numEvents int,
 			Times(numEvents),
 	)
 
-	s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, selectors, backupConfig{})
+	s3ObjProc := newS3ObjectProcessorFactory(logp.NewLogger(inputName), nil, mockS3API, selectors, backupConfig{}, 1)
 	ack := awscommon.NewEventACKTracker(ctx)
 	err := s3ObjProc.Create(ctx, logp.NewLogger(inputName), mockPublisher, ack, s3Event).ProcessS3Object()