Parse CloudTrail log and remap each event

[omers]

Dear Community,

I am planing to use vector to ingest cloudtrail logs into clickhouse.
The batch of events looks like:

{"Records": [{
    "eventVersion": "1.08",
    "userIdentity": {
        "type": "IAMUser",
        "principalId": "AIDA6ON6E4XEGITEXAMPLE",
        "arn": "arn:aws:iam::888888888888:user/Mary",
        "accountId": "888888888888",
        "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
        "userName": "Mary",
        "sessionContext": {
            "sessionIssuer": {},
            "webIdFederationData": {},
            "attributes": {
                "creationDate": "2023-07-19T21:11:57Z",
                "mfaAuthenticated": "false"
            }
        }
    },
    "eventTime": "2023-07-19T21:25:09Z",
    "eventSource": "iam.amazonaws.com",
    "eventName": "CreateUser",
    "awsRegion": "us-east-1",
    "sourceIPAddress": "192.0.2.0",
    "userAgent": "aws-cli/2.13.5 Python/3.11.4 Linux/4.14.255-314-253.539.amzn2.x86_64 exec-env/CloudShell exe/x86_64.amzn.2 prompt/off command/iam.create-user",
    "requestParameters": {
        "userName": "Richard"
    },
    "responseElements": {
        "user": {
            "path": "/",
            "arn": "arn:aws:iam::888888888888:user/Richard",
            "userId": "AIDA6ON6E4XEP7EXAMPLE",
            "createDate": "Jul 19, 2023 9:25:09 PM",
            "userName": "Richard"
        }
    },
    "requestID": "2d528c76-329e-410b-9516-EXAMPLE565dc",
    "eventID": "ba0801a1-87ec-4d26-be87-EXAMPLE75bbb",
    "readOnly": false,
    "eventType": "AwsApiCall",
    "managementEvent": true,
    "recipientAccountId": "888888888888",
    "eventCategory": "Management",
    "tlsDetails": {
        "tlsVersion": "TLSv1.2",
        "cipherSuite": "ECDHE-RSA-AES128-GCM-SHA256",
        "clientProvidedHostHeader": "iam.amazonaws.com"
    },
    "sessionCredentialFromConsole": "true"
},
{
    "eventVersion": "1.08",
    "userIdentity": {
        "type": "IAMUser",
        "principalId": "AIDA6ON6E4XEGITEXAMPLE",
        "arn": "arn:aws:iam::888888888888:user/Mary",
        "accountId": "888888888888",
        "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
        "userName": "Mary",
        "sessionContext": {
            "sessionIssuer": {},
            "webIdFederationData": {},
            "attributes": {
                "creationDate": "2023-07-19T21:11:57Z",
                "mfaAuthenticated": "false"
            }
        }
    },
    "eventTime": "2023-07-19T21:25:09Z",
    "eventSource": "iam.amazonaws.com",
    "eventName": "CreateUser",
    "awsRegion": "us-east-1",
    "sourceIPAddress": "192.0.2.0",
    "userAgent": "aws-cli/2.13.5 Python/3.11.4 Linux/4.14.255-314-253.539.amzn2.x86_64 exec-env/CloudShell exe/x86_64.amzn.2 prompt/off command/iam.create-user",
    "requestParameters": {
        "userName": "Richard"
    },
    "responseElements": {
        "user": {
            "path": "/",
            "arn": "arn:aws:iam::888888888888:user/Richard",
            "userId": "AIDA6ON6E4XEP7EXAMPLE",
            "createDate": "Jul 19, 2023 9:25:09 PM",
            "userName": "Richard"
        }
    },
    "requestID": "2d528c76-329e-410b-9516-EXAMPLE565dc",
    "eventID": "ba0801a1-87ec-4d26-be87-EXAMPLE75bbb",
    "readOnly": false,
    "eventType": "AwsApiCall",
    "managementEvent": true,
    "recipientAccountId": "888888888888",
    "eventCategory": "Management",
    "tlsDetails": {
        "tlsVersion": "TLSv1.2",
        "cipherSuite": "ECDHE-RSA-AES128-GCM-SHA256",
        "clientProvidedHostHeader": "iam.amazonaws.com"
    },
    "sessionCredentialFromConsole": "true"
}
]}    

My configuration is:

sources:
  cloudtrailsource:
    type: http
    address: 0.0.0.0:8090
    method: POST
    path: /cloudtrail
    encoding: text
    response_code: 200

transforms:
  transformcloudtrail:
    type: remap
    inputs:
      - cloudtrailsource
    source: |
      . = parse_json!(.message)
      .active = true
sinks:
  stdout:
    type: console
    inputs:
      - transformcloudtrail
    encoding:
      codec: json

How can I iterate over each log record, and do transformation for each log event?

[jszwedko]

The easiest way is to use unnest to break up the event into multiple. See https://vector.dev/guides/advanced/cloudwatch-logs-firehose/ for an example of what this looks like.

If you want to leave it as one event and just map over the entries you can use the map_values VRL function.

[omers]

I managed to do it with the following configuration:

api:
  enabled: true
  address: "0.0.0.0:8686"

log_schema:
  timestamp_key: "eventTime"

sources:
  vector_metrics:
    type: "internal_metrics"

  cloudtrailsource:
    type: http_server
    address: 0.0.0.0:8090
    method: POST
    path: /cloudtrail
    strict_path: true
    encoding: text
    response_code: 200

transforms:
  transformcloudtrail:
    type: remap
    inputs:
      - cloudtrailsource
    source: |
      parsed = parse_json!(.message)
      . = unnest!(parsed.Records)
      . = map_values(.) -> |value| {
        event = del(value.Records)
        value |= object!(event)
      }
  mutate:
    type: remap
    inputs:
      - transformcloudtrail
    source: |
      del(.tlsDetails)
      del(.eventVersion)
      del(.requestID)
      del(.eventID)
      del(.readOnly)
      del(.tlsDetails)
      .account = .userIdentity.accountId
      if (.account != "888888888888") {
        .mycompany = false
      }


sinks:
  prometheus:
    type: prometheus_exporter
    inputs:
      - vector_metrics
    address: 0.0.0.0:9598
    flush_period_secs: 60
  stdout:
    type: console
    inputs:
      - mutate
    encoding:
      codec: json