AWS VPC Flow Logs: capture and analyze with Athena

AWS VPC Flow Logs provide granular visibility into network traffic within your Virtual Private Cloud. They record metadata about IP traffic flowing to and…

AWS VPC Flow Logs provide granular visibility into network traffic within your Virtual Private Cloud. They record metadata about IP traffic flowing to and from network interfaces, acting as a crucial diagnostic and security tool. This data is invaluable for troubleshooting connectivity issues, identifying potential security breaches, understanding network dependencies, and ensuring compliance. While powerful, efficiently querying and analyzing large volumes of flow log data requires a robust strategy, and integrating with Amazon S3 and Amazon Athena offers a cost-effective and scalable solution.

Enabling VPC Flow Logs to S3

To begin capturing flow logs, you need to enable them on your VPC, a specific subnet, or even an individual Elastic Network Interface (ENI). Directing logs to S3 is generally preferred over CloudWatch Logs for long-term storage and analytical workloads due to cost efficiency and integration capabilities with services like Athena. When configuring, specify the destination S3 bucket and the log record format.

The following AWS CLI command demonstrates how to enable flow logs for an entire VPC, sending all traffic (ALL) to an S3 bucket named acme-flowlogs within a prod/ prefix. We use a custom log format to ensure all necessary fields are captured for later analysis.

aws ec2 create-flow-logs \
  --resource-type VPC \
  --resource-ids vpc-0abc1234567890def \
  --traffic-type ALL \
  --log-destination-type s3 \
  --log-destination arn:aws:s3:::acme-flowlogs/prod/ \
  --log-format '${account-id} ${vpc-id} ${interface-id} ${srcaddr} ${dstaddr} ${srcport} ${dstport} ${protocol} ${packets} ${bytes} ${start} ${end} ${action} ${log-status}' \
  --max-aggregation-interval 60

Key parameters:

--resource-type VPC: Specifies that logs should be captured for the entire VPC. Other options include Subnet and NetworkInterface.
--resource-ids vpc-0abc1234567890def: The ID of the VPC to monitor.
--traffic-type ALL: Captures both accepted and rejected traffic. Other options are ACCEPT and REJECT.
--log-destination-type s3: Designates Amazon S3 as the log destination.
--log-destination arn:aws:s3:::acme-flowlogs/prod/: The ARN of the S3 bucket and prefix where logs will be stored. AWS will automatically create a folder structure like s3://acme-flowlogs/prod/AWSLogs/<account-id>/vpcflowlogs/<region>/<year>/<month>/<day>/.
--log-format: Crucial for defining the data fields in each log record. The example format includes essential fields like interface-id, packets, and log-status in addition to the standard ones. Ensure this matches the Athena table definition.
--max-aggregation-interval 60: Flow logs are aggregated over 60 seconds (default is 600 seconds). A shorter interval provides more granular data but increases log volume.

Configuring Athena for Flow Log Analysis

Once flow logs are streaming to S3, you need to define an Athena table that maps to the S3 data. This allows you to query the raw log files using standard SQL.

Creating the External Table

The Athena table should mirror the fields defined in your --log-format. Since flow logs are written as space-separated values, we use ROW FORMAT DELIMITED FIELDS TERMINATED BY ' '. Partitioning by date (dt string) is critical for performance and cost optimization, as Athena can prune partitions and only scan relevant data.

CREATE EXTERNAL TABLE IF NOT EXISTS vpc_flow_logs (
  account_id STRING,
  vpc_id STRING,
  interface_id STRING,
  srcaddr STRING,
  dstaddr STRING,
  srcport INT,
  dstport INT,
  protocol INT,
  packets BIGINT,
  bytes BIGINT,
  start_time BIGINT,
  end_time BIGINT,
  action STRING,
  log_status STRING
)
PARTITIONED BY (
  dt STRING COMMENT 'Format: YYYY/MM/DD'
)
ROW FORMAT DELIMITED
  FIELDS TERMINATED BY ' '
  ESCAPED BY '\\'
  LINES TERMINATED BY '\n'
LOCATION 's3://acme-flowlogs/prod/AWSLogs/<account-id>/vpcflowlogs/<region>/'
TBLPROPERTIES (
  "skip.header.line.count"="0",
  "external.table.purge.data"="TRUE",
  "projection.enabled" = "true",
  "projection.dt.type" = "date",
  "projection.dt.range" = "2023/01/01,NOW",
  "projection.dt.format" = "yyyy/MM/dd",
  "projection.dt.interval" = "1",
  "projection.dt.interval.unit" = "DAYS",
  "storage.location.template" = "s3://acme-flowlogs/prod/AWSLogs/<account-id>/vpcflowlogs/<region>/${dt}"
);

