Find S3 and DynamoDB Gateway Endpoint Gaps With Aurora PostgreSQL's New S3 Queries

aws rds create-db-cluster-parameter-group \
  --db-cluster-parameter-group-name aurora-analytics-pg17 \
  --db-parameter-group-family aurora-postgresql17 \
  --description "aurora_analytics enabled"

aws rds modify-db-cluster-parameter-group \
  --db-cluster-parameter-group-name aurora-analytics-pg17 \
  --parameters ParameterName=aurora_analytics.enabled,ParameterValue=1,ApplyMethod=immediate

aws rds create-db-cluster \
  --db-cluster-identifier aurora-analytics \
  --engine aurora-postgresql --engine-version 17.11 \
  --db-cluster-parameter-group-name aurora-analytics-pg17 \
  --master-username postgres --manage-master-user-password \
  --database-name inventory \
  --db-subnet-group-name <your-subnet-group> \
  --vpc-security-group-ids <your-db-security-group> \
  --serverless-v2-scaling-configuration MinCapacity=0,MaxCapacity=8,SecondsUntilAutoPause=600 \
  --storage-encrypted

aws rds create-db-instance \
  --db-instance-identifier aurora-analytics-1 \
  --db-cluster-identifier aurora-analytics \
  --engine aurora-postgresql --db-instance-class db.serverless

aws rds add-role-to-db-cluster \
  --db-cluster-identifier aurora-analytics \
  --role-arn arn:aws:iam::111122223333:role/aurora-analytics-s3-role \
  --feature-name AuroraAnalytics

create extension aurora_analytics;

kind: source
spec:
  name: aws
  path: cloudquery/aws
  registry: cloudquery
  version: "v34.9.0"
  tables:
    - aws_ec2_vpcs
    - aws_ec2_vpc_endpoints
    - aws_ec2_flow_logs
    - aws_ec2_nat_gateways
  destinations: ["postgresql"]
  spec:
    regions: ["us-east-1"]
---
kind: destination
spec:
  name: postgresql
  path: cloudquery/postgresql
  registry: cloudquery
  version: "v8.0.0"
  spec:
    connection_string: "${PG_CONN}"

create or replace view vpc_gateway_endpoint_gaps as
with gw as (
  select vpc_id,
         bool_or(service_name like '%.s3')       as has_s3_gateway,
         bool_or(service_name like '%.dynamodb') as has_ddb_gateway
  from aws_ec2_vpc_endpoints
  where vpc_endpoint_type = 'Gateway' and state = 'available'
  group by vpc_id
),
nat as (
  select vpc_id, count(*) as nat_gateways
  from aws_ec2_nat_gateways where state = 'available'
  group by vpc_id
)
select distinct on (v.vpc_id)
       v.account_id, v.region, v.vpc_id,
       coalesce(gw.has_s3_gateway, false)  as has_s3_gateway,
       coalesce(gw.has_ddb_gateway, false) as has_ddb_gateway,
       coalesce(nat.nat_gateways, 0)       as nat_gateways,
       fl.flow_log_id, fl.traffic_type,
       replace(fl.log_destination, 'arn:aws:s3:::', 's3://') as flow_log_s3_uri,
       fl.destination_options->>'FileFormat' as file_format
from aws_ec2_vpcs v
join aws_ec2_flow_logs fl
  on fl.resource_id = v.vpc_id
 and fl.flow_log_status = 'ACTIVE'
 and fl.log_destination_type = 's3'
left join gw  on gw.vpc_id  = v.vpc_id
left join nat on nat.vpc_id = v.vpc_id
where not (coalesce(gw.has_s3_gateway, false) and coalesce(gw.has_ddb_gateway, false))
order by v.vpc_id,
         (fl.destination_options->>'FileFormat' = 'parquet') desc,
         (fl.traffic_type = 'ALL') desc,
         fl.creation_time desc;

do $$
declare
  r record;
  t text;
  parts text[] := '{}';
