Log Architecture & Structural Overview

The AWS VPC Flow Log Parser (Default Format) is an essential telemetry stream within the AWS Cloud Infrastructure ecosystem. Extract IP protocol numbers, packets, bytes, and accept/reject decisions from AWS VPC Flow Logs.

This schema defines a structure of 14 extracted attributes, including 8 numeric metrics and 6 string dimensions. In production observability architectures, these tokens provide high-cardinality indexing keys for telemetry pipelines before shipping to storage backends such as ClickHouse, Elasticsearch, Amazon S3, or Datadog.

Raw Telemetry Ingestion Profile

A typical raw event line for aws-vpc-flow-logs averages 112 bytes across 14 tokens. Modern collectors such as Fluent Bit and Vector require zero-backtracking regular expressions to avoid CPU spikes during traffic surges.

Extracted Field Schema & Data Types

The transpiled Grok pattern extracts the following schema fields from each raw event line. Data collectors cast these values according to the typed mappings below.

Field Name	Inferred Type	Description & Collector Semantics
version	integer	VPC Flow Log version (default 2).
account_id	string	AWS Account ID owning the ENI.
interface_id	string	Elastic Network Interface (ENI) identifier.
src_addr	string	Source IPv4 or IPv6 address.
dst_addr	string	Destination IPv4 or IPv6 address.
src_port	integer	Source transport layer port.
dst_port	integer	Destination transport layer port.
protocol	integer	IANA protocol number (6=TCP, 17=UDP, 1=ICMP).
packets	integer	Number of packets transferred during window.
bytes	integer	Number of bytes transferred during window.
start_time	integer	Unix epoch seconds marking capture window start.
end_time	integer	Unix epoch seconds marking capture window end.
action	string	Security Group / NACL decision (ACCEPT or REJECT).
log_status	string	Logging health status (OK, NODATA, SKIPDATA).

Common Regex Traps & Production Edge Cases

Engineers frequently encounter ingestion failures or pipeline drops due to subtle variations in real-world event logs. Watch out for these verified pitfalls:

1When no traffic occurs during the capture aggregation window, src_addr and dst_addr appear as hyphens with NODATA status.

2Custom format VPC Flow Logs can append TCP flags and traffic path headers; ensure your parser matches your specific ENI format definition.

3Ports for non-TCP/UDP protocols (like ICMP) are set to 0 or -1.

Production Collector Setup & Configurations

Pre-configured parser definitions ready to be dropped into your infrastructure repository.

Fluent Bit (parsers.conf)

Format: regex

# ==============================================================================
# Fluent Bit Parser Configuration (parsers.conf)
# ==============================================================================
[PARSER]
    Name        logpipeline_parser
    Format      regex
    Regex       ^(?<version>(?:[+-]?(?:[0-9]+))) (?<account_id>\S+) (?<interface_id>\S+) (?<src_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<dst_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<src_port>(?:[+-]?(?:[0-9]+))) (?<dst_port>(?:[+-]?(?:[0-9]+))) (?<protocol>(?:[+-]?(?:[0-9]+))) (?<packets>(?:[+-]?(?:[0-9]+))) (?<bytes>(?:[+-]?(?:[0-9]+))) (?<start_time>(?:[+-]?(?:[0-9]+))) (?<end_time>(?:[+-]?(?:[0-9]+))) (?<action>\b\w+\b) (?<log_status>\b\w+\b)$
    Time_Key    timestamp
    Time_Format %Y-%m-%dT%H:%M:%S%z
    Types       version:integer src_port:integer dst_port:integer protocol:integer packets:integer bytes:integer start_time:integer end_time:integer

# ==============================================================================
# Fluent Bit Pipeline Filter (fluent-bit.conf)
# ==============================================================================
[FILTER]
    Name         parser
    Match        *
    Key_Name     log
    Parser       logpipeline_parser
    Reserve_Data On

Vector.dev (Remap VRL)

parse_regex!

