[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

CHANGELOG.next.asciidoc

@@ -55,6 +55,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Filebeat*
 
+- Fix s3 input hanging with GetObjectRequest API call by adding context_timeout config. {issue}15502[15502] {pull}15590[15590]
 - Add shared_credential_file to cloudtrail config {issue}15652[15652] {pull}15656[15656]
 - Fix typos in zeek notice fileset config file. {issue}15764[15764] {pull}15765[15765]
 

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

@@ -45,9 +45,9 @@ 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.
@@ -61,6 +61,14 @@ can be assigned the name of the field. This setting will be able to split the
 messages under the group value into separate events. For example, CloudTrail logs
 are in JSON format and events are found under the JSON object "Records":
 
+[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.
+
 ["source","json"]
 ----
 {

x-pack/filebeat/filebeat.reference.yml

@@ -111,6 +111,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
 
@@ -126,6 +134,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
+
   vpcflow:
     enabled: false
 
@@ -141,6 +157,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
+
   cloudtrail:
     enabled: false
 
@@ -156,6 +180,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

@@ -18,6 +18,7 @@ type config struct {
 	VisibilityTimeout         time.Duration       `config:"visibility_timeout"`
 	AwsConfig                 awscommon.ConfigAWS `config:",inline"`
 	ExpandEventListFromField  string              `config:"expand_event_list_from_field"`
+	APITimeout                time.Duration       `config:"api_timeout"`
 }
 
 func defaultConfig() config {
@@ -26,6 +27,7 @@ func defaultConfig() config {
 			Type: "s3",
 		},
 		VisibilityTimeout: 300 * time.Second,
+		APITimeout:        120 * time.Second,
 	}
 }
 
@@ -34,5 +36,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
 }