Security Insightsは、すべてのCloudflareアカウントに対して実行可能なセキュリティ推奨を提供します。これらのインサイトを検出するために、すべてのアカウント、ゾーン、およびDNSレコードを定期的にスキャンし、潜在的なセキュリティリスクや設定ミスを探しています。しかし、2つの主要な課題が明らかになりました。まずスキャン頻度が低すぎたことです。スキャンは週に1回か2回しか行われておらず、新たに導入されたセキュリティリスクが最大で2週間検出されない可能性がありました。次に、自動スキャンが多くの無料プランアカウントでオプトイン式だったため、多数のアカウントがまったくスキャンされていませんでした。自動化攻撃が加速する中で、スキャンが稀または存在しないことのリスクは高まっています。すべてのお客様に対してこれらの問題を確実に発見することは、「より良いインターネットを構築する」という我々の目標にとって非常に重要です。
我々は、スキャン頻度を上げ、すべてのアカウントで自動スキャンを有効にするには、平均してスキャンスループットを約10倍に増やす必要があると算出しました — 10 scans/秒 から 100 scans/秒へ。しかし、当時のシステムは既に負荷で苦しんでおり、数百万件のイベントがバックログに溜まり、APIは頻繁にタイムアウトし、プロセスがクラッシュしていました。システムを修正し、スケールさせる必要がありました。
以下は、Security Insightsのスキャンスループットを10倍以上に引き上げ、数百万の顧客に対してセキュリティインサイトの自動化を有効にし、すべての顧客のスキャン頻度を倍増させた取り組みの記録です。
どのようにセキュリティインサイトをスキャンしているか
高レベルでは、自動スキャンはスケジューラによってトリガーされます。アカウントやゾーンがスキャン対象になると、スケジューラはApache Kafkaにメッセージ（または複数のメッセージ）を公開します。これらのメッセージは複数のチェッカー（特定のアセットや設定をスキャンする専門のGoマイクロサービス）にファンアウトします。各メッセージについて、各チェッカーは検出した結果（セキュリティインサイト）を内部APIに送信し、内部APIはそれらをPostgresデータベースに永続化します。
スケールさせるために行ったこと
Scaling Kafka
Apache Kafkaは厳密にはキューではなく、パーティション化されたイベントストリームです（最近ではキュー的なセマンティクスも得ています）。各パーティション内では、メッセージは順序通りに消費・処理されなければなりません。一般的なキューでは消費順序は保たれても処理は並び替えられることがあるのに対し、Kafkaではそうではありません。その結果、consumer group内の各パーティションに対してアクティブなコンシューマは1つだけしか持てません。これには2つの影響があります:
- 処理に時間がかかるメッセージが次のメッセージへの進行をブロックする
- 各チェッカーについて、作成できるコンシューマ数はパーティション数を上限とする（各チェッカーは独自のconsumer groupを持つ）
パーティションを増やしてスケールしようとすることも可能でしたが、Kafkaブローカー自体（多くの他サービスと共有している）のリソース使用量が増えるため、最終手段として残し、まずはコードとアーキテクチャの改善を目指しました。
並列処理の導入
メッセージは順序で消費しなければなりませんが、同時に複数のメッセージを消費することは妨げられません。チェッカーを変更して、メッセージをバッチで消費し、各メッセージを別々のgoroutineで処理するようにしました。トレードオフとしては、プロセスがバッチ途中でクラッシュした場合にやり直す作業が増えることと、メモリ使用量が少し増えることですが、我々のケースではこれらは許容できるものでした。
ヘッドオブライン（head-of-line）ブロッキングの回避
一部のチェッカーが処理するメッセージの中には、他より圧倒的に処理に時間がかかるものがあります。たとえば、あるアカウント／ゾーンが非常に多くのアセットを持っている場合などです。最悪の場合、平均が数秒またはミリ秒であるのに対し、そのようなメッセージは数分〜数時間かかることがあります。簡潔なアプローチとして、consumer groupとチェッカーを「slow lane」と「fast lane」の2つに分割しました。メッセージが速く処理できるか遅いかを素早く判定し、fast lane のチェッカーが遅いメッセージに出会った場合はそれをスキップするようにしました。これにより、遅いメッセージは専用のリソースで処理され最小限の遅延で済み、速いメッセージは高速度を維持できます。
データベースクエリの最適化
検出した各インサイトはPostgresに書き込まれます。これはチェッカーがインサイトのリストを渡す単一のAPIエンドポイントで処理されます。実装は当初次のようになっていました:
for _, issue := range issues {
_, err = tx.Exec(ctx, `INSERT INTO table ... VALUES ($1, $2, ...) ON CONFLICT DO UPDATE ...`, ...)
if err != nil {
return err
}
}
ご明察の通り、大きなインサイト集合ではこのコードはインサイトごとにデータベースへの往復を行い、最大で観測されたサイズ500,000の場合、単一のAPIコールで50万回の往復、クエリ、トランザクションが発生していました。最初はPostgresでのバルク挿入のゴールドスタンダードであるCOPYを一時テーブルに対して使ってみましたが、Postgresのシステムテーブルのbloat（膨張）を引き起こすことが分かりました。
最終的にハイブリッドアプローチを採用しました:
- issuesの数が閾値未満の場合はUNNESTを使用
- 閾値を超える場合はCOPYを使用
これにより、巨大なセットに対しては秒単位での挿入が可能になり、小さなセットに対してはミリ秒単位でのさらに高速な挿入が可能になりました。
APIタイムアウトの調査
スケールさせようとする中で、内部APIにいくつか奇妙な振る舞いが見られました:
- 大量のリクエストがクライアント側のタイムアウトを引き起こす
- 多くのチェッカーが単一のAPIコールに対して処理時間の20〜90%を費やしている
- 多量のスキャンをトリガーすると、スループットが高めに始まりその後劣化する
これらの問題の根本原因は全てレイテンシでした。プライマリデータベースはポートランド（オレゴン）にありましたが、APIはポートランドとアムステルダムでactive-active構成で稼働していました。光速でもポートランドとアムステルダム間の往復レイテンシは約50msになります。そのため、アムステルダムのAPIインスタンスからのデータベースクエリははるかに時間がかかり、クライアント側のコネクションプールの接続を長時間占有してしまいました。大量のAPIリクエストが来ると、コネクションプールはすぐに枯渇し、空き接続を待つタイムアウトが発生していました。ポートランドでは平均APIコールが10msで完了したのに対し、アムステルダムではほぼ3秒かかっていました。
ではなぜメッセージスループットが低下したのでしょうか。各チェッカープロセスはKafkaストリームのパーティション群を割り当てられて消費します。APIはロードバランスされており、各プロセスは処理のライフサイクルを通じて一つのAPI接続を保持します。その結果、あるプロセスはアムステルダムのAPIに接続し、別のプロセスはポートランドのAPIに接続することになりました。ポートランドに紐づくパーティションは速やかに処理されましたが、アムステルダムに紐づくものは遅延し、結果としてKafkaのlagがパーティションごとに偏ってしまいました。ロードバランサがトラフィックを均等に分配していたため、この問題は顕著になりました。
簡単な対処として、APIをactive-passiveに切り替え、アクティブなAPIがプライマリデータベースに従うようにしました。これによりレイテンシ問題は一夜で解消しました。
スケジューラの再設計
Kafkaをスケールし、データベースクエリを最適化し、APIの問題を修正しましたが、まだ解決すべき問題が残っていました：スキャンが時間的に概ね均等に分散されることを保証する必要がありました。一度にすべてのスキャンをキューに入れることはできませんでした。なぜなら我々のKafkaトピックは時間ベースの保持ポリシーを使っているため、スキャンがKafka内に積み上がり、処理される前に削除されてしまうからです。元のスケジューラはスキャンを均等に分散するのが不得手で、トリガーされるスキャン数は時間によってスパイクし予測不可能でした。週の特定のタイミングでは、数十万件のスキャンが数分のうちにトリガーされてしまうことがありました。
スケジューラは固定の再発期間でスキャンをトリガーしていました（擬似コード）:
Loop forever:
Find accounts where last_scheduled_at + scanning frequency <= now
For each account:
Trigger scan for account
Trigger scan for all zones in the account
Update last_scheduled_at = now
last_scheduled_atが多くのアカウントで類似していたためデータベース内に不均一性があり、これが不均一性の一因でした。しかし、完全に均等に分布していても、スキャン頻度を上げるとこの問題は拡大します。例えばスキャン頻度を15日ごとから7日ごとに変えると、アカウントの約53%が同時にスキャン対象になります。
もう一つの問題は、一部のアカウントが非常に大量のゾーンを持っている場合、そのアカウントがスケジュールされるとすべてのゾーンに対して一斉にスキャンが発生し、Kafkaパーティションを飽和させ、小規模アカウントのスキャンが遅延することでした。
これらの問題を解決するため、我々は3つの主要な変更を加えました:
- ゾーンをアカウントから独立してスケジュールする：各ゾーンに独自のlast_scheduled_atフィールドを持たせる
- 既存のアカウントとゾーンのlast_scheduled_at時刻をランダム化する（この過程でスキャンが遅延しないように注意）
- スキャンスケジューリングに対して適応的なレート制限を導入する
ゾーンを独立してスケジュールすることは大きなアカウント問題の明白な解決策です。last_scheduled_atをランダム化することで既存データベースの不均一性を解消できます。適応的レート制限はもう少し興味深いアイデアです。レート制限により、スキャン頻度を変更した際に発生するスパイクを平滑化できます。例えばスキャン頻度を7日に変更し、アカウント数が5,000万件ある場合、約83 scans/秒のレート制限を設定すれば7日間に均等に広げられます。しかしもし1,000万件のアカウントを追加した場合、その固定レートは全体のスキャンに8日を要することになります。そこで適応的にする利点が出てきます：レート制限は半時間ごとに非同期で、アカウントとゾーンの総数およびスキャン頻度に基づいて再計算されます。これにより、数千〜数百万のアカウントやゾーンをオンボードしても、スキャンを時間通りに続けられます。
次の関数はその計算の例です:
func computeRate(free, pro, biz, ent int64) rate.Limit {
r := float64(free)/freeScanInterval.Seconds() + float64(pro)/proScanInterval.Seconds() + float64(biz)/bizScanInterval.Seconds() + float64(ent)/entScanInterval.Seconds()
// Guard against zero counts. We always want to schedule at least one scan per second.
if r < 1 {
r = 1
}
r *= rateLimitBufferFactor
return rate.Limit(r)
}
現在の状況
これらの修正により、チェッカーごとの7日移動平均スループットは時間を通じて10倍以上に増加しました。改善前はおよそ10 scans/秒で実行していましたが、目標の100 scans/秒との差は大きく見えました。リソースを増やす、Kafkaのパーティションを増やす、あるいはアーキテクチャ全体を投げ捨てるといった話も出ましたが、これらの修正が決定的な効果をもたらしました。現在、Security Insightsはピークスケジューリング時に120 scans/秒以上を維持しており、10倍改善目標を上回っています。内部APIはタイムアウトしなくなり、Kafkaのlag指標も健全になりました。
これらのスケーラビリティ改善により、無料アカウントとゾーンに対して自動スキャンを有効化し、すべての顧客のスキャン頻度を引き上げることができました:
- Free: every 7 days
- Pro and Business: every