Como usar o Athena para analisar logs de fluxos da Amazon VPC?

Breve descrição

É possível usar o editor de consultas do console Athena para criar um banco de dados, criar uma tabela para os Logs de fluxo da VPC e executar exemplos de consultas. Use os logs de fluxos para analisar padrões de tráfego de rede e identificar ameaças e riscos em toda a sua rede da Amazon VPC.

Resolução

Criar um banco de dados com o editor de consultas do console Athena

Conclua as etapas a seguir:

1. Abra o editor de consultas do console Athena.
2. Execute o seguinte comando:

   CREATE DATABASE test_db_vpclogs;

   Observação: substitua test_db_vpclogs pelo nome do seu banco de dados.

Importante: é uma prática recomendada criar o banco de dados na mesma região da AWS que o bucket do Amazon Simple Storage Service (Amazon S3) no qual os logs de fluxo estão armazenados.

Criar uma tabela para os logs de fluxo no banco de dados

Conclua as etapas a seguir:

1. No editor de consultas do console Athena, execute o seguinte comando:

   CREATE EXTERNAL TABLE `vpcflow_logs`(
    `version` int,
    `resource_type` string,
    `account_id` string,
    `tgw_id` string,
    `tgw_attachment_id` string,
    `tgw_src_vpc_account_id` string,
    `tgw_dst_vpc_account_id` string,
    `tgw_src_vpc_id` string,
    `tgw_dst_vpc_id` string,
    `tgw_src_subnet_id` string,
    `tgw_dst_subnet_id` string,
    `tgw_src_eni` string,
    `tgw_dst_eni` string,
    `tgw_src_az_id` string,
    `tgw_dst_az_id` string,
    `tgw_pair_attachment_id` string,
    `srcaddr` string,
    `dstaddr` string,
    `srcport` int,
    `dstport` int,
    `protocol` bigint,
    `packets` bigint,
    `bytes` bigint,
    `start` bigint,
    `end` bigint,
    `log_status` string,
    `type` string,
    `packets_lost_no_route` bigint,
    `packets_lost_blackhole` bigint,
    `packets_lost_mtu_exceeded` bigint,
    `packets_lost_ttl_expired` bigint,
    `tcp_flags` int,
    `region` string,
    `flow_direction` string,
    `pkt_src_aws_service` string,
    `pkt_dst_aws_service` string)
   PARTITIONED BY (
    `aws_region` string,
    `log_time` string)
   ROW FORMAT DELIMITED
    FIELDS TERMINATED BY ' '
   STORED AS INPUTFORMAT
    'org.apache.hadoop.mapred.TextInputFormat'
   OUTPUTFORMAT
    'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
   LOCATION
    's3://AWS_S3_BUCKET/awsexampleprefix/AWSLogs/AWS_ACCOUNT_NUMBER/vpcflowlogs/'
   TBLPROPERTIES (
    'projection.aws_region.type'='enum',
    'projection.aws_region.values'='us-east-1,us-west-2,ap-southeast-2',
    'projection.log_time.format'='yyyy/MM/dd',
    'projection.log_time.range'='2025/01/01,NOW',
    'projection.log_time.type'='date',
    'projection.enabled'='true',
    'skip.header.line.count'='1',
    'storage.location.template'='s3://AWS_S3_BUCKET/awsexampleprefix/AWSLogs/AWS_ACCOUNT_NUMBER/vpcflowlogs/${aws_region}/${log_time}')

   Observação: substitua test_table_vpclogs na consulta pelo nome da sua tabela.

Modifique o parâmetro LOCATION para apontar para o bucket do Amazon S3 que contém seus dados de log. Defina o primeiro valor de projection.log_time.range com base nos primeiros logs disponíveis em seus dados. Por exemplo, se seus logs começarem em 1.º de março de 2023, defina o período como '2025/03/01,NOW'. Modifique o valor de acordo com a data real de início dos seus logs para uma projeção de dados precisa. O comando anterior usa a projeção de partição para criar uma tabela, particionar a tabela e preencher automaticamente as partições. Se uma partição projetada não existir no Amazon S3, o Athena ainda projetará a partição. A prática recomendada é usar atributos particionados em suas consultas.

Executar instruções SQL na tabela de logs de fluxo

Use o editor de consultas do console Athena para executar instruções SQL na tabela. É possível salvar consultas, visualizar consultas anteriores ou baixar resultados de consultas no formato de arquivo .csv.

Exemplos de consultas

Observação: nos exemplos de consultas a seguir, substitua test\ _table\ _vpclogs pelo nome da sua tabela. Modifique os valores da coluna e outras variáveis para seu caso de uso.

Para visualizar as primeiras 100 entradas do log de fluxo em ordem cronológica para um período de tempo especificado, execute a seguinte consulta:

SELECT *  FROM test_table_vpclogs
 WHERE day >= '2021/02/01' AND day < '2021/02/28'
 ORDER BY day ASC
 LIMIT 100;

Para ver qual servidor recebe os dez maiores números de pacotes HTTP para um determinado período de tempo, execute a seguinte consulta:

SELECT SUM(packets) AS packetcount,
        dstaddr
FROM test_table_vpclogs
WHERE dstport = 443
  AND day >= '2021/03/01'
  AND day < '2021/03/31'
GROUP BY dstaddr
ORDER BY packetcount DESC
LIMIT 10;

Observação: a consulta anterior conta o número de pacotes recebidos na porta HTTPS 443 e os agrupa por endereço IP de destino. Em seguida, a consulta retorna as 10 principais entradas da semana anterior.

Para verificar os logs que são criados em um intervalo de tempo especificado, execute a seguinte consulta:

SELECT interface_id,       
       srcaddr,
       action,
       protocol,
       to_iso8601(from_unixtime(start)) AS start_time,
       to_iso8601(from_unixtime("end")) AS end_time
FROM test_table_vpclogs
WHERE DAY >= '2021/04/01'
  AND DAY < '2021/04/30';

Para visualizar os logs de fluxo de um endereço IP de origem específico entre um intervalo de tempo especificado, execute a seguinte consulta:

SELECT *FROM test_table_vpclogs
WHERE srcaddr = '10.117.1.22'
  AND day >= '2021/02/01'
  AND day < '2021/02/28';

Para listar as conexões TCP rejeitadas entre um intervalo de tempo especificado, execute a seguinte consulta:

SELECT day,       
interface_id,
       srcaddr,
       action,
       protocol
FROM test_table_vpclogs
WHERE action = 'REJECT'
    AND protocol = 6
    AND day >= '2021/02/01' AND day < '2021/02/28'
LIMIT 10;

Para visualizar os logs de fluxo do intervalo de endereços IP que começa com 10.117, execute a seguinte consulta:

SELECT *FROM test_table_vpclogs
WHERE split_part(srcaddr,'.', 1)='10'
  AND split_part(srcaddr,'.', 2) ='117'

Para visualizar os logs de fluxo de um endereço IP de destino específico entre um intervalo de tempo, execute a seguinte consulta:

SELECT *FROM test_table_vpclogs
WHERE dstaddr = '10.0.1.14'
AND day >= '2021/01/01'
AND day < '2021/01/31'

Informações relacionadas

Como monitorar o tráfego na minha VPC com logs de fluxo?

Análise de logs de fluxo de VPC usando o Amazon Athena e o Amazon Quick Sight