aws-s3 input: Split S3 poller and SQS reader into explicit input objects

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

@@ -9,6 +9,10 @@ import (
 	"fmt"
 	"time"
 
+	awssdk "github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/dustin/go-humanize"
 
 	"github.com/elastic/beats/v7/libbeat/common/cfgtype"
@@ -222,3 +226,59 @@ func (rc *readerConfig) InitDefaults() {
 	rc.MaxBytes = 10 * humanize.MiByte
 	rc.LineTerminator = readfile.AutoLineTerminator
 }
+
+func (c config) getBucketName() string {
+	if c.NonAWSBucketName != "" {
+		return c.NonAWSBucketName
+	}
+	if c.BucketARN != "" {
+		return getBucketNameFromARN(c.BucketARN)
+	}
+	return ""
+}
+
+func (c config) getBucketARN() string {
+	if c.NonAWSBucketName != "" {
+		return c.NonAWSBucketName
+	}
+	if c.BucketARN != "" {
+		return c.BucketARN
+	}
+	return ""
+}
+
+// An AWS SDK callback to apply the input configuration's settings to an S3
+// options struct.
+// Should be provided as a parameter to s3.NewFromConfig.
+func (c config) s3ConfigModifier(o *s3.Options) {
+	if c.NonAWSBucketName != "" {
+		o.EndpointResolver = nonAWSBucketResolver{endpoint: c.AWSConfig.Endpoint}
+	}
+
+	if c.AWSConfig.FIPSEnabled {
+		o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
+	}
+	o.UsePathStyle = c.PathStyle
+
+	o.Retryer = retry.NewStandard(func(so *retry.StandardOptions) {
+		so.MaxAttempts = 5
+		// Recover quickly when requests start working again
+		so.NoRetryIncrement = 100
+	})
+}
+
+// An AWS SDK callback to apply the input configuration's settings to an SQS
+// options struct.
+// Should be provided as a parameter to sqs.NewFromConfig.
+func (c config) sqsConfigModifier(o *sqs.Options) {
+	if c.AWSConfig.FIPSEnabled {
+		o.EndpointOptions.UseFIPSEndpoint = awssdk.FIPSEndpointStateEnabled
+	}
+}
+
+func (c config) getFileSelectors() []fileSelectorConfig {
+	if len(c.FileSelectors) > 0 {
+		return c.FileSelectors
+	}
+	return []fileSelectorConfig{{ReaderConfig: c.ReaderConfig}}
+}