diff --git a/x-pack/filebeat/_meta/config/filebeat.inputs.reference.xpack.yml.tmpl b/x-pack/filebeat/_meta/config/filebeat.inputs.reference.xpack.yml.tmpl index 16964b2c84e..f083b4c814b 100644 --- a/x-pack/filebeat/_meta/config/filebeat.inputs.reference.xpack.yml.tmpl +++ b/x-pack/filebeat/_meta/config/filebeat.inputs.reference.xpack.yml.tmpl @@ -67,8 +67,8 @@ #session_token: '${AWS_SESSION_TOKEN:"”}' #credential_profile_name: test-s3-input - # Queue urls (required) to receive queue messages from - #queue_urls: ["https://sqs.us-east-1.amazonaws.com/1234/test-s3-logs-queue"] + # Queue url (required) to receive queue messages from + #queue_url: "https://sqs.us-east-1.amazonaws.com/1234/test-s3-logs-queue" # The duration (in seconds) that the received messages are hidden from subsequent # retrieve requests after being retrieved by a ReceiveMessage request. diff --git a/x-pack/filebeat/filebeat.reference.yml b/x-pack/filebeat/filebeat.reference.yml index 4973a0ae915..120516dc9c4 100644 --- a/x-pack/filebeat/filebeat.reference.yml +++ b/x-pack/filebeat/filebeat.reference.yml @@ -2374,8 +2374,8 @@ filebeat.inputs: #session_token: '${AWS_SESSION_TOKEN:"”}' #credential_profile_name: test-s3-input - # Queue urls (required) to receive queue messages from - #queue_urls: ["https://sqs.us-east-1.amazonaws.com/1234/test-s3-logs-queue"] + # Queue url (required) to receive queue messages from + #queue_url: "https://sqs.us-east-1.amazonaws.com/1234/test-s3-logs-queue" # The duration (in seconds) that the received messages are hidden from subsequent # retrieve requests after being retrieved by a ReceiveMessage request. diff --git a/x-pack/filebeat/input/s3/collector.go b/x-pack/filebeat/input/s3/collector.go index 73e12acfeab..b5d96b05fda 100644 --- a/x-pack/filebeat/input/s3/collector.go +++ b/x-pack/filebeat/input/s3/collector.go @@ -159,8 +159,10 @@ func (c *s3Collector) processorKeepAlive(svcSQS sqsiface.ClientAPI, message sqs. for { select { case <-c.cancellation.Done(): + fmt.Println("------- c.cancellation.Done()") return nil case err := <-errC: + fmt.Println("------- err = ", err) if err != nil { if err == context.DeadlineExceeded { c.logger.Info("Context deadline exceeded, updating visibility timeout") diff --git a/x-pack/filebeat/input/s3/input.go b/x-pack/filebeat/input/s3/input.go index d76e5b8b728..9c31ca7879c 100644 --- a/x-pack/filebeat/input/s3/input.go +++ b/x-pack/filebeat/input/s3/input.go @@ -5,6 +5,7 @@ package s3 import ( + "context" "fmt" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -67,7 +68,12 @@ func (in *s3Input) Run(ctx v2.Context, pipeline beat.Pipeline) error { defer collector.publisher.Close() collector.run() - return ctx.Cancelation.Err() + + if ctx.Cancelation.Err() == context.Canceled { + return nil + } else { + return ctx.Cancelation.Err() + } } func (in *s3Input) createCollector(ctx v2.Context, pipeline beat.Pipeline) (*s3Collector, error) {