---
description: Pipeline for AWS VPC Flow Logs

processors:
  - drop:
      if: 'ctx.message.startsWith("version") || ctx.message.startsWith("instance-id")'
  - set:
      field: event.ingested
      value: '{{_ingest.timestamp}}'
  - set:
      field: ecs.version
      value: '8.0.0'
  - rename:
      field: message
      target_field: event.original
      ignore_missing: true
  - set:
      field: event.kind
      value: event
  - set:
      field: event.category
      value: [network]
  - set:
      field: cloud.provider
      value: aws

  # parse_aws_vpc_flow_log removes the original values to avoid duplication
  # but to avoid a breaking change continue to populate these fields.
  - set:
      copy_from: cloud.account.id
      field: aws.vpcflow.account_id
      ignore_empty_value: true
  - set:
      copy_from: cloud.instance.id
      field: aws.vpcflow.instance_id
      ignore_empty_value: true
  - uppercase:
      field: event.action
      target_field: aws.vpcflow.action
      ignore_missing: true

  # IP Geolocation Lookup
  - geoip:
      field: source.ip
      target_field: source.geo
      ignore_missing: true
  - geoip:
      field: destination.ip
      target_field: destination.geo
      ignore_missing: true

  # IP Autonomous System (AS) Lookup
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: source.ip
      target_field: source.as
      properties:
        - asn
        - organization_name
      ignore_missing: true
  - geoip:
      database_file: GeoLite2-ASN.mmdb
      field: destination.ip
      target_field: destination.as
      properties:
        - asn
        - organization_name
      ignore_missing: true
  - rename:
      field: source.as.asn
      target_field: source.as.number
      ignore_missing: true
  - rename:
      field: source.as.organization_name
      target_field: source.as.organization.name
      ignore_missing: true
  - rename:
      field: destination.as.asn
      target_field: destination.as.number
      ignore_missing: true
  - rename:
      field: destination.as.organization_name
      target_field: destination.as.organization.name
      ignore_missing: true

  - remove:
      field: event.original
      if: "ctx?.tags == null || !(ctx.tags.contains('preserve_original_event'))"
      ignore_failure: true
      ignore_missing: true
on_failure:
  - set:
      field: 'error.message'
      value: '{{{ _ingest.on_failure_message }}}'