KMC活動ブログ

京大マイコンクラブの活動の様子を紹介します！！

AS59128 のフロー情報収集と Amazon Athena での分析

こんにちは、 id:sora_h です。これは KMC Advent Calendar 2023 12 日目の記事です (大遅刻)。

KMC ではインターネット接続手段の 1 つとして AS59128 を 2017 年頃より運用して、部室内のサーバーや一部の部員が利用しています。これまでフロー情報の収集は行ってきませんでしたが、今年、フロー情報の統計を収集して分析を可能にしたため、その実装を軽く紹介します。地味に pmacctd のドキュメントが難解だったので…。

経緯

AS59128 は運用初期から複数のトランジットやピア、東西に跨った複数拠点が存在していますが、外部の経路由来の障害や性能劣化についての調査はフロー情報なしで実施していて、特に変化前のトラフィックを確認することがそれなしでは難しくエスパーを繰り返してました。

このままでは障害時の対応が手探りで安定運用に支障がある、また (D)DoS 等異常なトラフィックが起きた場合の対応も困難である、という理由でフロー分析が出来る低い運用・金銭的コストで分析できるようにする手段があれば… というところで保留していたところ、意外に簡単に発見されたので実装しました。

実装

AS59128 のトラフィック量は低いのもあり、低コストで調達できる Linux ルータ (Ubuntu, BIRD) で運用しています。なので 、ルータ上で直接ある程度の処理ができると考え、下記の構成に:

  1. pmacctd
  2. fluent-bit
  3. Amazon Kinesis Data Firehose
  4. Amazon S3
  5. AWS Glue Catalog (& Crawler)
  6. Amazon Athena

無視できるほどゴミのようなデータ量 であれば金銭的にも特に気にする必要はないし、サーバーフルとして運用しないといけないのも pmacctd, fluent-bit 程度でお手軽構成なんじゃないと思います。

pmacctd

先述のように全ての BGP ルータが Linux マシンであるため、pmacct のうち pmacctd を利用して pcap から直接フローログを生成しています。ある程度サンプリングレートを引き上げれば CPU もそこまで消費しません。また、基本的にしょぼい SATA SSD を利用していることから、出力先は tmpfs です。ディスク壊れてほしくなさすぎる。

BGP 経路情報から ASN を取得するためローカルの bird へ接続する設定を入れています。この他、各種 map ファイルは peer_as_src の生成で必要になりますが、これは bird.conf の生成でも利用しているデータからルータごとに自動生成されます。プロビジョン時の自動生成によってある程度の面倒さは解消しました *1。ifindex を自動生成して interfaces.map と bgp_peer_src_as.map でピア ASN を設定しています。bgp_agent.map は bgp_ip=127.0.0.1 ip=0.0.0.0/0 で固定です。

また、集計頻度はそもそもの最終的な処理データ量に直結するし、データが細かすぎても困るため 300s で運用しています。tmpfs から古いデータの削除は systemd-tmpfiles にやってもらってます。

この設定でだいたい RAM 消費量が 500MB ちょっと。おそらくだいたいが BGP 経路の影響。

# pmacctd.conf 抜粋
plugins: print

print_output: json
print_refresh_time: 300
print_history: 300s
print_history_roundoff: m
print_output_file_append: true
print_output_file: /dev/shm/pmacct/%Y-%m-%d-%H.log
print_latest_file: /dev/shm/pmacct/current

pcap_interfaces_map: /etc/pmacct/interfaces.map
pcap_direction: in
pcap_ifindex: map
sampling_rate: 80

pmacctd_as: bgp
pmacctd_net: bgp
bgp_daemon: true
bgp_agent_map: /etc/pmacct/bgp_agent.map
bgp_daemon_ip: 127.0.0.1
bgp_daemon_port: 17917
bgp_peer_src_as_type: map
bgp_peer_src_as_map: /etc/pmacct/bgp_peer_src_as.map
# bird.conf 抜粋
protocol bgp bgp_mon_pmacctd
{
  local 127.0.0.1 as AS_SELF;
  neighbor 127.0.0.1 port 17917 as AS_SELF;
  rr client;

