[Filebeat] Add basic json support into s3 input

[kaiyan-sheng]

This PR is to add basic json support for s3 input so the work of adding cloudtrail metricset can continue. expand_event_list_from_field is added in s3 input config for users to specify the top level key name in the JSON object. For example, in cloudtrail log example below, Records should be the expand_event_list_from_field in order for s3 input to parse the log correctly.

This PR is only for adding expand_event_list_from_field to implement #15357

cloudtrail log example:

{"Records": [{
    "eventVersion": "1.0",
    "userIdentity": {
        "type": "IAMUser",
        "principalId": "EX_PRINCIPAL_ID",
        "arn": "arn:aws:iam::123456789012:user/Alice",
        "accessKeyId": "EXAMPLE_KEY_ID",
        "accountId": "123456789012",
        "userName": "Alice"
    },
    "eventTime": "2014-03-06T21:22:54Z",
    "eventSource": "ec2.amazonaws.com",
    "eventName": "StartInstances",
    "awsRegion": "us-east-2",
    "sourceIPAddress": "205.251.233.176",
    "userAgent": "ec2-api-tools 1.6.12.2",
    "requestParameters": {"instancesSet": {"items": [{"instanceId": "i-ebeaf9e2"}]}},
    "responseElements": {"instancesSet": {"items": [{
        "instanceId": "i-ebeaf9e2",
        "currentState": {
            "code": 0,
            "name": "pending"
        },
        "previousState": {
            "code": 80,
            "name": "stopped"
        }
    }]}}
}]}

How to test this

In order to get json log files into S3 bucket, you can manually upload json file or you can enable cloudtrail log in AWS to send logs to a specific S3 bucket.
Also you need to create an SQS queue and setup notifications for any new object created in S3 bucket.
Run filebeat with s3 input enabled in filebeat.yml:

filebeat.inputs:
- type: s3
  queue_url:   https://sqs.us-east-1.amazonaws.com/428152502467/test-fb-ks
  credential_profile_name: elastic-beats
  expand_event_list_from_field: Records

This will enable Filebeat s3 input to parse json logs.

Output

{
  "_index": "filebeat-8.0.0-2019.12.23-000001",
  "_type": "_doc",
  "_id": "52a0364d21-000000001506",
  "_version": 1,
  "_score": null,
  "_source": {
    "@timestamp": "2020-01-07T23:31:18.846Z",
    "cloud": {
      "region": "us-east-1",
      "provider": "aws"
    },
    "agent": {
      "hostname": "KaiyanMacBookPro",
      "id": "7578d49c-6588-4843-85cc-ad3859f99ed1",
      "version": "8.0.0",
      "type": "filebeat",
      "ephemeral_id": "16bf080d-b115-46b4-8caa-c272fb7f4f88"
    },
    "ecs": {
      "version": "1.2.0"
    },
    "container": {
      "id": "CloudTrail"
    },
    "message": "{\"additionalEventData\":{\"AuthenticationMethod\":\"AuthHeader\",\"CipherSuite\":\"ECDHE-RSA-AES128-SHA\",\"SSEApplied\":\"SSE_S3\",\"SignatureVersion\":\"SigV4\",\"bytesTransferredIn\":712,\"bytesTransferredOut\":0,\"x-amz-id-2\":\"9n9oPkU2VnSa3NBi095vtVm0yQeTtVdCLbUlmjN9W5vYuyhn/FyLlNzhga3YEtwz7qQiEBcYoWs=\"},\"awsRegion\":\"us-east-1\",\"eventID\":\"dd7d6dd1-e0fb-4dcc-959a-394980524786\",\"eventName\":\"PutObject\",\"eventSource\":\"s3.amazonaws.com\",\"eventTime\":\"2020-01-07T23:18:03Z\",\"eventType\":\"AwsApiCall\",\"eventVersion\":\"1.05\",\"readOnly\":false,\"recipientAccountId\":\"428152502467\",\"requestID\":\"4D8B8F6053BDBA1E\",\"requestParameters\":{\"Host\":\"test-fb-ks.s3.amazonaws.com\",\"bucketName\":\"test-fb-ks\",\"key\":\"AWSLogs/428152502467/CloudTrail-Digest/ap-south-1/2020/01/07/428152502467_CloudTrail-Digest_ap-south-1_test-cloudtrail-ks_us-east-1_20200107T222151Z.json.gz\",\"x-amz-acl\":\"bucket-owner-full-control\",\"x-amz-server-side-encryption\":\"AES256\"},\"resources\":[{\"ARN\":\"arn:aws:s3:::test-fb-ks/AWSLogs/428152502467/CloudTrail-Digest/ap-south-1/2020/01/07/428152502467_CloudTrail-Digest_ap-south-1_test-cloudtrail-ks_us-east-1_20200107T222151Z.json.gz\",\"type\":\"AWS::S3::Object\"},{\"ARN\":\"arn:aws:s3:::test-fb-ks\",\"accountId\":\"428152502467\",\"type\":\"AWS::S3::Bucket\"}],\"responseElements\":{\"x-amz-server-side-encryption\":\"AES256\"},\"sharedEventID\":\"370d7d2e-10a1-4a29-994b-c51add403ad2\",\"sourceIPAddress\":\"cloudtrail.amazonaws.com\",\"userAgent\":\"cloudtrail.amazonaws.com\",\"userIdentity\":{\"invokedBy\":\"cloudtrail.amazonaws.com\",\"type\":\"AWSService\"}}",
    "log": {
      "offset": 1506,
      "file.path": "https://test-fb-ks.s3-us-east-1.amazonaws.com/AWSLogs/428152502467/CloudTrail/us-east-1/2020/01/07/428152502467_CloudTrail_us-east-1_20200107T2320Z_KRw46eDNsf54qTpu.json.gz"
    },
    "aws": {
      "s3": {
        "object.key": "AWSLogs/428152502467/CloudTrail/us-east-1/2020/01/07/428152502467_CloudTrail_us-east-1_20200107T2320Z_KRw46eDNsf54qTpu.json.gz",
        "bucket": {
          "name": "test-fb-ks",
          "arn": "arn:aws:s3:::test-fb-ks"
        }
      }
    },
    "input": {
      "type": "s3"
    },
    "host": {
      "name": "KaiyanMacBookPro",
      "hostname": "KaiyanMacBookPro",
      "architecture": "x86_64",
      "os": {
        "build": "17G10021",
        "platform": "darwin",
        "version": "10.13.6",
        "family": "darwin",
        "name": "Mac OS X",
        "kernel": "17.7.0"
      },
      "id": "9C7FAB7B-29D1-5926-8E84-158A9CA3E25D"
    }
  },
  "fields": {
    "suricata.eve.timestamp": [
      "2020-01-07T23:31:18.846Z"
    ],
    "@timestamp": [
      "2020-01-07T23:31:18.846Z"
    ]
  },
  "sort": [
    1578439878846
  ]
}

