如何在 FluentD 中拆分 AWS CloudTrail JSON

Question

我有以下 JSON，我需要将名为 Records 的数组拆分为 FluentD 中的不同消息。我想这样做是因为该数组中的所有元素都被提取到 ES 中的同一个文档中，而我只能看到 Kibana 中的第一个元素。

“JSON 是一个单一的散列，键“记录”指向一个散列，其中包含一个数组，其中每个元素代表一个 API 事件。FluentD 现在正在接收这些文件，但只报告一个事件从每个文件中。”

我尝试过但没有成功的事情：

• fluent-plugin-split_record
• fluent-plugin-jq
• fluent-plugin-split-array
• fluent-plugin-split
• parsing-amazon-cloudtrail-jsoon-with-a-customized-logstash-build 使用不同工具阅读类似问题

有人知道如何解决这个问题吗？

{
  "_index": "abcd",
  "_type": "abcd",
  "_id": "abcd",
  "_version": 1,
  "_score": null,
  "_source": {
    "Records": [
      {
        "eventVersion": "1.05",
        "userIdentity": {
          "type": "abcd",
          "principalId": "abcd",
          "accountId": "1234"
        },
        "eventTime": "2020-11-11T09:18:34Z",
        "eventSource": "abcd",
        "eventName": "abcd",
        "awsRegion": "us-east-1",
        "sourceIPAddress": "x.x.x.x",
        "userAgent": "abcd",
        "requestParameters": {
          "roleArn": "abcd",
          "roleSessionName": "abcd",
          "externalId": "1234"
        },
        "responseElements": {
          "credentials": {
            "accessKeyId": "",
            "expiration": "Nov 11, 2020 10:18:34 AM",
            "sessionToken": ""
          },
          "assumedRoleUser": {
            "assumedRoleId": "abcd",
            "arn": "abcd"
          }
        },
        "requestID": "0f34e4e7-0869-44ec-8185-189aa074ff23",
        "eventID": "d205f07f-1f30-4ba1-b99f-3fb929cdb9b7",
        "resources": [
          {
            "accountId": "123",
            "type": "abcd",
            "ARN": "abcd"
          }
        ],
        "eventType": "AwsApiCall",
        "recipientAccountId": "1234",
        "sharedEventID": "d10ccd8d-0489-4e56-9453-e3e3b00915d3"
      },
      {
        "eventVersion": "1.05",
        "userIdentity": {
          "type": "abcd",
          "principalId": "abcd",
          "arn": "abcd",
          "accountId": "1234",
          "accessKeyId": "",
          "sessionContext": {
            "sessionIssuer": {
              "type": "abcd",
              "principalId": "",
              "arn": "abcd",
              "userName": "abcd"
            },
            "webIdFederationData": {},
            "attributes": {
              "mfaAuthenticated": "false",
              "creationDate": "2020-11-11T08:35:54Z"
            }
          }
        },
        "eventTime": "2020-11-11T09:18:17Z",
        "eventSource": "abcd",
        "eventName": "abcd",
        "awsRegion": "us-east-1",
        "sourceIPAddress": "x.x.x.x",
        "userAgent": "abcd",
        "errorCode": "abcd",
        "errorMessage": "abcd",
        "requestParameters": null,
        "responseElements": null,
        "requestID": "cf92658d-c91b-cac4-97b7-cb14cd3db39a",
        "eventID": "dacac287-f47b-4299-94dd-d4b05b47325b",
        "eventType": "abcd",
        "recipientAccountId": "123"
      }
    ],
    "@timestamp": "2020-11-11T09:29:46.023523479+00:00",
    "@log_name": "cloudtrail.logs"
  },
  "fields": {
    "@timestamp": [
      "2020-11-11T09:29:46.023Z"
    ]
  },
  "sort": [
    1234
  ]
}

Answer 1

如果你有同样的问题，答案是下面的插件fluent-plugin-record_splitter

例子：

<match raw.cloudtrail.logs*>
  @type record_splitter
  tag cloudtrail.logs
  split_key Records 
</match>