  ipv4 {
    preference 1;
    table t4_bgp;
    igp table t4_igp;
    import none;
    export filter
    {
      bgp_next_hop = 0.0.0.0; /* ←実際にはここにダミーの内部 IP アドレス */
      if ((AS_SELF, C_ROUTE_DEFAULT) ~ bgp_community) then reject;

      if (bgp_is_ospf_route()) then {
        bgp_path.prepend(64512);
        accept;
      }
      if (bgp_is_exportable()) then accept;
      reject;
    };
  };

  /* ipv6 も似たような設定なので省略 */
}

fluent-bit

pmacctd は tmpfs に出力していることから分かるように最終保存先はローカルのストレージではないです。したがって fluent-bit の in_tail で回収して外部へ送信します。

pmacct print plugin の JSON 出力でタイムスタンプを unix timestamp で表示するようにしても string になってしまうので、parser filter で integer に加工するなどのステップを挟んだ。このために正規表現エンジン動いてるのちょっと無駄な感じする。

[FILTER]
  Name parser
  Match flowlog.pmacctd
  Parser stamp_inserted
  Key_Name stamp_inserted
  Reserve_Data True

[FILTER]
  Name parser
  Match flowlog.pmacctd
  Parser stamp_updated
  Key_Name stamp_updated
  Reserve_Data True
[PARSER]
  Name pmacctd
  Format json
  Time_Key stamp_inserted
  Time_Keep true
  Time_Format %s
  Time_Offset +0000

[PARSER]
  Name stamp_inserted
  Format regex
  Regex ^(?<stamp_inserted>\d+)$
  Types stamp_inserted:integer

[PARSER]
  Name stamp_updated
  Format regex
  Regex ^(?<stamp_updated>\d+)$
  Types stamp_updated:integer

Kinesis Data Firehose

データの送信先には Amazon Kinesis Data Firehose を使ってみました。jsonl 形式で fluent-bit から S3 に書き出してもこの規模なら対した問題はないと思うところ、Firehose を通すと AWS マネージドでバッファを抱えて Parquet に落としてくれるので便利 *2。一般論として Athena や Redshift Spectrum を通して S3 からデータを読む時、ある程度オブジェクトが細切れになっているより固まっていたほうがパフォーマントで、Standard-IA や Glacier に沈める時もオーバーヘッドを気にする必要がないという利点があり扱いやすい。

もっとも、この規模なら jsonl 形式でも良いし、Glacier に沈めるほど長期保存する(べき)ではないし、試しておきたかったという側面での採用という感じ。適当な規模のデータなら安価に使えるけど、がっつり業務で使うとそこそこお金はかかりそうな雰囲気。

なお、fluent-bit に持たせている AWS 資格情報は static credentials です。そもそもオンプレ上というのと、どのみち PutEvents しか出来ないのでリスクは低いはず *3。

[OUTPUT]
  name  kinesis_firehose
  match flowlog.pmacctd
  region ap-northeast-1
  delivery_stream as-flowlog
  time_key time
  time_key_format %Y-%m-%dT%H:%M:%SZ
