Virtual Private Cloud:VPC フローログ

コンソール

API

フローログを作成する前に、フローログ機能が有効になっており、SLS でプロジェクトと Logstore が作成されていることを確認してください。

• OpenFlowLogService を呼び出して、フローログ機能を有効にします。

• CreateProject を呼び出してプロジェクトを作成し、CreateLogstore を呼び出して Logstore を作成します。

これらの前提条件を満たした後、次の操作を実行できます。

• CreateFlowLog を呼び出してフローログを作成します。CreateIndex を呼び出してインデックスを作成することもできます。

• DeactiveFlowLog を呼び出してフローログを停止します。

• ActiveFlowLog を呼び出してフローログを開始します。

• DeleteFlowLog を呼び出してフローログを削除します。

Terraform

リソース：alicloud_log_project、alicloud_log_store、alicloud_vpc_flow_log

# フローログを作成するリージョンを指定します。
provider "alicloud" {
  region = "cn-hangzhou"
}

# プロジェクトの説明、Logstore とフローログの名前を指定します。
variable "name" {
  default = "vpc-flowlog-example"
}

# プロジェクト名用にランダムな文字列を生成します。
resource "random_uuid" "example" {
}

# Simple Log Service (SLS) プロジェクトを作成します。
resource "alicloud_log_project" "example" {
  project_name = substr("tf-example-${replace(random_uuid.example.result, "-", "")}", 0, 16)
  description  = var.name
}

# Simple Log Service (SLS) Logstore を作成します。
resource "alicloud_log_store" "example" {
  project_name          = alicloud_log_project.example.project_name
  logstore_name         = var.name
  shard_count           = 3
  auto_split            = true
  max_split_shard_count = 60
  append_meta           = true
}

# VPC フローログを作成します。
resource "alicloud_vpc_flow_log" "example" {
  flow_log_name        = var.name
  log_store_name       = alicloud_log_store.example.logstore_name
  description          = var.name
  traffic_path         = ["all"] # すべてのシナリオからトラフィックをキャプチャします。
  project_name         = alicloud_log_project.example.project_name
  resource_type        = "VPC" # リソースタイプは VPC です。
  resource_id          = "vpc-bp1ekmgzch0bo3hxXXXXXX" # VPC の ID です。
  aggregation_interval = "1" # キャプチャウィンドウは 1 分です。
  traffic_type         = "All" # アクセス制御ルールによって許可されるか拒否されるかに関わらず、すべてのトラフィックをキャプチャします。
}

コンソール

API

GetLogsV2 を呼び出して、フローログをクエリおよび分析します。

インターネットソース IP の分析

ポート 80 でリッスンし、インターネットからアクセス可能な Web サーバーがあるとします。サーバーへのアクセスは、送信元 IP アドレスに基づいてセキュリティグループによって制限されています。

この場合、フローログを作成して、ポート 80 にアクセスする送信元 IP アドレスをクエリし、セキュリティグループによって許可または拒否されたリクエストをカウントできます。

1. フローログの作成

   • リソースインスタンス には、Web サーバーの ネットワークインターフェイス を選択します。

   • トラフィックタイプ には、すべてのトラフィック を選択します。

   • 配信設定には、SLS に送信 と FlowLog レポート分析機能の有効化 を選択します。

   • 他のオプションはデフォルト設定のままにします。

2. フローログの分析

   1. クエリ文

      次のクエリを実行して、10.0.0.1 のポート 80 にアクセスする送信元 IP アドレスをフィルタリングします。クエリはまた、セキュリティグループが各 IP アドレスを許可または拒否した回数も表示します。

      dstaddr:10.0.0.1 AND dstport:80 | SELECT -- 宛先 IP アドレスが 10.0.0.1 で、宛先ポートが 80 のログをフィルタリングします。
      srcaddr,
      SUM(CASE WHEN action = 'ACCEPT' THEN 1 ELSE 0 END) AS accept_count, -- 各 ACCEPT (許可) アクションに対して 1 をカウントします。
      SUM(CASE WHEN action = 'REJECT' THEN 1 ELSE 0 END) AS reject_count -- 各 REJECT (拒否) アクションに対して 1 をカウントします。
      FROM log
      GROUP BY srcaddr -- 送信元 IP アドレスでグループ化します。
      ORDER BY accept_count + reject_count DESC -- 許可されたリクエストと拒否されたリクエストの合計数で結果を降順にソートします。

   2. プレビュー

      srcaddr カラムにはポート 80 にアクセスしたソース IP が表示され、accept_count カラムと reject_count カラムには、クエリ時間範囲内にセキュリティグループによって許可されたフローログエントリと拒否されたフローログエントリの数が、各ソース IP ごとにそれぞれカウントされます：

      • 5 つの送信元 IP アドレスがポート 80 にアクセスしました：120.26.XX.XX、121.43.XX.XX、154.212.XX.XX、176.65.XX.XX、および 198.235.XX.XX。

      • 120.26.XX.XX からのすべてのリクエストは許可されましたが、他のパブリック IP アドレスからのすべてのリクエストは拒否されました。