[andrewkroh]

We need a way to be able to tell it to iterate over nested objects and create one document per object. Like if the message_key were Records[] then it would iterate over each element in Records and generate an event. (That syntax I borrowed from http://jmespath.org/).

[leehinman]

I tested with cloudtrail and it worked for both single event in Records array and multiple events in Records array. :-)

I did see a panic if json.message_key: Records wasn't included with Cloudtrail S3 input.

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x571a5e8]

goroutine 259 [running]:
github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).handleS3Objects(0xc000118000, 0x618c820, 0xc0007f6010, 0xc0007e5e80, 0x1, 0x1, 0xc00005a420, 0x0, 0x0)
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:384 +0xa98
github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).processMessage(0xc000118000, 0x618c820, 0xc0007f6010, 0x0, 0xc000946020, 0xc000946050, 0x0, 0x0, 0xc000946080, 0xc0009460c0, ...)
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:257 +0x2f6
created by github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).processor
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:240 +0x19c

[kaiyan-sheng]

We need a way to be able to tell it to iterate over nested objects and create one document per object. Like if the message_key were Records[] then it would iterate over each element in Records and generate an event. (That syntax I borrowed from http://jmespath.org/).

@andrewkroh Thanks for the review. With this PR, if json.message_key is given, then it will iterate over each element and generate one event per element under the message key, such as Records. Is there any use case that when we specify a json.message_key but still only want to report one event for all elements under the key?

[kaiyan-sheng]

I tested with cloudtrail and it worked for both single event in Records array and multiple events in Records array. :-)

I did see a panic if json.message_key: Records wasn't included with Cloudtrail S3 input.

panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x8 pc=0x571a5e8]

goroutine 259 [running]:
github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).handleS3Objects(0xc000118000, 0x618c820, 0xc0007f6010, 0xc0007e5e80, 0x1, 0x1, 0xc00005a420, 0x0, 0x0)
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:384 +0xa98
github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).processMessage(0xc000118000, 0x618c820, 0xc0007f6010, 0x0, 0xc000946020, 0xc000946050, 0x0, 0x0, 0xc000946080, 0xc0009460c0, ...)
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:257 +0x2f6
created by github.com/elastic/beats/x-pack/filebeat/input/s3.(*s3Input).processor
	/go/src/github.com/elastic/beats/x-pack/filebeat/input/s3/input.go:240 +0x19c

@leehinman Thanks for testing it! Panic bug is fixed now!

[kaiyan-sheng]

I dont think CI failures are related. Merging this PR.