x-pack/filebeat/input/awss3: relax queue_url constraints for testing

Currently, the queue_url option is used to obtain the region name for
the SQS receiver. This prevents pointing the input at other sources for
testing, so add a region_name option to provide an alternative way to
provide the region name for non-AWS endpoints.

To avoid confusion, prevent using the region_name option in conjunction
with amazonaws.com endpoints and document this.

CHANGELOG-developer.next.asciidoc

@@ -153,6 +153,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Add the file path of the instance lock on the error when it's is already locked {pull}33788[33788]
 - Add DropFields processor to js API {pull}33458[33458]
 - Add support for different folders when testing data {pull}34467[34467]
+- Allow non-AWS endpoints for testing Filebeat awss3 input. {issue}35496[35496] {pull}35520[35520]
 
 ==== Deprecated
 

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

@@ -257,6 +257,12 @@ configuring multiline options.
 
 URL of the AWS SQS queue that messages will be received from. (Required when `bucket_arn` and `non_aws_bucket_name` are not set).
 
+[float]
+==== `region_name`
+
+The name of the AWS region of the end point for testing purposes. This option
+must not be set when using amazonaws.com end points.
+
 [float]
 ==== `visibility_timeout`
 

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

@@ -7,6 +7,8 @@ package awss3
 import (
 	"errors"
 	"fmt"
+	"net/url"
+	"strings"
 	"time"
 
 	"github.com/dustin/go-humanize"
@@ -27,6 +29,7 @@ type config struct {
 	SQSScript           *scriptConfig        `config:"sqs.notification_parsing_script"`
 	MaxNumberOfMessages int                  `config:"max_number_of_messages"`
 	QueueURL            string               `config:"queue_url"`
+	RegionName          string               `config:"region_name"`
 	BucketARN           string               `config:"bucket_arn"`
 	NonAWSBucketName    string               `config:"non_aws_bucket_name"`
 	BucketListInterval  time.Duration        `config:"bucket_list_interval"`
@@ -78,6 +81,16 @@ func (c *config) Validate() error {
 		return fmt.Errorf("number_of_workers <%v> must be greater than 0", c.NumberOfWorkers)
 	}
 
+	if c.QueueURL != "" && c.RegionName != "" {
+		u, err := url.Parse(c.QueueURL)
+		if err != nil {
+			return fmt.Errorf("invalid queue_url: %w", err)
+		}
+		if strings.HasSuffix(u.Host, ".amazonaws.com") {
+			return fmt.Errorf("region_name <%s> must not be set with an amazonaws.com queue_url", c.RegionName)
+		}
+	}
+
 	if c.QueueURL != "" && (c.VisibilityTimeout <= 0 || c.VisibilityTimeout.Hours() > 12) {
 		return fmt.Errorf("visibility_timeout <%v> must be greater than 0 and "+
 			"less than or equal to 12h", c.VisibilityTimeout)

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

@@ -115,9 +115,12 @@ func (in *s3Input) Run(inputContext v2.Context, pipeline beat.Pipeline) error {
 	defer cancelInputCtx()
 
 	if in.config.QueueURL != "" {
-		regionName, err := getRegionFromQueueURL(in.config.QueueURL, in.config.AWSConfig.Endpoint)
-		if err != nil {
-			return fmt.Errorf("failed to get AWS region from queue_url: %w", err)
+		regionName := in.config.RegionName
+		if regionName == "" {
+			regionName, err = getRegionFromQueueURL(in.config.QueueURL, in.config.AWSConfig.Endpoint)
+			if err != nil {
+				return fmt.Errorf("failed to get AWS region from queue_url: %w", err)
+			}
 		}
 
 		in.awsConfig.Region = regionName