Skip to content

Commit

Permalink
x-pack/filebeat/input/awss3: relax queue_url constraints for testing
Browse files Browse the repository at this point in the history
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.
  • Loading branch information
efd6 committed May 22, 2023
1 parent f8f25ad commit ea9a2c6
Show file tree
Hide file tree
Showing 4 changed files with 26 additions and 3 deletions.
1 change: 1 addition & 0 deletions CHANGELOG-developer.next.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
6 changes: 6 additions & 0 deletions x-pack/filebeat/docs/inputs/input-aws-s3.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -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`

Expand Down
13 changes: 13 additions & 0 deletions x-pack/filebeat/input/awss3/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ package awss3
import (
"errors"
"fmt"
"net/url"
"strings"
"time"

"github.com/dustin/go-humanize"
Expand All @@ -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"`
Expand Down Expand Up @@ -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)
Expand Down
9 changes: 6 additions & 3 deletions x-pack/filebeat/input/awss3/input.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down

0 comments on commit ea9a2c6

Please sign in to comment.