ECS トラフィックの分析

(srcaddr:10.0.0.1 AND dstaddr:10.0.0.*) OR (srcaddr:10.0.0.* AND dstaddr:10.0.0.1 )  | select -- ECS1 と他の ECS インスタンス間のトラフィックをフィルタリングします。
date_format(from_unixtime(__time__ - __time__% 60), '%H:%i:%S') as time, -- UNIX タイムスタンプを読み取り可能な時間形式に変換します。
concat(srcaddr,'->', dstaddr) as src_to_dst, -- 送信元と宛先の IP アドレスを「source_ip->destination_ip」の形式のセッションペアに連結します。
sum(bytes*8/60) as bandwidth -- バイトをビットに変換し、1 分のキャプチャウィンドウで割って、レートを bit/s で取得します。
group by time,srcaddr,dstaddr -- 時間、送信元 IP、宛先 IP でグループ化します。
order by time asc  -- 時間で昇順にソートします。
limit 100 -- 最初の 100 件の結果を表示します。

(srcaddr:10.0.* AND dstaddr:172.16.*) OR (srcaddr:172.16.* AND dstaddr:10.0.*)  | select -- 2 つの VPC 間のセッションをフィルタリングします。
date_format(from_unixtime(__time__ - __time__% 60), '%H:%i:%S') as time, -- UNIX タイムスタンプを読み取り可能な時間形式に変換します。
concat(srcaddr,'->', dstaddr) as src_to_dst, -- 送信元と宛先の IP アドレスを「source_ip->destination_ip」の形式のセッションペアに連結します。
sum(bytes*8/60) as bandwidth -- バイトをビットに変換し、1 分のキャプチャウィンドウで割って、レートを bit/s で取得します。
group by time,srcaddr,dstaddr -- 時間、送信元 IP、宛先 IP でグループ化します。
order by time asc  -- 時間で昇順にソートします。
limit 100 -- 最初の 100 件の結果を表示します。

NAT ゲートウェイトラフィックの分析

あるリージョンに複数の ECS インスタンスがあるとします。すべてのインスタンスは同じ vSwitch にデプロイされ、インターネット NAT Gateway の SNAT を使用してインターネットにアクセスします。

最近、NAT ゲートウェイからのアウトバウンドトラフィックが急増し、サーバーの応答が遅くなっていることに気づきました。フローログを使用して、トラフィックの大部分を生成している ECS インスタンスを特定します。

1. フローログの作成

   • リソースインスタンス には、インターネット NAT Gateway が配置されている vSwitch2 を選択します。

   • 配信設定には、SLS に送信 と FlowLog レポート分析機能の有効化 を選択します。

   • 他のオプションはデフォルト設定のままにします。

2. 特定のトラフィックパスをフィルタリングする方法

   このシナリオで特定のパスのトラフィックをフィルタリングするには、クエリ文で異なる条件を指定します。

   図	No.	フィルタリング方法
   	①	ECS から NAT Gateway へのトラフィックをフィルタリングする場合： direction は in、srcaddr は ECS のプライベート IP アドレスです。
   	②	NAT ゲートウェイからパブリックネットワークへのトラフィックをフィルターするには、direction は out で、srcaddr は NAT ゲートウェイのプライベート IP アドレスです。
   	③	インターネットから NAT ゲートウェイへのトラフィックをフィルター：direction は in で、dstaddr は NAT ゲートウェイのプライベート IP アドレスです。
   	④	NAT ゲートウェイから ECS インスタンスへのトラフィックをフィルターします。direction は out で、dstaddr は ECS インスタンスのプライベート IP アドレスです。

