x-pack/filebeat/input/awss3 Add AWS Endpoint resolver (#36208)

* Add AWS Endpoint resolver * Add Signing region to resolver
elastic · Aug 14, 2023 · 4586146 · 4586146
1 parent 5236157
commit 4586146
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 17 deletions.
diff --git a/CHANGELOG.next.asciidoc b/CHANGELOG.next.asciidoc
@@ -232,6 +232,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Add support for localstack based input integration testing {pull}35727[35727]
 - Allow parsing bytes in and bytes out as long integer in CEF processor. {issue}36100[36100] {pull}36108[36108]
 - Add support for registered owners and users to AzureAD entity analytics provider. {pull}36092[36092]
+- Add support for endpoint resolver in AWS config {pull}36208[36208]
 - Added support for Okta OAuth2 provider in the httpjson input. {pull}36273[36273]
 - Add support of the interval parameter in Salesforce setupaudittrail-rest fileset. {issue}35917[35917] {pull}35938[35938]
 - Add device handling to Okta input package for entity analytics. {pull}36049[36049]

diff --git a/x-pack/filebeat/input/awss3/config.go b/x-pack/filebeat/input/awss3/config.go
@@ -100,10 +100,10 @@ func (c *config) Validate() error {
 	}
 
 	if c.AWSConfig.FIPSEnabled && c.NonAWSBucketName != "" {
-		return errors.New("fips_enabled cannot be used with a non-AWS S3 bucket.")
+		return errors.New("fips_enabled cannot be used with a non-AWS S3 bucket")
 	}
-	if c.PathStyle && c.NonAWSBucketName == "" {
-		return errors.New("path_style can only be used when polling non-AWS S3 services")
+	if c.PathStyle && c.NonAWSBucketName == "" && c.QueueURL == "" {
+		return errors.New("path_style can only be used when polling non-AWS S3 services or SQS/SNS QueueURL")
 	}
 	if c.ProviderOverride != "" && c.NonAWSBucketName == "" {
 		return errors.New("provider can only be overridden when polling non-AWS S3 services")

diff --git a/x-pack/filebeat/input/awss3/config_test.go b/x-pack/filebeat/input/awss3/config_test.go
@@ -392,7 +392,7 @@ func TestConfig(t *testing.T) {
 				"number_of_workers":   5,
 				"fips_enabled":        true,
 			},
-			expectedErr: "fips_enabled cannot be used with a non-AWS S3 bucket.",
+			expectedErr: "fips_enabled cannot be used with a non-AWS S3 bucket",
 			expectedCfg: nil,
 		},
 		{
@@ -408,19 +408,6 @@ func TestConfig(t *testing.T) {
 			expectedErr: "path_style can only be used when polling non-AWS S3 services",
 			expectedCfg: nil,
 		},
-		{
-			name:           "error on path_style with AWS SQS Queue",
-			queueURL:       queueURL,
-			s3Bucket:       "",
-			nonAWSS3Bucket: "",
-			config: mapstr.M{
-				"queue_url":         queueURL,
-				"number_of_workers": 5,
-				"path_style":        true,
-			},
-			expectedErr: "path_style can only be used when polling non-AWS S3 services",
-			expectedCfg: nil,
-		},
 		{
 			name:           "error on provider with AWS native S3 Bucket",
 			queueURL:       "",

diff --git a/x-pack/filebeat/input/awss3/input.go b/x-pack/filebeat/input/awss3/input.go
@@ -69,6 +69,18 @@ type s3Input struct {
 
 func newInput(config config, store beater.StateStore) (*s3Input, error) {
 	awsConfig, err := awscommon.InitializeAWSConfig(config.AWSConfig)
+
+	if config.AWSConfig.Endpoint != "" {
+		// Add a custom endpointResolver to the awsConfig so that all the requests are routed to this endpoint
+		awsConfig.EndpointResolverWithOptions = awssdk.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (awssdk.Endpoint, error) {
+			return awssdk.Endpoint{
+				PartitionID:   "aws",
+				URL:           config.AWSConfig.Endpoint,
+				SigningRegion: awsConfig.Region,
+			}, nil
+		})
+	}
+
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize AWS credentials: %w", err)
 	}