# 抜粋
resource "aws_kinesis_firehose_delivery_stream" "flowlog" {
  name        = "as-flowlog"
  destination = "extended_s3"

  extended_s3_configuration {
    role_arn   = aws_iam_role.firehose.arn
    bucket_arn = aws_s3_bucket.flowlog.arn

    prefix             = "firehose/data/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
    compression_format = "UNCOMPRESSED" # Parquet の場合 Parquet format レイヤで圧縮されるので destination レイヤは uncompressed にしないといけない

    error_output_prefix = "firehose/error/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/errtype=!{firehose:error-output-type}/"

    buffer_interval = 900
    buffer_size     = 64

    data_format_conversion_configuration {
      input_format_configuration {
        deserializer {
          open_x_json_ser_de {}
        }
      }

      output_format_configuration {
        serializer {
          parquet_ser_de {
            compression    = "SNAPPY"
            writer_version = "V1" # default
          }
        }
      }

      schema_configuration {
        role_arn      = aws_iam_role.firehose.arn
        database_name = aws_glue_catalog_table.flowlog.database_name
        table_name    = aws_glue_catalog_table.flowlog.name
      }
    }
  }
}

Glue Data Catalog と Crawler

前述の schema_configuration でも利用されているように, AWS の各種 Analytics 系サービスと連携しようとするとスキーマを Glue で管理する必要があり、さらにデータの置き場所とパーティションの情報も Glue に入れる必要がある。Firehose の Parquet の設定から Athena のパーティションの設定まで一箇所にデータカタログとして放り込んでおけば後はいい感じになるという点では便利ですね。細分化しすぎて building block の極みという感じするけど。

resource "aws_glue_catalog_database" "flowlog" {
  name = "as_flowlog"
}
resource "aws_glue_catalog_table" "flowlog" {
  database_name = aws_glue_catalog_database.flowlog.name
  name          = "flowlog"

  table_type = "EXTERNAL_TABLE"

  parameters = {
    EXTERNAL              = "TRUE"
    "parquet.compression" = "SNAPPY"
  }

  partition_keys {
    name = "year"
    type = "string"
  }
  partition_keys {
    name = "month"
    type = "string"
  }
  partition_keys {
    name = "day"
    type = "string"
  }

  storage_descriptor {
    input_format  = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat"
    output_format = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"

    location = "s3://${bucket}/firehose/data/"

    ser_de_info {
      serialization_library = "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"
      parameters = {
        "serialization.format" = 1
      }
    }

    columns {
      name = "time"
      type = "timestamp"
    }
    # columns  以下省略
  }
}

Crawler

Firehose は新規に増えたパーティションのことをケアしてくれないというか、Firehose は S3 キーのテンプレートからの生成はしてくれるけどパーティションにそれがどう利用されるかは関知しないので、Glue Catalog に増えたパーティションを別途反映する必要がある。Glue には crawler 機能があって source (S3) にあるオブジェクトからパーティション情報をメンテナンスしてくれるので、これを毎日仕掛けておく。

ただこれは都度全部 prefix を読んで全パーティションの情報を再生成してくれるので遅いし日次用途ならやや高め。もっと安価にやるなら Lambda で毎日一定時刻でコード書いてパーティション追加するのがいいと思うけど、コード投げるとランタイムの deprecation の面倒とかを見ないといけないので一旦無視してる。recrawl_behavior = "CRAWL_NEW_FOLDERS_ONLY" が確かなにかの設定とコンフリクトしたんだったはず…。

resource "aws_glue_crawler" "flowlog" {
  database_name = aws_glue_catalog_database.flowlog.name
  name          = "flowlog"
  role          = aws_iam_role.glue.name
  schedule      = "cron(0 1 * * ? *)"

  configuration = jsonencode(
    {
      Version = 1
      Grouping = {
        TableGroupingPolicy = "CombineCompatibleSchemas"
      }
    }
  )

  catalog_target {
    database_name = aws_glue_catalog_database.flowlog.name
    tables = [
      "flowlog",
    ]
  }

  lake_formation_configuration {
    use_lake_formation_credentials = false
  }

  lineage_configuration {
    crawler_lineage_settings = "DISABLE"
  }

  recrawl_policy {
    recrawl_behavior = "CRAWL_EVERYTHING"
  }

  schema_change_policy {
    delete_behavior = "LOG"
    update_behavior = "UPDATE_IN_DATABASE"
  }
}

Athena で分析