3. フローログの分析

   1. クエリ文

      ECS インスタンスから NAT ゲートウェイへのパスで、特定のパブリック IP アドレスへのトラフィックを分析します。

      direction: 'in' and srcaddr: 10.0.0.* and dstaddr: 120.26.XX.XX | select -- 特定のパブリック IP アドレスにアクセスする ECS インスタンスのログをフィルタリングします。
      date_format(from_unixtime(__time__ - __time__% 60), '%H:%i:%S') as time, srcaddr,  -- UNIX タイムスタンプを読み取り可能な時間形式に変換します。
      sum(bytes*8/60) as bandwidth  -- バイトをビットに変換し、1 分のキャプチャウィンドウで割って、レートを bit/s で取得します。
      group by time,srcaddr -- 時間と送信元 IP アドレスでグループ化します。
      order by time asc  -- 時間で昇順にソートします。
      limit 100 -- 最初の 100 件の結果を表示します。

      その他の一般的なクエリ

      • NAT ゲートウェイから ECS インスタンスへのパスで、特定のパブリック IP アドレスからすべての ECS インスタンスへのインバウンドトラフィックをフィルタリングします。

        direction: 'out' and dstaddr: 10.0.0.* and srcaddr: 120.26.XX.XX | select -- 特定のパブリック IP アドレスにアクセスする ECS インスタンスのログをフィルタリングします。
        date_format(from_unixtime(__time__ - __time__% 60), '%H:%i:%S') as time,   -- UNIX タイムスタンプを読み取り可能な時間形式に変換します。
        dstaddr,
        sum(bytes*8/60) as bandwidth  -- バイトをビットに変換し、1 分のキャプチャウィンドウで割って、レートを bit/s で取得します。
        group by time,dstaddr -- 時間と宛先 IP アドレスでグループ化します。
        order by time asc  -- 時間で昇順にソートします。
        limit 100 -- 最初の 100 件の結果を表示します。

      • ECS インスタンスから NAT ゲートウェイへのパスで、ECS インスタンスからすべてのパブリック IP アドレスへのアウトバウンドトラフィックをフィルタリングします。

        direction: 'in' and srcaddr: 10.0.0.*  | select -- すべてのパブリック IP アドレスにアクセスする ECS インスタンスのログをフィルタリングします。
        date_format(from_unixtime(__time__ - __time__% 60), '%H:%i:%S') as time,  -- UNIX タイムスタンプを読み取り可能な時間形式に変換します。
        concat(srcaddr,'->', dstaddr), -- 送信元と宛先の IP アドレスを「source_ip->destination_ip」の形式のセッションペアに連結します。
        sum(bytes*8/60) as bandwidth  -- バイトをビットに変換し、1 分のキャプチャウィンドウで割って、レートを bit/s で取得します。
        group by time,srcaddr,dstaddr -- 時間、送信元 IP、宛先 IP でグループ化します。
        order by time asc  -- 時間で昇順にソートします。
        limit 100 -- 最初の 100 件の結果を表示します。

   2. プレビュー

      クエリ結果の可視化方法は、シナリオ 2 と同様です。[面グラフ] タイプを選択し、集計列を [srcaddr] に設定します。

      NAT ゲートウェイへのパスでは、10.0.0.1 (ECS1) からパブリック IP アドレス 120.26.XX.XX へのトラフィックが最も高く、約 12 Kbps です。

Express Connect トラフィック比率の分析

ある企業が Alibaba Cloud リージョンで 2 つの VPC を使用して異なるサービスをデプロイし、オンプレミスデータセンターを Express Connect 回線と Cloud Enterprise Network (CEN) を使用して Alibaba Cloud に接続しています。

IT 部門は、フローログを使用して、VPC 内のさまざまなビジネスアプリケーションからのトラフィックが Express Connect 回線をどのように使用しているかを監視および分析する予定です。この分析は、ネットワークリソースプランニングとパフォーマンス向上に役立ちます。

