Come posso utilizzare Athena per analizzare i log di flusso Amazon VPC?

Breve descrizione

Puoi utilizzare l'editor di query della console Athena per creare un database, creare una tabella per i ](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html)log di flusso VPC[ ed eseguire esempi di query. Utilizza i log di flusso per analizzare i modelli di traffico di rete e identificare minacce e rischi nella rete Amazon VPC.

Risoluzione

Crea un database con l'editor di query della console Athena

Completa i seguenti passaggi:

1. Apri l'editor di query della console Athena.
2. Esegui questo comando:

   CREATE DATABASE test_db_vpclogs;

   Nota: sostituisci test_db_vpclogs con il nome del tuo database.

Importante: è consigliabile creare il database nella stessa Regione AWS del bucket Amazon Simple Storage Service (Amazon S3) in cui sono archiviati i log di flusso.

Crea una tabella per i log di flusso nel database

Completa i seguenti passaggi:

1. Nell'editor di query della console Athena, esegui questo 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}')

   Nota: sostituisci test_table_vpclogs con il nome della tua tabella.

Modifica il parametro LOCATION in modo che punti verso il bucket Amazon S3 che contiene i dati di log. Imposta il primo valore di projection.log_time.range in base ai primi log disponibili nei dati. Ad esempio, se i log iniziano il 1° marzo 2023, imposta l'intervallo di tempo come '2025/03/01,NOW'. Modifica il valore in base alla data di inizio effettiva dei log per una proiezione accurata dei dati. Il comando precedente utilizza la proiezione delle partizioni per creare una tabella, partizionarla e popolare automaticamente le partizioni. Se una partizione proiettata non esiste in Amazon S3, Athena la proietta comunque. È consigliabile utilizzare attributi partizionati nelle query.

Esegui istruzioni SQL sulla tabella per i log di flusso

Utilizza l'editor di query della console Athena per eseguire istruzioni SQL sulla tabella. Puoi salvare le query, visualizzare quelle precedenti o scaricare i risultati delle query in formato .csv.

Esempi di query

Nota: nei seguenti esempi di query, sostituisci test_table_vpclogs con il nome della tua tabella. Modifica i valori delle colonne e altre variabili in base al caso d'uso.

Per visualizzare le prime 100 voci del log di flusso in ordine cronologico per un periodo di tempo specificato, esegui questa query:

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

Per visualizzare il server che riceve i primi dieci pacchetti HTTP per un periodo di tempo specificato, esegui questa query:

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;

Nota: la query precedente conta il numero di pacchetti ricevuti sulla porta HTTPS 443 e li raggruppa per indirizzo IP di destinazione. Quindi restituisce le prime 10 voci della settimana precedente.

Per controllare i log creati in un determinato intervallo di tempo, esegui questa query:

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';

Per visualizzare i log di flusso per un indirizzo IP di origine specifico in un determinato intervallo di tempo, esegui questa query:

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

Per elencare le connessioni TCP rifiutate in un determinato intervallo di tempo, esegui questa query:

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;

Per visualizzare i log di flusso per l'intervallo di indirizzi IP che inizia con 10.117, esegui questa query:

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

Per visualizzare i log di flusso per un indirizzo IP di destinazione specifico in un determinato intervallo di tempo, esegui questa query:

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

Informazioni correlate

How do I monitor traffic in my VPC with flow logs? (Come posso monitorare il traffico nel mio VPC con i log di flusso?)

Analyzing VPC Flow Logs using Amazon Athena, and Amazon QuickSight (Analisi dei log di flusso VPC utilizzando Amazon Athena e Amazon QuickSight)