Add awscloudwatch filebeat input

[kaiyan-sheng]

What does this PR do?

This PR is to add awscloudwatch input into Filebeat. FilterLogEvents AWS API is used to get all log events from a given log group.

The config for using awscloudwatch input looks like below:

filebeat.inputs:
  - type: awscloudwatch
    credential_profile_name: elastic-beats
    log_group_arn: arn:aws:logs:us-east-1:428152502467:log-group:test:*
    region: us-east-1
    scan_frequency: 30s
    start_position: beginning
    api_timeout: 5m

With this config, Filebeat will enable awscloudwatch input to collect all logs from log group test in region us-east-1 (this info is parsed from the log group ARN) starting from the beginning of the log group and then check for new log events every 1-minute(defined by the scan_frequency).

If users only wants to collect new log events/messages from now going forward, then start_position can be specified to be end.

User can also specify a list of log streams under the log group to collect logs from or a log_stream_prefix to collect events only from log streams that have names starting with this prefix.

Sample output document:

{
  "_index": "filebeat-8.0.0-2020.06.26-000001",
  "_type": "_doc",
  "_id": "35535245334778869563151408522177877742806142229743009792",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2020-06-29T18:31:50.000Z",
    "input": {
      "type": "awscloudwatch"
    },
    "ecs": {
      "version": "1.5.0"
    },
    "event": {
      "ingested": "2020-06-29T18:33:21.207Z",
      "id": "35535245334778869563151408522177877742806142229743009792"
    },
    "message": "Apr 23 17:40:01 ip-172-31-81-156 systemd: Removed slice User Slice of root.",
    "log.file.path": "test/test1",
    "agent": {
      "version": "8.0.0",
      "ephemeral_id": "8b7ddd05-a81f-44c3-a0b2-c9a78396b97e",
      "id": "7578d49c-6588-4843-85cc-ad3859f99ed1",
      "name": "KaiyanMacBookPro",
      "type": "filebeat"
    },
    "awscloudwatch": {
      "log_group": "test",
      "log_stream": "test1",
      "ingestion_time": "2020-06-29T18:31:51.000Z"
    },
    "cloud": {
      "provider": "aws",
      "region": "us-east-1"
    }
}

Why is it important?

This input allows user to collect logs from CloudWatch without sending them into S3 bucket with SQS setup for notification.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

1. Go to AWS CloudWatch portal and click Logs https://console.aws.amazon.com/cloudwatch/home?region=us-east-1#logsV2:log-groups
2. Create a new log group and log stream
3. Click Action and Create log event
4. Start Filebeat with filebeat.yml contains awscloudwatch input like below:

filebeat.inputs:
  - type: awscloudwatch
    credential_profile_name: elastic-beats
    log_group_arn: arn:aws:logs:us-east-1:428152502467:log-group:test:*
    region: us-east-1
    scan_frequency: 30s
    start_position: beginning
    api_timeout: 5m

5. You should see logs in Kibana and if you create new log events in AWS CloudWatch, these new logs should be collected after the scan_frequency as well.

Related issues

closes #17292

Screenshots

When setting start_position to beginning, all existing logs will be collected and then Filebeat will scan every 30 seconds(based on the scan_frequency):

[elasticmachine]

Pinging @elastic/integrations-platforms (Team:Platforms)

[elasticmachine]

💚 Build Succeeded

Expand to view the summary

Build stats

• Build Cause: [Pull request #19025 updated]

• Start Time: 2020-07-01T19:45:44.592+0000

• Duration: 35 min 17 sec

Test stats 🧪

Test	Results
Failed	0
Passed	555
Skipped	128
Total	683

[exekias]

Great to see this progressing! I did a first pass, I think we can better leverage nextToken when reading from the API, and avoid using timestamps for filtering

[exekias]

api_sleep and scan_frequency are very similar concepts with very different names here. Would it make sense to unify these a little bit?

[kaiyan-sheng]

scan_frequency defines the sleep time between this Filebeat input recheck for new logs
api_sleep defines the sleep time between each FilterLogEvents API call in the same Filebeat collection cycle.

How about scan_frequency and api_freqency?

[exekias]

I see your point, let's keep api_sleep and make sure this is well documented

[exekias]

I wonder if this can be mapped to event.ingested https://www.elastic.co/guide/en/ecs/current/ecs-event.html

[kaiyan-sheng]

event.ingested in ECS is the timestamp when an event arrived in the central data store. My understanding is this is the timestamp when event gets to Elasticsearch. But this ingestion_time is the time the event was ingested into AWS CloudWatch.

Maybe I understand event.ingested in ECS wrong? 😬

[exekias]

you may be right here, let's keep current naming for now