Cherry-pick #19025 to 7.x: Add awscloudwatch filebeat input (#19570)

* Add awscloudwatch filebeat input (#19025)

* Add awscloudwatch filebeat input
* Use log group ARN instead of log group name and region name
* add api_sleep, log_group_name and region_name config

(cherry picked from commit 07639fe)

CHANGELOG.next.asciidoc

@@ -496,6 +496,7 @@ field. You can revert this change by configuring tags for the module and omittin
 - Add support for v1 consumer API in Cloud Foundry input, use it by default. {pull}19125[19125]
 - Add new mode to multiline reader to aggregate constant number of lines {pull}18352[18352]
 - Explicitly set ECS version in all Filebeat modules. {pull}19198[19198]
+- Add awscloudwatch input. {pull}19025[19025]
 - Add support for timezone offsets and `Z` to decode_cef timestamp parser. {pull}19346[19346]
 
 *Heartbeat*

filebeat/docs/fields.asciidoc

@@ -16,6 +16,7 @@ grouped in the following categories:
 * <<exported-fields-apache>>
 * <<exported-fields-auditd>>
 * <<exported-fields-aws>>
+* <<exported-fields-awscloudwatch>>
 * <<exported-fields-azure>>
 * <<exported-fields-beat-common>>
 * <<exported-fields-cef>>
@@ -2040,6 +2041,47 @@ type: keyword
 The type of traffic: IPv4, IPv6, or EFA.
 
 
+type: keyword
+
+--
+
+[[exported-fields-awscloudwatch]]
+== awscloudwatch fields
+
+Fields from AWS CloudWatch logs.
+
+
+
+[float]
+=== awscloudwatch
+
+Fields from AWS CloudWatch logs.
+
+
+
+*`awscloudwatch.log_group`*::
++
+--
+The name of the log group to which this event belongs.
+
+type: keyword
+
+--
+
+*`awscloudwatch.log_stream`*::
++
+--
+The name of the log stream to which this event belongs.
+
+type: keyword
+
+--
+
+*`awscloudwatch.ingestion_time`*::
++
+--
+The time the event was ingested in AWS CloudWatch.
+
 type: keyword
 
 --

x-pack/filebeat/docs/inputs/input-awscloudwatch.asciidoc

@@ -0,0 +1,119 @@
+[role="xpack"]
+
+:libbeat-xpack-dir: ../../../../x-pack/libbeat
+
+:type: awscloudwatch
+
+[id="{beatname_lc}-input-{type}"]
+=== awscloudwatch input
+
+++++
+<titleabbrev>awscloudwatch</titleabbrev>
+++++
+
+beta[]
+
+`awscloudwatch` input can be used to retrieve all logs from all log streams in a
+specific log group. `filterLogEvents` AWS API is used to list log events from
+the specified log group. Amazon CloudWatch Logs can be used to store log files
+from Amazon Elastic Compute Cloud(EC2), AWS CloudTrail, Route53, and other sources.
+
+A log group is a group of log streams that share the same retention, monitoring,
+and access control settings. You can define log groups and specify which streams
+to put into each group. There is no limit on the number of log streams that can
+belong to one log group.
+
+A log stream is a sequence of log events that share the same source. Each
+separate source of logs in CloudWatch Logs makes up a separate log stream.
+
+["source","yaml",subs="attributes"]
+----
+{beatname_lc}.inputs:
+- type: awscloudwatch
+  log_group_arn: arn:aws:logs:us-east-1:428152502467:log-group:test:*
+  scan_frequency: 1m
+  credential_profile_name: elastic-beats
+  start_position: beginning
+----
+
+The `awscloudwatch` input supports the following configuration options plus the
+<<{beatname_lc}-input-{type}-common-options>> described later.
+
+[float]
+==== `log_group_arn`
+ARN of the log group to collect logs from.
+
+==== `log_group_name`
+Name of the log group to collect logs from. Note: region_name is required when
+log_group_name is given.
+
+==== `region_name`
+Region that the specified log group belongs to.
+
+[float]
+==== `log_streams`
+A list of strings of log streams names that Filebeat collect log events from.
+
+[float]
+==== `log_stream_prefix`
+A string to filter the results to include only log events from log streams
+that have names starting with this prefix.
+
+[float]
+==== `start_position`
+`start_position` allows user to specify if this input should read log files from
+the `beginning` or from the `end`.
+
+* `beginning`: reads from the beginning of the log group (default).
+* `end`: read only new messages from current time minus `scan_frequency` going forward
+
+For example, with `scan_frequency` equals to `30s` and current timestamp is
+`2020-06-24 12:00:00`:
+
+* with `start_position = beginning`:
+** first iteration: startTime=0, endTime=2020-06-24 12:00:00
+** second iteration: startTime=2020-06-24 12:00:00, endTime=2020-06-24 12:00:30
+
+* with `start_position = end`:
+** first iteration: startTime=2020-06-24 11:59:30, endTime=2020-06-24 12:00:00
+** second iteration: startTime=2020-06-24 12:00:00, endTime=2020-06-24 12:00:30
+
+[float]
+==== `scan_frequency`
+This config parameter sets how often Filebeat checks for new log events from the
+specified log group. Default `scan_frequency` is 1 minute, which means Filebeat
+will sleep for 1 minute before querying for new logs again.
+
+[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]
+==== `api_sleep`
+This is used to sleep between AWS `FilterLogEvents` API calls inside the same
+collection period. `FilterLogEvents` API has a quota of 5 transactions per
+second (TPS)/account/Region. By default, `api_sleep` is 200 ms. This value should
+only be adjusted when there are multiple Filebeats or multiple Filebeat inputs
+collecting logs from the same region and AWS account.
+
+[float]
+==== `aws credentials`
+In order to make AWS API calls, `awscloudwatch` input requires AWS credentials.
+Please see <<aws-credentials-config,AWS credentials options>> for more details.
+
+[float]
+=== AWS Permissions
+Specific AWS permissions are required for IAM user to access awscloudwatch:
+----
+logs:FilterLogEvents
+----
+
+[id="{beatname_lc}-input-{type}-common-options"]
+include::../../../../filebeat/docs/inputs/input-common-options.asciidoc[]
+
+[id="aws-credentials-config"]
+include::{libbeat-xpack-dir}/docs/aws-credentials-config.asciidoc[]
+
+:type!:

x-pack/filebeat/input/awscloudwatch/_meta/fields.yml

@@ -0,0 +1,20 @@
+- key: awscloudwatch
+  title: "awscloudwatch"
+  description: >
+    Fields from AWS CloudWatch logs.
+  fields:
+    - name: awscloudwatch
+      type: group
+      default_field: false
+      description: >
+        Fields from AWS CloudWatch logs.
+      fields:
+        - name: log_group
+          type: keyword
+          description: The name of the log group to which this event belongs.
+        - name: log_stream
+          type: keyword
+          description: The name of the log stream to which this event belongs.
+        - name: ingestion_time
+          type: keyword
+          description: The time the event was ingested in AWS CloudWatch.

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

@@ -0,0 +1,57 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package awscloudwatch
+
+import (
+	"errors"
+	"time"
+
+	"github.com/elastic/beats/v7/filebeat/harvester"
+	awscommon "github.com/elastic/beats/v7/x-pack/libbeat/common/aws"
+)
+
+type config struct {
+	harvester.ForwarderConfig `config:",inline"`
+	LogGroupARN               string              `config:"log_group_arn"`
+	LogGroupName              string              `config:"log_group_name"`
+	RegionName                string              `config:"region_name"`
+	LogStreams                []string            `config:"log_streams"`
+	LogStreamPrefix           string              `config:"log_stream_prefix"`
+	StartPosition             string              `config:"start_position" default:"beginning"`
+	ScanFrequency             time.Duration       `config:"scan_frequency" validate:"min=0,nonzero"`
+	APITimeout                time.Duration       `config:"api_timeout" validate:"min=0,nonzero"`
+	APISleep                  time.Duration       `config:"api_sleep" validate:"min=0,nonzero"`
+	AwsConfig                 awscommon.ConfigAWS `config:",inline"`
+}
+
+func defaultConfig() config {
+	return config{
+		ForwarderConfig: harvester.ForwarderConfig{
+			Type: "awscloudwatch",
+		},
+		StartPosition: "beginning",
+		ScanFrequency: 10 * time.Second,
+		APITimeout:    120 * time.Second,
+		APISleep:      200 * time.Millisecond, // FilterLogEvents has a limit of 5 transactions per second (TPS)/account/Region: 1s / 5 = 200 ms
+	}
+}
+
+func (c *config) Validate() error {
+	if c.StartPosition != "beginning" && c.StartPosition != "end" {
+		return errors.New("start_position config parameter can only be " +
+			"either 'beginning' or 'end'")
+	}
+
+	if c.LogGroupARN == "" && c.LogGroupName == "" {
+		return errors.New("log_group_arn and log_group_name config parameter" +
+			"cannot be both empty")
+	}
+
+	if c.LogGroupName != "" && c.RegionName == "" {
+		return errors.New("region_name is required when log_group_name " +
+			"config parameter is given")
+	}
+	return nil
+}