OpenCTIコネクタメッセージングにおける自動バックプレッシャー制御
OpenCTIにおけるコネクタの管理は、スムーズなデータ取り込みを維持し、特にRabbitMQのシステム過負荷を回避するために重要です。バージョン6.2.12以前は、コネクタの実行間隔の管理が複雑で直感的でなく、さまざまなタイプの間隔が存在するためエラーが発生しやすい状況でした。現在、すべての「External Import」コネクタに対して、このプロセスを自動化および簡素化する新しいスケジューリング機能が導入されました。
この記事では、この新機能、関連する環境変数、そしてそれがもたらすメリットについて詳しく解説します。
スケジューラの目的
機能概要
スケジューラワークフローの全体図
スケジューラにより、コネクタの実行をより効率的に管理できます。コネクタが実行予定になると、最初の実行を除き、スケジューラはコネクタプロセスを開始する前にRabbitMQに蓄積されたメッセージのサイズを最初にチェックします。このサイズが環境変数queue_thresholdで定義されたサーバー容量を超えると、コネクタは「Buffering」モードに切り替わり、環境変数duration_periodに従って実行を延期します。
この機能は、キューで発生する可能性のある過負荷を防ぐために不可欠であり、手動介入なしでコネクタがよりスムーズにデータを取り込めるようにします。さらに、OpenCTIインターフェースには、「Last run」、「Next run」、「Server capacity」などのより正確な情報が表示されるようになり、これらはpingAliveを通じて40秒ごとに更新されます。
新しい環境変数
この機能を強化するために、2つの新しい環境変数が導入されました:
duration_period: ISO 8601形式を使用して実行期間を指定します。queue_threshold: コネクタが「Buffering」モードに切り替わるキューのしきい値(MB単位)を定義します。デフォルトでは、このしきい値は500 MBに設定されています。
これらの変数により、環境の特定のニーズに基づいてスケジューラの動作を調整できます。
この機能が対象とするコネクタのタイプは?
この新機能は、「External Import」コネクタにのみ適用されます。これらのコネクタは、外部APIからデータをインポートするために定期的なスケジューリングが必要ですが、他のタイプのコネクタは定期的な再実行を必要としません。
さまざまなタイプのコネクタについて詳しく知りたい方は、ドキュメントをご覧ください: https://docs.opencti.io/latest/deployment/connectors/
なぜこれが重要なのか?
主要なユースケース
新しいスケジューリング機能は、各コネクタのデータ取り込みを綿密に監視する必要があるプラットフォーム管理者にとって特に重要です。実際、コネクタは非常に短時間でOpenCTIに大量のデータを送信する可能性があります。このプロセスが適切に管理されない場合、重大な問題につながる可能性があります。
ユースケース#1: 当初、プラットフォームチームはキューを常時監視し、問題のあるコネクタを手動で特定する必要がありました。RabbitMQのキューの過負荷を避けるために、ワーカーが残りのデータを処理できるようにしながら、これらのコネクタを(状態をリセットせずに)停止する必要がありました。このアプローチは機能的ではありましたが、多くの時間とリソースを消費するため面倒でした。
ユースケース#2: コネクタがデータの取り込みに苦労し、適時に停止されない場合、コネクタの間隔が短すぎるとRabbitMQキューが満杯になり続け、最終的にホストシステムの物理メモリが完全に飽和し、最悪のシナリオの1つに陥ります。この状況はRabbitMQでエラーやクラッシュを引き起こす可能性があり、チームは過負荷のキューを手動でクリアしてRabbitMQを再起動せざるを得なくなります。この介入により、すべての保留中のデータが失われ、コネクタの状態をリセットする必要があります。その後、チームは可能な限り近い状態を手動で復元し、コネクタを再起動し、さらなる過負荷を防ぐために取り込みを綿密に監視する必要があります。
解決策: 「Scheduler」の追加により、プラットフォーム管理者はすべてのコネクタを常時監視する必要がなくなりました。事前定義されたしきい値に達すると、コネクタは自動的に「Buffering」モードに切り替わります。これにより、重い手動介入や潜在的にリスクの高い作業なしに、RabbitMQキューのより効率的で安全な管理を保証しながら、大幅な時間とリソースを節約できます。
どのように機能するのか?
コネクタ実行間隔の管理は、2つの異なる方法で標準化および簡素化されました:
schedule_iso() (破壊的変更あり)
schedule_iso()メソッドはISO 8601形式を使用し、コネクタ実行間隔の標準化された正確なスケジューリングを可能にし、ユーザーにとってより良い可読性を提供します。例えば、「P1D」は1日の期間を表し、「PT24H」は24時間の期間を示します。このアプローチは現在、すべての新しいコネクタ統合の標準となっており、コネクタ実行間隔の管理における明確性と一貫性を保証します。
schedule_iso()が機能するには、message_callbackとduration_periodの2つの引数が必要ですmessage_callback: コネクタプロセスの開始に相当します。duration_period: その値はISO 8601形式である必要があり、対応する環境変数が設定に存在する必要があります。この国際的に認められた形式により、期間の一貫した解釈が保証され、将来の統合が読みやすくなり、エラーが最小限に抑えられます。
実装例
config.ymlまたはdocker-compose.ymlの場合:
Copied!
1# config.yml2duration_period: "PT5H" # この変数を追加3queue_threshold: 600 # デフォルト値(500 MB)と異なる場合に追加する変数
Copied!
1# docker-compose.yml2- CONNECTOR_DURATION_PERIOD=PT5H # この変数を追加3- CONNECTOR_QUEUE_THRESHOLD=600 # デフォルト値(500 MB)と異なる場合に追加する変数
コネクタコードの場合:
Copied!
1# コネクタ設定を取得2self.duration_period = get_config_variable(3 "CONNECTOR_DURATION_PERIOD", ["connector", "duration_period"], config4)
Copied!
1def run(self):2 # schedule_iso()メソッドを使用3 # message_callbackでは、コネクタプロセス(ここではself.starter)を指定する必要があります4 # 標準化されたISO 8601形式を考慮した新しいduration_period5 self.helper.schedule_iso(message_callback=self.starter, duration_period=self.duration_period) # duration_period: "PT5H" => 5時間
schedule_unit() (破壊的変更なし)
schedule_unit()メソッドを使用すると、時間や分などの時間単位を指定することで、既存のコネクタ間隔を再利用できます。このアプローチにより、コネクタや進行中の操作に即座に影響を与えることなく、新しいスケジューリングシステムへの段階的な移行が容易になりますが、このメソッドは時間の経過とともに段階的に廃止される予定です。
schedule_unit()が機能するには、message_callback、duration_period、time_unitの3つの引数が必要ですmessage_callback: コネクタプロセスの開始に相当します。duration_period:schedule_iso()とは異なり、duration_periodは既存の間隔(例:interval_hours=6)を使用するため、環境変数に含める必要はありません。通常、これは数値です。time_unit: この引数はduration_periodが使用する時間単位を指定します。例では、time_unitを「HOURS」に設定する必要があります。
実装例
schedule_iso()とは異なり、このメソッドは既存のコネクタ間隔変数を使用するため、config.ymlやdocker-compose.ymlを設定する必要はありません。
コネクタコードの場合:
Copied!
1def run(self):2 # schedule_unit()メソッドを使用3 # message_callbackでは、コネクタプロセス(ここではself.starter)を指定する必要があります4 # duration_periodは既存のコネクタ間隔を再利用します(ここではself.interval_hours)5 # time_unitは後方互換性のため - 有効な列挙型(YEARS, WEEKS, DAYS, HOURS, MINUTES, SECONDS)6 self.helper.schedule_unit(message_callback=self.starter, duration_period=self.interval_hours, time_unit=self.helper.TimeUnit.HOURS)
スケジューラに関する追加情報
以前は、コネクタはwhile trueループとtime.sleep(self.interval)を使用してコード内で直接実行間隔を管理しており、間隔は設定でカスタマイズ可能でした。現在、これらすべての要素は削除され、コネクタ内の「Run & Terminate」の実装も同様です。この管理は完全に「Scheduler」によって処理されます。
ユーザーインターフェースでの表示
「Run & Terminate」モードのコネクタの動作と表示
「Run & Terminate」モードでのスケジューラの機能図
コネクタが「Run & Terminate」モードで設定されている場合、「Next run」の値は「External schedule」で定義されます。このモードを選択する組織は通常、外部スケジューリングを使用して実行間隔を管理し、コネクタの実行サイクルの管理に柔軟性を持たせています。
コネクタが「Run & Terminate」モードの場合のユーザーインターフェースの概要
注: 特別なケースが実装されています。設定内のduration_periodが、例えば0、「0」、「P0D」、「PT0S」などに設定されている場合、コネクタは「Run & Terminate」モードと同じ動作を示します。
「Buffering」モードのコネクタの詳細表示
RabbitMQキューの容量が定義されたしきい値を超えた場合(例えば、queue_message_sizeが9.90 MBでqueue_thresholdが8 MBに設定されている場合)、コネクタは自動的に「Buffering」モードに切り替わります。
「Buffering」モードでは、キューの容量が指定されたしきい値を下回るまで、コネクタの実行が一時停止されます。ユーザーインターフェースには、警告メッセージや「Server Capacity」セクションの色の変化など、この状態変更を示す視覚的なインジケータが表示されます。
コネクタが「Buffering」モードの場合のユーザーインターフェースでの視覚的アラートの概要
スケジューラを使用しないコネクタの詳細表示
スケジューラをサポートしないコネクタは、ユーザーインターフェースに「Not Provided」と表示されます。
「Scheduler」を使用せず、状態を持たないコネクタ
注: 状態にlast_runが存在し、タイムスタンプ形式の場合、自動的に変換され、「Last run」タイトルの横に「(from State)」という表記が表示され、この情報がスケジューラからではなく状態から来ていることをユーザーに示します。
標準のUnixタイムスタンプの場合
浮動小数点タイムスタンプの場合
まとめ
OpenCTIバージョン6.2.12で導入された新しいスケジューリング機能(Scheduler)は、「External Import」コネクタに対して、より標準化され、安全で、自動化された管理を提供します。RabbitMQキューの過負荷や手動介入に関連するリスクを軽減し、ユーザーがより価値の高いタスクに集中するための貴重な時間を解放します。
schedule_iso()およびschedule_unit()メソッドにより、ユーザーは環境の特定のニーズに基づいてコネクタの実行を調整できると同時に、新機能への段階的な移行が可能になります。schedule_iso()メソッドが現在推奨されており、schedule_unit()メソッドはスケジューラがない状態からschedule_iso()の使用への移行を容易にするためにのみ存在することに注意することが重要です。
この改善により、ユーザーインターフェースでコネクタ実行プロセスの「Last run」、「Next run」、「Server Capacity」を含む、より理解しやすく直感的な概要も提供されます。この新機能は、「external-import」テンプレートで、またはスケジューラをすでに統合しているコネクタ(CISA KEV、Mandiant、SEKOIA、AlienVault、Recorded Future、CrowdStrike (v6.3))で既にテストできます。
この記事が、この新機能によって提供されるメリットと可能性をより深く理解するのに役立つことを願っています。OpenCTIに関するより多くのヒントやアドバイスに興味がありますか? Slackコミュニティに参加して、他のユーザーとつながり、アイデアやソリューションを共有しましょう。
続きを読む
関連トピックとインサイトをご覧ください