ここまで設定が終わって無事に Glue Crawler がパーティション情報を足してくれていれば何もせずに Athena に反映されていて分析ができます。具体的なデータには言及できませんが、AS 安定運用のために必要最小限の範囲で収集とクエリを行います。データに関しても、一定期間を置いて破棄するように設定されています。

実例

BGP の仕組みを考えれば当たり前で、経路テーブルで把握できるのはあくまでも自 AS からの outbound, first hop がどのピアになるかで、実際に inbound, 戻りのトラフィックがどう辿ってくるかは実際にフロー情報を確認しないと分からないです。特に同じ AS と複数の接続があると、他 AS の経路制御の影響を受けるためどの接続を辿ってきているかは確実に分からないし、複数のトランジットがあると当然のように戻りのパケットは別の経路でやってきている…ということは当たり前のように起きます。これまではエスパーする他なく、この手の問題が発生した時に調査する手がかりが増えたのは大きい。

今年は実際に一部の inbound traffic がこちらからの first hop とは別の AS から届いていてその経路が不調と思われたので当該 AS が吸い込まなくする制御を行ったり*4、何故か一部の Web サイトだけタイムアウト率が高いためフローを確認したり…とトラブルシュートに役立ったシーンがあった。

わたしは普段からサーバー用途以外でも AS59128 で暮らしていて、自分の日常の一例だと AS17685 とのトラフィックが特にパケットロス率とレイテンシに センシティブで、 実際にこれで不調を感じて問題の調査を行うことがしばしばあります。

AWS へのログイン

今回 KMC でちゃんと AWS サービスを利用するのは初だったため、ログイン手段もついでに整備を行いました。

root アカウントについては payer account を握るわたししかアクセスを許可しないのは前提で、必要に応じて https://github.com/sorah/himari と https://github.com/sorah/himari2amc で AWS アクセスキーやコンソールへアクセスできるようにしています。himari は omniauth で任意の認証手段から自由に ID token のクレームを制御して OIDC プロバイダとして動作し、himari2amc が ID token を元に AWS の資格情報を取得します。himari に GitHub ログインを設定して、AS59128 に関わる最低限のメンバーにアクセスをつけています。Terraform moduleで全部サーバーレスってやつのデプロイが出来て、外部のアカウントをそのまま使わせることができるので小規模団体におすすめです。

なお、フロー情報は pmacctd の時点で基本個人が特定できない形で収集しています。ただ、一部その限りではないため *5、中期的なデータや分析システムを部内の root と呼ばれる権限持ちに通常の管理者権限があるからといってそのまま開示するのは避けたかった背景があります。そのような理由もあり部内で構築せず独立して権限管理ができる環境として AWS を選定した側面もあります。

コスト

  • Kinesis Firehose: 0.09 USD/day
  • Glue: 0.05 USD/day
  • S3: 0.01 USD/day
  • Athena: クエリ量に依存

ということで 4.5 USD/month 前後で運用できています。この中だと Glue crawler がやっぱり安くはなくてコスト最適化できるポイントではある。

まとめ

AS59128 におけるフロー情報の統計データ収集について解説しました。これ以外の用途でも Kinesis Data Firehose を使うと雑にある程度粒度にまとめた Parquet オブジェクトを生成できてとりあえず Athena でクエリ可能な状態にできるという解説でもあるので、何らかの役にたってたらうれしいです。

*1:itamae と sorah/hocho で構成管理してます

*2:Firehose の Parquet サポート (2018) 以前は Athena で select insert 的な感じで変換するとか、自前で何か動かして変換するくらいしかなかった

*3:別の用途で Vault や step-ca で CA とクライアント証明書組みたいという気持ちは若干ある

*4:トランジット AS でその先の細かな経路制御ができる制御用コミュニティが提供されているとありがたい

*5:接続手段によってはその限りではないので統計データの取得については実験・非営利ネットワークという前提もあるので理解してもらっている