# ==============================================================================
# Vector.dev Remap Language (VRL) Transform
# Use inside a 'remap' transform in vector.yaml
# ==============================================================================
.parsed, err = parse_regex(.message, r'^(?<version>(?:[+-]?(?:[0-9]+))) (?<account_id>\S+) (?<interface_id>\S+) (?<src_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<dst_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<src_port>(?:[+-]?(?:[0-9]+))) (?<dst_port>(?:[+-]?(?:[0-9]+))) (?<protocol>(?:[+-]?(?:[0-9]+))) (?<packets>(?:[+-]?(?:[0-9]+))) (?<bytes>(?:[+-]?(?:[0-9]+))) (?<start_time>(?:[+-]?(?:[0-9]+))) (?<end_time>(?:[+-]?(?:[0-9]+))) (?<action>\b\w+\b) (?<log_status>\b\w+\b)$')

if err == null {
    . = merge(., .parsed)
    del(.parsed)

    # Type coercions
    .version = to_int!(.version)
    .src_port = to_int!(.src_port)
    .dst_port = to_int!(.dst_port)
    .protocol = to_int!(.protocol)
    .packets = to_int!(.packets)
    .bytes = to_int!(.bytes)
    .start_time = to_int!(.start_time)
    .end_time = to_int!(.end_time)

} else {
    log("LogPipeline parsing warning: " + err, level: "warn")
}

# ==============================================================================
# vector.yaml Pipeline Component
# ==============================================================================
transforms:
  parse_logs:
    type: remap
    inputs: ["source_logs"]
    source: |
      .parsed, err = parse_regex(.message, r'^(?<version>(?:[+-]?(?:[0-9]+))) (?<account_id>\S+) (?<interface_id>\S+) (?<src_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<dst_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<src_port>(?:[+-]?(?:[0-9]+))) (?<dst_port>(?:[+-]?(?:[0-9]+))) (?<protocol>(?:[+-]?(?:[0-9]+))) (?<packets>(?:[+-]?(?:[0-9]+))) (?<bytes>(?:[+-]?(?:[0-9]+))) (?<start_time>(?:[+-]?(?:[0-9]+))) (?<end_time>(?:[+-]?(?:[0-9]+))) (?<action>\b\w+\b) (?<log_status>\b\w+\b)$')
      if err == null {
        . = merge(., .parsed)
        del(.parsed)
      }

Datadog Log Pipeline Grok Parser

match_rules

# ==============================================================================
# Datadog Log Processing Pipeline Grok Parser
# Navigate to: Logs -> Configuration -> Pipelines -> Add Processor -> Grok Parser
# ==============================================================================

# Match Rule:
rule %{INT:version} %{NOTSPACE:account_id} %{NOTSPACE:interface_id} %{IP:src_addr} %{IP:dst_addr} %{INT:src_port} %{INT:dst_port} %{INT:protocol} %{INT:packets} %{INT:bytes} %{INT:start_time} %{INT:end_time} %{WORD:action} %{WORD:log_status}

# Complete Datadog Pipeline Processor JSON:
{
  "type": "grok-parser",
  "name": "LogPipeline Grok Parser",
  "is_enabled": true,
  "source": "message",
  "samples": [],
  "grok": {
    "match_rules": "rule %{INT:version} %{NOTSPACE:account_id} %{NOTSPACE:interface_id} %{IP:src_addr} %{IP:dst_addr} %{INT:src_port} %{INT:dst_port} %{INT:protocol} %{INT:packets} %{INT:bytes} %{INT:start_time} %{INT:end_time} %{WORD:action} %{WORD:log_status}",
    "support_rules": ""
  }
}

# Target Fields Created:
# version (integer), account_id (string), interface_id (string), src_addr (string), dst_addr (string), src_port (integer), dst_port (integer), protocol (integer), packets (integer), bytes (integer), start_time (integer), end_time (integer), action (string), log_status (string)

OpenTelemetry Collector (transform processor)

regex_parser

