[Filebeat] Add timeout to GetObjectRequest for s3 input

CHANGELOG.next.asciidoc

@@ -46,6 +46,7 @@ TLS or Beats that accept connections over TLS and validate client certificates.
 
 *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]
 
 *Heartbeat*

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 Filebeat 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]
+==== `context_timeout`
+
+The maximum duration of GetObjectRequest AWS API can take. If it exceeds the
+timeout, GetObjectRequest will be interrupted.
+The default context 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 GetObject request will be interrupted by context
+    # Default to be 120s
+    #var.context_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 GetObject request will be interrupted by context
+    # Default to be 120s
+    #var.context_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 GetObject request will be interrupted by context
+    # Default to be 120s
+    #var.context_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 GetObject request will be interrupted by context
+    # Default to be 120s
+    #var.context_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"`
+	ContextTimeout            time.Duration       `config:"context_timeout"`
 }
 
 func defaultConfig() config {
@@ -26,6 +27,7 @@ func defaultConfig() config {
 			Type: "s3",
 		},
 		VisibilityTimeout: 300 * time.Second,
+		ContextTimeout:    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.ContextTimeout < 0 || c.ContextTimeout > c.VisibilityTimeout/2 {
+		return fmt.Errorf("context timeout %v needs to be larger than"+
+			" 0s and smaller than half of the visibility timeout", c.ContextTimeout)
+	}
 	return nil
 }