1. フローログの作成

   2 つのフローログを作成し、ログを同じ Logstore に配信します。各フローログの主要な設定は次のとおりです。

   • リソースインスタンス には、VPC の VPC1 と VPC2 を選択します。

   • サンプリングパス には Transit Router を選択します。

   • 配信設定を SLS に送信 に設定し、2 つのフローログに同じ Logstore を選択し、FlowLog レポート分析機能の有効化 を有効にします。

   • 他のオプションはデフォルト設定のままにします。

2. フローログの分析

   1. クエリ文

      異なる VPC からオンプレミスデータセンターへのトラフィックの比率を分析するには、次のクエリを実行します。

      action: ACCEPT and srcaddr: 192.168.* and dstaddr:10.1.* | 
      WITH 
          vpc1_traffic AS (
              SELECT 
                  date_trunc('minute',__time__) AS minute,
                  SUM(bytes*8/(case WHEN "end"-start=0 THEN 1 else "end"-start end)) AS total_vpc1_traffic
              FROM 
                  log
              WHERE 
                  srcaddr LIKE '192.168.20.%'
              GROUP BY 
                  date_trunc('minute',__time__)
          ),
          vpc2_traffic AS (
              SELECT 
                  date_trunc('minute',__time__) AS minute,
                  SUM(bytes*8/(case WHEN "end"-start=0 THEN 1 else "end"-start end)) AS total_vpc2_traffic
              FROM 
                  log
              WHERE 
                  srcaddr LIKE '192.168.10.%'
              GROUP BY 
                  date_trunc('minute',__time__)
          )
      SELECT 
          COALESCE(vpc1_traffic.minute, vpc2_traffic.minute) AS minute,
          (COALESCE(vpc1_traffic.total_vpc1_traffic, 0) * 100/ NULLIF((COALESCE(vpc1_traffic.total_vpc1_traffic, 0) + COALESCE(vpc2_traffic.total_vpc2_traffic, 0)), 0)) AS vpc1_percentage, 
          (COALESCE(vpc2_traffic.total_vpc2_traffic, 0) * 100/ NULLIF((COALESCE(vpc1_traffic.total_vpc1_traffic, 0) + COALESCE(vpc2_traffic.total_vpc2_traffic, 0)), 0)) AS vpc2_percentage
      FROM vpc1_traffic FULL OUTER JOIN vpc2_traffic ON vpc1_traffic.minute = vpc2_traffic.minute 
      ORDER BY minute

      以下は SQL ステートメントの説明です。

      • フィルター条件：

        • srcaddr: 192.168.*、ソースアドレスが 192.168.* で始まるログをフィルタリングします。

        • dstaddr: 10.1.*、送信先アドレスが 10.1.* で始まるログをフィルターします。

        • action：ACCEPT。action フィールドの値が ACCEPT のログをフィルタリングします。

      • メインクエリ

        • FULL OUTER JOIN を使用して、vpc1_traffic テーブルと vpc2_traffic テーブルを minute フィールドで結合します。

        • 各 VPC の分ごとのトラフィックパーセンテージを計算します。

          • vpc1_percentage は、総トラフィックのうち VPC1 にルーティングされるトラフィックのパーセンテージを指定します。

          • vpc2_percentage は、VPC2 トラフィックが総トラフィックに占める割合を示します。

        • クエリ結果は、minute の昇順でソートされます。

      • WITH サブクエリ：

        SQL ステートメントには、2 つのサブクエリ、vpc1_traffic と vpc2_traffic が含まれています。以下では、vpc1_traffic サブクエリを例に説明します。

        • date_trunc 関数を使用して、UNIX タイムスタンプ (__time__ フィールド) の精度を分単位に下げ、minute という名前を付けます。

        • SUM 関数を使用して、特定の分の総トラフィックレートを bit/s 単位で計算し、total_vpc1_traffic という名前を付けます。

        • 送信元アドレスが 192.168.20.* CIDR ブロック (VPC1 に属する) のトラフィックレコードをフィルタリングします。

        • 結果を分単位でグループ化します。

   2. プレビュー

      クエリ結果の可視化設定は、シナリオ 2 と同様です。[面グラフ] タイプを選択して、時間の経過に伴う異なる VPC のトラフィック比率の傾向を表示します。

      結果は、14:50 から 15:50 の期間中、VPC1 からオンプレミスデータセンターへのトラフィックの割合が高いことを示しています。