# ==============================================================================
# OpenTelemetry Collector Configuration (otel-collector-config.yaml)
# Option 1: Filelog Receiver with regex_parser Operator
# ==============================================================================
receivers:
  filelog:
    include: [ /var/log/**/*.log ]
    start_at: beginning
    operators:
      - type: regex_parser
        id: logpipeline_regex_parser
        regex: '^(?<version>(?:[+-]?(?:[0-9]+))) (?<account_id>\S+) (?<interface_id>\S+) (?<src_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<dst_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<src_port>(?:[+-]?(?:[0-9]+))) (?<dst_port>(?:[+-]?(?:[0-9]+))) (?<protocol>(?:[+-]?(?:[0-9]+))) (?<packets>(?:[+-]?(?:[0-9]+))) (?<bytes>(?:[+-]?(?:[0-9]+))) (?<start_time>(?:[+-]?(?:[0-9]+))) (?<end_time>(?:[+-]?(?:[0-9]+))) (?<action>\b\w+\b) (?<log_status>\b\w+\b)$'
        timestamp:
          parse_from: attributes.timestamp
          layout: '%Y-%m-%dT%H:%M:%S%z'

# ==============================================================================
# Option 2: Transform Processor (OTel Transformation Language - OTTL)
# ==============================================================================
processors:
  transform:
    error_mode: ignore
    log_statements:
      - context: log
        statements:
          - merge_maps(attributes, extract_patterns(body, "^(?<version>(?:[+-]?(?:[0-9]+))) (?<account_id>\\S+) (?<interface_id>\\S+) (?<src_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<dst_addr>(?:(?:(?:(?:[0-9A-Fa-f]{1,4}:){7}[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,7}:|(?:[0-9A-Fa-f]{1,4}:){1,6}:[0-9A-Fa-f]{1,4}|(?:[0-9A-Fa-f]{1,4}:){1,5}(?::[0-9A-Fa-f]{1,4}){1,2}|(?:[0-9A-Fa-f]{1,4}:){1,4}(?::[0-9A-Fa-f]{1,4}){1,3}|(?:[0-9A-Fa-f]{1,4}:){1,3}(?::[0-9A-Fa-f]{1,4}){1,4}|(?:[0-9A-Fa-f]{1,4}:){1,2}(?::[0-9A-Fa-f]{1,4}){1,5}|[0-9A-Fa-f]{1,4}:(?:(?::[0-9A-Fa-f]{1,4}){1,6})|:(?:(?::[0-9A-Fa-f]{1,4}){1,7}|:)|fe80:(?::[0-9A-Fa-f]{0,4}){0,4}%[0-9a-zA-Z]+|::(?:ffff(?::0{1,4})?:)?(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])|(?:[0-9A-Fa-f]{1,4}:){1,4}:(?:(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])\\.){3}(?:25[0-5]|(?:2[0-4]|1?[0-9])?[0-9])))|(?:(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)))) (?<src_port>(?:[+-]?(?:[0-9]+))) (?<dst_port>(?:[+-]?(?:[0-9]+))) (?<protocol>(?:[+-]?(?:[0-9]+))) (?<packets>(?:[+-]?(?:[0-9]+))) (?<bytes>(?:[+-]?(?:[0-9]+))) (?<start_time>(?:[+-]?(?:[0-9]+))) (?<end_time>(?:[+-]?(?:[0-9]+))) (?<action>\\b\\w+\\b) (?<log_status>\\b\\w+\\b)$"), "insert")

service:
  pipelines:
    logs:
      receivers: [filelog]
      processors: [transform]
      exporters: [otlp]

Logstash Filter Configuration

filter.grok

# ==============================================================================
# Logstash Pipeline Configuration (/etc/logstash/conf.d/logpipeline.conf)
# ==============================================================================
filter {
  grok {
    match => { "message" => "%{INT:version:integer} %{NOTSPACE:account_id} %{NOTSPACE:interface_id} %{IP:src_addr} %{IP:dst_addr} %{INT:src_port:integer} %{INT:dst_port:integer} %{INT:protocol:integer} %{INT:packets:integer} %{INT:bytes:integer} %{INT:start_time:integer} %{INT:end_time:integer} %{WORD:action} %{WORD:log_status}" }
    tag_on_failure => [ "_grokparsefailure" ]
  }

  date {
    match => [ "timestamp", "ISO8601", "dd/MMM/yyyy:HH:mm:ss Z" ]
    target => "@timestamp"
    remove_field => [ "timestamp" ]
  }
}