Amazon VPC Flow Logs の Athena 統合を使って、S3 に保存した VPC Flow Logs を分析する機会があったので備忘録。Athena は難しそうなイメージがあったのですが、Athena 統合のおかげで Athena 側のややこしいところは自動で設定してくれるようになっていて、ずいぶんと敷居が低くなっていました。とはいえ、各種設定の意味をドキュメントで調べたりしたので、手順や参考にしたドキュメントを残します。
目次
- VPC Flow Logs を有効化
- VPC Flow logs が S3 に保存されているか確認する
- Athena 統合で CloudFormation テンプレートを作成する
- CloudFormation テンプレートから Athena のリソースを作成する
- Athena で Flow Logs をクエリする
- よく使う Athena のクエリ
- エクセルで unixtime を date に変換する
VPC Flow Logs を有効化
VPC Flow ログは、VPC の詳細画面の「Flow Logs」のタブの「Create Flow Logs」ボタンから有効化ができます。
フローログを Amazon S3 に発行する - Amazon Virtual Private Cloud を見ながら設定していきます。ポイントを以下で解説します。ここに触れているもの以外はデフォルトで良いと思います。
S3 バケットポリシーは VPC Flow Logs を作成したときに自動で設定してくれる
VPC Flow Logs を保存するための S3 バケットには、バケットポリシーを設定する必要があります。フローログのための Amazon S3 バケットのアクセス許可 に説明がありますが、VPC Flow Logs を作成したときに S3 のバケットのポリシーが自動で設定されるので、バケットポリシーは何もしなくても大丈夫でした。Terraform で VPC Flow Logs を作成したときも自動で設定されました。
Log record format は Custom format を指定する
default format だと得られる情報が少ないので Custom format を指定しましょう。使用可能なフィールド を見ながら追加のフィールドを指定します。慣れないうちはすべて指定しても良いと思います。
Terraform で項目を指定する場合は以下のようにエスケープします。
log_format = "$${version} $${account-id} $${interface-id} $${srcaddr} $${dstaddr} $${srcport} $${dstport} $${protocol} $${packets} $${bytes} $${start} $${end} $${action} $${log-status} $${subnet-id} $${instance-id} $${tcp-flags} $${type} $${pkt-srcaddr} $${pkt-dstaddr} $${region} $${az-id} $${sublocation-type} $${sublocation-id} $${pkt-src-aws-service} $${pkt-dst-aws-service} $${flow-direction} $${traffic-path}"Log file format は Parquet を選択
高速でストレージ料金も安くなるので Parquet を選択します。詳細はフローログファイルを参照。
Partition logs by time は Every 1 hour (60 minutes) を選択
VPC Flow logs は量が多くなることが多いです。ログを時間単位で分割し、Athena のクエリの検索条件に時間を指定することで、より高速な結果が得られ、クエリコストを節約できます。詳細はフローログファイルを参照。
VPC Flow logs が S3 に保存されているか確認する
設定が完了したら 10 分ほど待ってから S3 バケットの中身をみてみます。うまくいっていれば日付でフォルダが作られて、ログが保存されています。
Athena 統合で CloudFormation テンプレートを作成する
作成された VPC Flow Logs を選択し「アクション」「Athena 統合の生成」の順に選択します。
コンソールを使用した CloudFormation テンプレートの生成 を参考に進めます。
- パーティションのロード頻度は「毎日」を選択します。パーティションについて詳細は Athena でのデータのパーティション化 を参照。
- パーティションの開始日と終了日は空欄にしておきます。
- CloudFormation テンプレートを出力するための S3 バケットを選択します。私は CloudFormation をマネジメントコンソールから利用すると自動で生成される cf-templates-xxxx-ap-northeast-1 のバケットを指定しました。
- Athena のクエリ結果を保存するための S3 バケットを作成し、選択します。
上記を設定し、「Athena 統合を生成」ボタンをクリックすると、S3 バケットに CloudFormation テンプレートが作成されるので、ローカル PC にダウンロードします。
CloudFormation テンプレートから Athena のリソースを作成する
ダウンロードした CloudFormation テンプレートを使って、Athena のリソースを作成します。
Athena で Flow Logs をクエリする
- Athena コンソールを開いて、右上の「ワークグループ」から CloudFormation テンプレートによって作成されたワークグループを選択します。
- 定義済みクエリの 1 つを選択し、必要に応じてパラメータを変更してから、クエリを実行します。
- Athena に結果が表示されます。S3 バケットににもクエリの結果が CSV 形式で保存されています。
よく使う Athena のクエリ
VPC Flow Logs を確認するときは何か問題が起きていることが多く、特定の IP アドレスやポート番号に絞って確認をすることが多いです。私がよく使うクエリは以下の通り。
特定の IP アドレスの通信を調べる。
SELECT * FROM fl0effxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
WHERE start BETWEEN 1669605300 AND 1669606200
AND ( srcaddr = '192.168.0.10' OR dstaddr = '192.168.0.10' )
AND month = '11' AND day = '28' AND hour = '03'
ORDER BY "start" ASC特定のポート番号の通信を調べる。
SELECT * FROM fl0effxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
WHERE start BETWEEN 1669605300 AND 1669606200
AND ( srcport = 443 OR dstport = 443 )
AND month = '11' AND day = '28' AND hour = '03'
ORDER BY "start" ASCmacOS で unixtime と date の変換
上記のクエリの例で WHERE 条件で指定している の start と end が unixtime なので、macOS で unixtime と datet の変換をするコマンドを紹介します。詳細は macOS の man date コマンドを参照してください。
macOS で UTC の date を unixtime に変換する。
date -u -j -f "%Y-%m-%d %H:%M:%S" "2022-11-28 03:15:00" +%s1669605300macOS で UTC の unixtime を date に変換する。
date -u -r 1669606200Mon Nov 28 03:30:00 UTC 2022エクセルで unixtime を date に変換する
VPC Flow Logs を csv でダウンロードしたあと、エクセルで内容を確認することもよくあります。その際、start, end の unixtime を date に変換する例です。
- K 列に start がある場合
- start の隣の列に新しい列を作成
- ブランクのセルに
=(((K2/60)/60)/24)+DATE(1970,1,1)という式を設定 - 列を選択して右クリック、セルのフォーマットを時間に変更
- 残りの空欄のセルに式を適用
VPC Flow Logs の削除
VPC Flow Logs が不要になったら、以下の手順で削除します。
- VPC Flow Logs を削除する
- Athena のワークグループを削除する
- Athena のクエリ結果のオブジェクトを削除する
- Athena 統合の CloudFormation のテンプレートを S3 から削除する
- CloudFormation のスタックを削除する