CloudWatch Logs Insights を使用してカスタム Amazon VPC フローログを分析するにはどうすればよいですか?

簡単な説明

CloudWatch Logs Insights は、デフォルト形式のフローログを自動検出しますが、カスタム形式のフローログは自動検出しません。

CloudWatch Logs Insights をカスタム形式のフローログで使用するには、クエリを変更する必要があります。

次に示すのは、カスタム形式のフローログの例です。

${account-id} ${vpc-id} ${subnet-id} ${interface-id} ${instance-id} ${srcaddr} ${srcport} ${dstaddr} ${dstport} ${protocol} ${packets} ${bytes} ${action} ${log-status} ${start} ${end} ${flow-direction} ${traffic-path} ${tcp-flags} ${pkt-srcaddr} ${pkt-src-aws-service} ${pkt-dstaddr} ${pkt-dst-aws-service} ${region} ${az-id} ${sublocation-type} ${sublocation-id}

以下のクエリの例では、ユースケースに合わせてクエリをどのようにカスタマイズおよび拡張するかを示しています。

解決策

最新のフローログを取得する

ログフィールドからデータを抽出するには、parse キーワードを使用します。例えば、次のクエリの出力は、フローログイベントの開始時刻でソートされ、最新の 2 つのログエントリに制限されます。

クエリ

#Retrieve latest custom VPC Flow Logs
parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| sort start desc
| limit 2

出力

account_id	vpc_id	subnet_id	interface_id	instance_id	srcaddr	srcport
123456789012	vpc-0b69ce8d04278ddd	subnet-002bdfe1767d0ddb0	eni-0435cbb62960f230e	-	172.31.0.104	55125
123456789012	vpc-0b69ce8d04278ddd1	subnet-002bdfe1767d0ddb0	eni-0435cbb62960f230e	-	91.240.118.81	49422

送信元と送信先の IP アドレスのペア別でデータ転送を要約する

次のクエリを使用して、送信元と送信先の IP アドレスのペア別にネットワークトラフィックを要約します。このサンプルクエリでは、sum 統計が bytes フィールドを集計します。sum 統計は、ホスト間で転送されるデータの累積合計を計算します。このため、flow_direction がクエリと出力に含まれます。集計の結果は一時的に Data_Transferred フィールドに割り当てられます。その後、結果が Data_Transferred の降順でソートされ、最も大きい 2 つのペアが返されます。

クエリ

parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| stats sum(bytes) as Data_Transferred by srcaddr, dstaddr, flow_direction
| sort by Data_Transferred desc
| limit 2

出力

srcaddr	dstaddr	flow_direction	Data_Transferred
172.31.1.247	3.230.172.154	egress	346952038
172.31.0.46	3.230.172.154	egress	343799447

Amazon EC2 インスタンス ID 別のデータ転送を分析する

カスタムフローログを使用して、Amazon Elastic Compute Cloud (Amazon EC2) インスタンス ID 別のデータ転送を分析できます。最もアクティブな EC2 インスタンスを確認するには、クエリに instance_id フィールドを含めます。

クエリ

parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| stats sum(bytes) as Data_Transferred by instance_id
| sort by Data_Transferred desc
| limit 5

出力

instance_id	Data_Transferred
-	1443477306
i-03205758c9203c979	517558754
i-0ae33894105aa500c	324629414
i-01506ab9e9e90749d	198063232
i-0724007fef3cb06f3	54847643

拒否された SSH トラフィックに対するフィルタリング

セキュリティグループとネットワークアクセスコントロールリスト (ネットワーク ACL) が拒否したトラフィックを分析するには、REJECT フィルターアクションを使用します。SSH トラフィックで拒否されたホストを識別するには、TCP プロトコルと宛先ポート 22 のトラフィックを含むようにフィルターを拡張します。次のサンプルクエリでは、TCP プロトコル 6 が使用されています。

クエリ

parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| filter action = "REJECT" and protocol = 6 and dstport = 22
| stats sum(bytes) as SSH_Traffic_Volume by srcaddr
| sort by SSH_Traffic_Volume desc
| limit 2

出力

srcaddr	SSH_Traffic_Volume
23.95.222.129	160
179.43.167.74	80

特定の送信元/送信先ペアの HTTP データストリームを分離する

データの傾向を分析するには、CloudWatch Logs Insights を使用して 2 つの IP アドレス間の双方向トラフィックを分離します。次のクエリでは、["172.31.1.247","172.31.11.212"] は IP アドレスを送信元 IP アドレスまたは送信先 IP アドレスとして使用してフローログを返します。filter ステートメントは、VPC フローログイベントを TCP プロトコル 6 およびポート 80 と照合して、HTTP トラフィックを分離します。使用可能なすべてのフィールドのサブセットを返すには、display キーワードを使用します。

クエリ

次のクエリを参照してください。

#HTTP Data Stream for Specific Source/Destination Pair
parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| filter srcaddr in ["172.31.1.247","172.31.11.212"] and dstaddr in ["172.31.1.247","172.31.11.212"] and protocol = 6 and (dstport = 80 or srcport=80)
| display interface_id,srcaddr, srcport, dstaddr, dstport, protocol, bytes, action, log_status, start, end, flow_direction, tcp_flags
| sort by start desc
| limit 2

出力

interface_id	srcaddr	srcport	dstaddr	dstport	protocol	bytes	action	log_status
eni-0b74120275654905e	172.31.11.212	80	172.31.1.247	29376	6.	5160876	ACCEPT	OK
eni-0b74120275654905e	172.31.1.247	29376	172.31.11.212	80	6.	97380	ACCEPT	OK

結果を棒グラフまたは円グラフとして視覚化する

CloudWatch Log Insights を使用して、結果を棒グラフまたは円グラフとして視覚化できます。結果が bin() 関数を含んでいる場合、クエリ出力はタイムスタンプ付きで返されます。その後、折れ線グラフまたは積み上げ面グラフで時系列を視覚化できます。

1 分間隔で転送される累積データを計算するには、stats sum(bytes) を Data_Trasferred by bin(1m) として使用します。このビジュアライゼーションを確認するには、CloudWatch Logs Insights コンソールで [ログ] テーブルと [ビジュアライゼーション] テーブルを切り替えます。

クエリ

parse @message "* * * * * * * * * * * * * * * * * * * * * * * * * * *" as account_id, vpc_id, subnet_id, interface_id,instance_id, srcaddr, srcport, dstaddr, dstport, protocol, packets, bytes, action, log_status, start, end, flow_direction, traffic_path, tcp_flags, pkt_srcaddr, pkt_src_aws_service, pkt_dstaddr, pkt_dst_aws_service, region, az_id, sublocation_type, sublocation_id
| filter srcaddr in ["172.31.1.247","172.31.11.212"] and dstaddr in ["172.31.1.247","172.31.11.212"] and protocol = 6 and (dstport = 80 or srcport=80)
| stats sum(bytes) as Data_Transferred by bin(1m)

出力

bin(1m)	Data_Transferred
2022-04-01 15:23:00.000	17225787
2022-04-01 15:21:00.000	17724499
2022-04-01 15:20:00.000	1125500
2022-04-01 15:19:00.000	101525
2022-04-01 15:18:00.000	81376

関連情報

Supported logs and discovered fields

CloudWatch Logs Insights query syntax