Explanation of key table properties:

LOCATION: Points to the root S3 prefix where your flow logs are stored. Note the absence of the year/month/day subfolders here, as partitioning handles that. Replace <account-id> and <region> with your actual values.
TBLPROPERTIES ("skip.header.line.count"="0"): VPC Flow Logs do not include a header line, so this should be set to 0.
"projection.enabled" = "true": Enables Partition Projection, which automatically infers new partitions as data arrives, eliminating the need for manual ALTER TABLE ADD PARTITION statements. This is highly recommended for frequently updated datasets like flow logs.
"projection.dt.type" = "date", "projection.dt.range", etc.: These properties configure partition projection for the dt column, allowing Athena to automatically scan for daily partitions within the specified range. Adjust the start date (2023/01/01) to an appropriate value before your flow logs started.

Loading Partitions (if not using Partition Projection)

If you opt not to use Partition Projection (e.g., for very old Athena versions or specific Glue catalog needs), you'd need to manually add partitions for each day's data using ALTER TABLE ADD PARTITION or run MSCK REPAIR TABLE. With the Partition Projection setup above, this step is automated.

Useful Query Examples

Athena's SQL capabilities allow for powerful analysis of your flow log data.

Top Talkers (Bytes Transferred)

Identify which source IPs are sending the most data in a given timeframe. Useful for identifying high-traffic instances or potential data exfiltration.

SELECT
  srcaddr,
  sum(bytes) AS total_bytes
FROM
  vpc_flow_logs
WHERE
  dt = CAST(DATE_FORMAT(current_date, '%Y/%m/%d') AS VARCHAR) -- Querying today's data
GROUP BY
  srcaddr
ORDER BY
  total_bytes DESC
LIMIT 20;

Rejected Traffic Analysis

Detect connection attempts to specific sensitive ports (e.g., SSH, RDP, MySQL) that were explicitly denied by security groups or NACLs.

SELECT
  srcaddr,
  dstaddr,
  dstport,
  count(*) AS hits,
  count(DISTINCT srcaddr) AS unique_src_ips
FROM
  vpc_flow_logs
WHERE
  action = 'REJECT'
  AND dstport IN (22, 3389, 3306, 1433, 5432) -- Common database/admin ports
  AND dt BETWEEN CAST(DATE_FORMAT(current_date - INTERVAL '7' DAY, '%Y/%m/%d') AS VARCHAR) AND CAST(DATE_FORMAT(current_date, '%Y/%m/%d') AS VARCHAR) -- Last 7 days
GROUP BY
  srcaddr,
  dstaddr,
  dstport
ORDER BY
  hits DESC
LIMIT 50;

Connections to Specific Destinations

Find all traffic from a particular ENI or instance to an external IP range or internal service.

SELECT
  interface_id,
  srcaddr,
  dstaddr,
  dstport,
  protocol,
  action,
  FROM_UNIXTIME(start_time) AS flow_start
FROM
  vpc_flow_logs
WHERE
  interface_id = 'eni-0abcdef1234567890'
  AND dstaddr LIKE '172.31.%' -- Traffic to a specific internal subnet
  AND dt = CAST(DATE_FORMAT(current_date, '%Y/%m/%d') AS VARCHAR)
ORDER BY
  flow_start DESC
LIMIT 100;

Egress by Remote AS/Country (Advanced)

For this, you would typically maintain an external IP-to-GeoIP database (e.g., MaxMind GeoLite2) uploaded to S3. You can then create another Athena table for this dataset and join it with your flow logs.

-- Example GeoIP table definition (assuming CSV from MaxMind GeoLite2 City/Country)
CREATE EXTERNAL TABLE IF NOT EXISTS geoip_country (
  network STRING,
  country_iso_code STRING,
  country_name STRING
)
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
WITH SERDEPROPERTIES (
  'separatorChar' = ',',
  'quoteChar' = '"',
  'escapeChar' = '\\'
)
LOCATION 's3://acme-flowlogs/geoip/'
TBLPROPERTIES (
  "skip.header.line.count"="1"
);

-- Query joining flow logs with GeoIP (requires more complex IP-to-CIDR matching)
-- This is a conceptual query as Athena lacks direct CIDR_MATCH or equivalent.
-- Often, you'd pre-process flow logs to add geo data using Glue or Lambda.
SELECT
  f.dstaddr,
  g.country_name,
  SUM(f.bytes) AS total_bytes_egress
FROM
  vpc_flow_logs f
JOIN
  geoip_country g ON CAST(SPLIT(f.dstaddr, '.')[0] AS BIGINT) BETWEEN CAST(SPLIT(SPLIT(g.network, '/')[0], '.')[0] AS BIGINT) AND CAST(SPLIT(SPLIT(g.network, '/')[1], '.')[0] AS BIGINT) -- Simplified example, actual IP range matching is complex
