Cherry-pick #15590 to 7.5: [Filebeat] Add timeout to GetObjectRequest…

… for s3 input (#15908)

* [Filebeat] Add timeout to GetObjectRequest for s3 input (#15590)

* Add timeout to GetObjectRequest which will cancel the request if it takes too long
* Close resp.Body from S3 GetObject API to prevent resource leak
* Change aws_api_timeout to api_timeout

(cherry picked from commit 86c3e63)

* update changelog

* Add default value in manifest.yml

CHANGELOG.next.asciidoc

@@ -51,6 +51,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 - cisco/asa fileset: Fix parsing of 302021 message code. {pull}14519[14519]
 - Fix filebeat azure dashboards, event category should be `Alert`. {pull}14668[14668]
+- Fix s3 input hanging with GetObjectRequest API call by adding context_timeout config. {issue}15502[15502] {pull}15590[15590]
 - Fix typos in zeek notice fileset config file. {issue}15764[15764] {pull}15765[15765]
 - Add shared_credential_file to cloudtrail config {issue}15652[15652] {pull}15656[15656]
 

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

@@ -45,13 +45,21 @@ URL of the AWS SQS queue that messages will be received from. Required.
 [float]
 ==== `visibility_timeout`
 
-The duration (in seconds) that the received messages are hidden from subsequent
+The duration that the received messages are hidden from subsequent
 retrieve requests after being retrieved by a ReceiveMessage request.
-This value needs to be a lot bigger than filebeat collection frequency so
+This value needs to be a lot bigger than {beatname_uc} collection frequency so
 if it took too long to read the s3 log, this sqs message will not be reprocessed.
 The default visibility timeout for a message is 300 seconds. The minimum
 is 0 seconds. The maximum is 12 hours.
 
+[float]
+==== `api_timeout`
+
+The maximum duration of AWS API can take. If it exceeds the timeout, AWS API
+will be interrupted.
+The default AWS API timeout for a message is 120 seconds. The minimum
+is 0 seconds. The maximum is half of the visibility timeout value.
+
 [float]
 ==== `aws credentials`
 

x-pack/filebeat/filebeat.reference.yml

@@ -93,6 +93,14 @@ filebeat.modules:
     # If not set the default profile is used
     #var.credential_profile_name: fb-aws
 
+    # The duration that the received messages are hidden from ReceiveMessage request
+    # Default to be 300s
+    #var.visibility_timeout: 300s
+
+    # Maximum duration before AWS API request will be interrupted
+    # Default to be 120s
+    #var.api_timeout: 120s
+
   elb:
     enabled: false
 
@@ -108,6 +116,14 @@ filebeat.modules:
     # If not set the default profile is used
     #var.credential_profile_name: fb-aws
 
+    # The duration that the received messages are hidden from ReceiveMessage request
+    # Default to be 300s
+    #var.visibility_timeout: 300s
+
+    # Maximum duration before AWS API request will be interrupted
+    # Default to be 120s
+    #var.api_timeout: 120s
+
 #-------------------------------- Azure Module --------------------------------
 - module: azure
   # All logs

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

@@ -17,6 +17,7 @@ type config struct {
 	QueueURL                  string              `config:"queue_url" validate:"nonzero,required"`
 	VisibilityTimeout         time.Duration       `config:"visibility_timeout"`
 	AwsConfig                 awscommon.ConfigAWS `config:",inline"`
+	APITimeout                time.Duration       `config:"api_timeout"`
 }
 
 func defaultConfig() config {
@@ -25,6 +26,7 @@ func defaultConfig() config {
 			Type: "s3",
 		},
 		VisibilityTimeout: 300 * time.Second,
+		APITimeout:        120 * time.Second,
 	}
 }
 
@@ -33,5 +35,9 @@ func (c *config) Validate() error {
 		return fmt.Errorf("visibility timeout %v is not within the "+
 			"required range 0s to 12h", c.VisibilityTimeout)
 	}
+	if c.APITimeout < 0 || c.APITimeout > c.VisibilityTimeout/2 {
+		return fmt.Errorf("api timeout %v needs to be larger than"+
+			" 0s and smaller than half of the visibility timeout", c.APITimeout)
+	}
 	return nil
 }