begin
  drop view if exists candidate_vpc_flow_logs;
  for r in
    select vpc_id, flow_log_id, flow_log_s3_uri
    from vpc_gateway_endpoint_gaps
    where file_format = 'parquet'
  loop
    t := 'flow_logs_' || replace(r.flow_log_id, '-', '_');
    execute format('drop foreign table if exists %I', t);
    begin
      execute format(
        'create foreign table %I () server aurora_analytics_server
           options (location %L, format %L)',
        t,
        rtrim(r.flow_log_s3_uri, '/') || '/AWSLogs/**/*_' || r.flow_log_id || '_*.parquet',
        'parquet');
    exception when others then
      raise notice 'skipping % (%): %', r.vpc_id, r.flow_log_id, sqlerrm;
      continue;
    end;
    parts := parts || format(
      'select %L::text as vpc_id, interface_id, srcaddr, dstaddr, srcport, dstport,
              protocol, packets, bytes, start, "end", action from %I',
      r.vpc_id, t);
  end loop;
  if array_length(parts, 1) is null then
    raise exception 'No candidate VPC has Parquet flow-log files yet. Let the flow logs collect data, then rerun this block.';
  end if;
  execute 'create view candidate_vpc_flow_logs as '
       || array_to_string(parts, ' union all ');
end $$;

curl -s https://ip-ranges.amazonaws.com/ip-ranges.json -o ip-ranges.json
{
  echo "begin;"
  echo "drop table if exists aws_service_ip_ranges;"
  echo "create table aws_service_ip_ranges (prefix cidr not null, service text not null,"
  echo "  region text not null, network_border_group text);"
  echo "insert into aws_service_ip_ranges values"
  jq -r '[.prefixes[] | {p:.ip_prefix, s:.service, r:.region, n:.network_border_group}]
       + [.ipv6_prefixes[] | {p:.ipv6_prefix, s:.service, r:.region, n:.network_border_group}]
       | map(select(.s=="S3" or .s=="DYNAMODB")) | unique
       | map("(\(.p|@sh), \(.s|@sh), \(.r|@sh), \(.n|@sh))")
       | join(",\n")' ip-ranges.json
  echo ";"
  echo "create index on aws_service_ip_ranges using gist (prefix inet_ops);"
  echo "commit;"
} > aws_service_ip_ranges.sql

psql "$PG_CONN" -f aws_service_ip_ranges.sql

with nat_enis as (
  select a->>'NetworkInterfaceId' as interface_id
  from aws_ec2_nat_gateways, jsonb_array_elements(nat_gateway_addresses) a
),
flows as (
  select f.vpc_id, f.interface_id, f.srcaddr::inet as src, f.dstaddr::inet as dst,
         f.bytes, f.start, f."end"
  from candidate_vpc_flow_logs f
  where f.srcaddr <> '-' and f.dstaddr <> '-'
    and not exists (select 1 from nat_enis n where n.interface_id = f.interface_id)
),
matched as (
  select f.*, r.service
  from flows f
  join lateral (
    select service from aws_service_ip_ranges r
    where f.dst <<= r.prefix or f.src <<= r.prefix
    order by masklen(r.prefix) desc
    limit 1
  ) r on true
),
window_days as (
  select vpc_id, greatest((max("end") - min(start)) / 86400.0, 1) as days
  from candidate_vpc_flow_logs group by vpc_id
)
select g.vpc_id, s.service,
       case s.service when 'S3' then g.has_s3_gateway else g.has_ddb_gateway end as has_gateway,
       g.nat_gateways,
       round(w.days, 1)                                        as days_of_logs,
       count(m.*)                                              as flow_records,
       count(distinct m.interface_id)                          as enis,
       round(coalesce(sum(m.bytes), 0) / 1e9, 3)               as gb_total,
       round(coalesce(sum(m.bytes), 0) / 1e9 / w.days * 30, 3) as gb_per_30_days,
       case when g.nat_gateways > 0
             and not (case s.service when 'S3' then g.has_s3_gateway else g.has_ddb_gateway end)
            then round(coalesce(sum(m.bytes), 0) / 1e9 / w.days * 30 * 0.045, 2)
            else 0 end                                         as max_nat_savings_usd_30d
from vpc_gateway_endpoint_gaps g
cross join (values ('S3'), ('DYNAMODB')) s(service)
join window_days w on w.vpc_id = g.vpc_id
left join matched m on m.vpc_id = g.vpc_id and m.service = s.service
group by g.vpc_id, s.service, g.has_s3_gateway, g.has_ddb_gateway, g.nat_gateways, w.days
order by g.vpc_id, s.service;