WHERE
  f.dt = CAST(DATE_FORMAT(current_date, '%Y/%m/%d') AS VARCHAR)
  AND f.action = 'ACCEPT'
  AND f.dstaddr NOT LIKE '10.%'
  AND f.dstaddr NOT LIKE '172.16.%'
  AND f.dstaddr NOT LIKE '192.168.%' -- Exclude RFC1918 private IPs
GROUP BY
  f.dstaddr,
  g.country_name
ORDER BY
  total_bytes_egress DESC
LIMIT 20;

Note on GeoIP joins: Direct SQL joins on IP ranges within Athena can be computationally expensive and complex. For large datasets, a more robust solution involves using AWS Glue ETL jobs to enrich flow log data with GeoIP information *before* querying with Athena. This pre-processing step can add columns like country_iso_code and asn directly to the flow log records.

Cost Control and Optimization

Managing the cost of storing and querying massive flow log datasets is crucial.

S3 over CloudWatch Logs: Storing raw flow logs in S3 is significantly cheaper than CloudWatch Logs, especially for long retention periods. CloudWatch Logs pricing often scales with ingest and storage, whereas S3 storage is purely based on volume and access.
Lifecycle Policies: Implement S3 Lifecycle policies to automatically transition older flow logs to lower-cost storage classes (e.g., S3 Standard-IA for infrequent access, or S3 Glacier for archival). For example, move logs older than 30 days to S3 Standard-IA, and older than 90 days to S3 Glacier Flexible Retrieval.
Data Conversion to Parquet: AWS VPC Flow Logs are initially stored as plain text. Converting them to columnar formats like Apache Parquet or ORC can drastically reduce storage costs and Athena query costs/performance. Parquet files are highly compressible and allow Athena to scan only relevant columns, not entire rows.
- Use an AWS Glue crawler to discover the S3 schema and create a Glue catalog table.
- Create an Athena CTAS (Create Table As Select) query to convert the original text files to Parquet, storing them in a new S3 prefix (e.g., s3://acme-flowlogs-parquet/). This query can also perform initial filtering or enrichment.
- Example CTAS for Parquet conversion:
```
CREATE TABLE vpc_flow_logs_parquet
WITH (
  format = 'PARQUET',
  parquet_compression = 'SNAPPY',
  partitioned_by = ARRAY['dt'],
  external_location = 's3://acme-flowlogs-parquet/prod/AWSLogs/<account-id>/vpcflowlogs/<region>/'
) AS
SELECT
  account_id, vpc_id, interface_id, srcaddr, dstaddr, srcport, dstport, protocol, packets, bytes,
  FROM_UNIXTIME(start_time) AS flow_start_ts,
  FROM_UNIXTIME(end_time) AS flow_end_ts,
  action, log_status
FROM
  vpc_flow_logs
WHERE
  dt >= CAST(DATE_FORMAT(current_date - INTERVAL '30' DAY, '%Y/%m/%d') AS VARCHAR); -- Convert recent data
```
  After conversion, point your analytical queries to vpc_flow_logs_parquet for better performance and lower cost.
Partitioning: As demonstrated, partitioning by date (dt) is crucial. Always include a date filter in your WHERE clause (e.g., WHERE dt = 'YYYY/MM/DD') to minimize the data scanned by Athena.

Common Pitfalls

Log Format Mismatch: Ensure your --log-format specified during flow log creation exactly matches the column order and types in your Athena CREATE EXTERNAL TABLE statement. Mismatches lead to corrupted data or query errors.
Missing Partitions: If not using Partition Projection, failing to regularly run MSCK REPAIR TABLE vpc_flow_logs or manually add partitions will prevent Athena from seeing new data.
Incorrect S3 Path: The LOCATION property in your Athena table must point to the correct root folder in S3. Remember AWS adds AWSLogs/<account-id>/vpcflowlogs/<region>/ before the date-based folders.
High Cardinality Joins: Joining flow logs with other large datasets (like GeoIP) directly in Athena can be slow and expensive. Consider pre-processing or using Glue ETL for enrichment.
Forgetting to Filter by Partition: Always include WHERE dt = 'YYYY/MM/DD' (or a range) in your queries. Forgetting this will cause Athena to scan all historical data, leading to high costs and slow queries.
Unnecessary Fields: Only include the fields you truly need in your --log-format. While not a huge impact, fewer fields can slightly reduce log size.
Default Aggregation Interval: The default --max-aggregation-interval of 600 seconds (10 minutes) might be too long for real-time troubleshooting. Consider lowering it to 60 seconds if finer granularity is required, but be aware of increased log volume.