プロセス連携でつくるイベント駆動システム


こんにちは。デバイスソフトウエア開発部の本間です。

弊社では、様々な IoT 機器を開発・販売しております。 私もこれまで、いくつかの IoT 機器のファームウェア開発に携わってきました。

IoT 機器に求められる機能は、センサデータの収集、異常の検知、クラウドとの連携、機器状態の管理など多岐にわたります。 これらの機能をどのような仕組みで実現するかを考えることは、IoT 機器のシステム設計における面白さの一つだと感じております。

今回は、そのようなシステム設計の一例として、Fluent Bit をイベントルーターとして利用し、複数のプロセスをイベントで連携させる構成を紹介します。

複数プロセスによる設計

組込みシステムと一言でいっても、ハードウェアの構成や性能はシステム毎に様々であり、ソフトウェアの設計もそれらの条件によって大きく変わります。

低リソースだったり、ハードリアルタイム性が求められたりするデバイスでは、ベアメタルや RTOS を採用することも多いかと思います。 一般的なベアメタル環境や多くの RTOS では、汎用 OS のようにユーザープロセスを実行する仕組みを持たないため、 デバイスドライバや制御ロジックなどのプログラムを単一の実行形式へリンクし、その中で関数やモジュール、RTOS のタスクとして責務を分けていく構成となります。

一方、組込み Linux などの汎用 OS が動作する比較的リソースに余裕のあるデバイスでは、プロセスを利用できます。 このようなデバイスでは、センサデータの収集だけでなく、データの監視・記録、ネットワーク通信、ソフトウェア更新など多くの機能を持つことも少なくありません。

そこで、機能ごとにプロセスを分け、それぞれに責務を持たせる設計が選択肢になります。 責務を適切に分けることができれば、機能同士を疎結合にしやすくなり、システム全体として機能追加や保守を行いやすくなります。

ただし、プロセスを分ければシステム全体が自動的に単純になるわけではありません。 それまで関数呼び出しで済んでいた機能間の連携を、今度はプロセス間通信として設計する必要があります。 通信方法やデータ形式など、プロセス間でデータをやり取りするための取り決めが必要になります。

ここからは、組込み Linux などの汎用 OS を搭載した IoT 機器を対象に、プロセス間の連携について考えてみたいと思います。

イベントでプロセスを連携させる

複数のプロセスを連携させる方法は様々です。 UNIX で古くから使われてきたパイプによるフィルタも、強力な手段の1つでしょう。 今回は、その中でもイベント駆動型アーキテクチャ(Event-Driven Architecture、EDA)の考え方を取り入れてみたいと思います。

EDA の考え方では、「ドアを開けた」「ボタンを押した」といった、システム内で起きたことをイベントとして通知し、必要な処理がそのイベントに反応します。 ある処理が別の処理を同期的に呼び出すのではなく、他の処理へ起きたことを非同期に知らせる点が特徴です。

基本的に、EDA は次の3つの要素で構成されます。

構成要素 役割
プロデューサー 出来事をイベントとして送信する
イベントルーター イベントの種類を見て配送先を決める
コンシューマー 受信したイベントを処理する

プロデューサーは最終的な配送先を知る必要がありません。コンシューマーを追加する場合も、プロデューサーを変更せず、イベントルーター側で配送先を追加できます。Pub/Sub 方式の MQTT で言えば、トピックへの publish がイベントの送信、トピックへの subscribe がコンシューマーの追加(イベント購読)に相当すると言えます。

EDA の考え方は、クラウド上のサービス間だけでなく、デバイス内のプロセス間連携にも適用できます。 本記事では、その一例として、イベントルーターを介して複数のプロセスを連携させる構成を作ってみたいと思います。

Fluent Bit をイベントルーターとして使う

本記事でイベントルーターとして使うのが Fluent Bit です。

Fluent Bit は、ログや各種データを収集・処理・転送できる軽量なツールです。テレメトリーエージェントというツールに分類されるそうです。Input・Filter・Output などを組み合わせてデータパイプラインを構成できます。Input するデータにはタグを付けることができ、そのタグを使って配送先を振り分けられます。

Fluent Bit は、MQTT や NATS のようなメッセージングシステムでも、ZeroMQ のようなメッセージングライブラリでもありません。一方で、ルーティングやフィルタリング、バッファリングなど、イベント配送にも利用できる機能を備えています。

IoT 機器のようにプロセス構成が比較的固定されているシステムであれば、イベントの配送経路の設定を各プロセスから切り離し、Fluent Bit の設定へ集約できます。今回は、この特徴を利用して、ホスト内のイベントを事前に設定した宛先へ配送するイベントルーターとして使ってみます。

そのような理由から本記事では Fluent Bit を使ってみますが、イベント連携の用途では、一般的には次のような技術がよく使われるようです。

技術・仕組み 向いているケース
ZeroMQ ブローカーレスの高速なプロセス間通信
MQTT サーバ・デバイス間の通信、QoS による配送制御
NATS 高速なプロセス間通信、JetStream による永続化・再配送

Fluent Bit でシステムを組んでみる

それでは Fluent Bit で簡単なイベント駆動システムを組んでみましょう。 温度の計測、温度の監視、LED の制御、データの記録をそれぞれプロセスに分け、Fluent Bit を介して連携させます。

サンプルは Linux 上での実行を想定しており、Fluent Bit 5.1.1、Python 3.10 で動作確認しています。温度センサと GPIO の読み書きについては、サンプルを実行しやすいように模擬実装とします。

システム構成

各プロセスの役割は次のとおりです。

プロセス 役割
temp-sensor ADC から値を1秒ごとに読み取り、温度データを生成する
monitor 温度を監視し、高温警報状態と LED の設定指示を送信する
led LED の設定指示を受けて GPIO 出力へ反映する
recorder 最新の温度と高温警報状態を30秒ごとに保存する

各プロセスは他のプロセスへ直接データを送らず、Fluent Bit を経由してイベントを送受信します。 例えば、temp-sensor が送信した温度イベントは、Fluent Bit によって monitor と recorder の両方へ配送されます。

プロセス間でやり取りするイベントは次の3つです。

イベント タグ プロデューサー 配送先
温度計測 sensor.temperature temp-sensor monitorrecorder
高温警報状態 alert.high-temperature monitor recorder
LED設定 led.set monitor led

Fluent Bit の設定

fluent-bit.yaml を次のように作成します。

Inputs の Forward Input は、127.0.0.1:24224 でイベントを受信します。listen の既定値は 0.0.0.0 なので、別のホストから接続されないよう、ループバックアドレスを明示しています。

Outputs では、タグに応じて各 Output へルーティングし、UDP Output からレコードを JSON Lines 形式で各コンシューマーへ送信します。sensor.temperaturemonitorrecorder へ、led.setled へ、alert.high-temperaturerecorder へ配送します。最後の File Output では、デバッグ用にすべてのイベントを /tmp/fluent-debug.log へ記録します。

設定ファイルは次のコマンドで検証できます。

Forward Protocol によるイベント送信

プロデューサーから Fluent Bit へのイベント送信には、Python の fluent-logger を使います。まず、fluent-logger パッケージをインストールします。

Fluent Bit の Forward Input が受け取るデータは、Forward Protocol に従った MessagePack です。FluentSender.emit() で送信するイベントは、Message mode に従った次の形式になります。

例えば、FluentSender("sensor") に対して emit("temperature", record) を呼ぶと、タグは sensor.temperature になります。

UDP によるイベント受信

Fluent Bit からコンシューマーへの送信には、UDP Outputjson_lines 形式を使っています。そのため、コンシューマー側では JSON レコードを1件ずつ簡単に受信できます。

受信処理は json_lines_server.py にまとめます。受信したデータを行ごとに JSON として読み込むだけのシンプルな処理です。

各プロセスを実装する

それでは、temp-sensormonitorledrecorder の各プロセスを実装していきます。

温度計測プロセス(temp-sensor

まず、temp_sensor.py で模擬温度を1秒ごとに送信します。

なお、このサンプルでは、実際の温度センサーからの ADC 入力を SAMPLES で模擬しています。実機では、IIO などを使って ADC の値を読み取ります。

状態監視プロセス(monitor

monitor.py は温度が30 ℃以上かどうかを判定し、高温警報状態が変わったときだけ alert.high-temperatureled.set を送信します。

LED 設定プロセス(led

led.py は led.set を受信し、状態が変わったときだけ LED を更新します。

なお、このサンプルでは、実際の LED を制御する GPIO 操作をログ出力で代用しています。実機では、libgpiod などで GPIO を操作します。

記録プロセス(recorder

最後に、recorder.py で最新の温度と高温警報状態をまとめ、30秒ごとに /tmp/record-data.jsonl へ保存します。イベント識別用の event は状態の更新にだけ使い、保存するレコードからは除きます。

動作確認

Fluent Bit と各プロセスを、別々のターミナルで起動します。

それでは、temp-sensormonitorled などのログを確認してみましょう。

temp-sensor では、次のように温度を1秒ごとに送信しています。

monitor では、温度に応じて高温警報状態が切り替わっています。

led では、高温警報状態に応じた ON/OFF 指示を受け、LED 状態が切り替わっています。

recorder のデータ記録内容は次のようになります。30秒周期でデータ記録されていることが確認できます。

動作確認中に発生した全てのイベントは、デバッグファイルから確認できます。

このように、各プロセスに責務を分け、Fluent Bit を利用して簡単なイベント駆動システムを構築することができました。

まとめ

今回は、組込み Linux などの汎用 OS 上で機能を複数のプロセスに分け、イベントで連携させる構成を作りました。複数プロセスに分けることで責務を分けやすくなる一方、プロセス間の連携方法についても設計する必要があります。

本記事では Fluent Bit をイベントルーターとして利用してみました。イベントの配送経路を各プロセスから切り離して Fluent Bit の設定にまとめることで、各プロセスは自身の処理に集中できる構成にしています。

サンプルの UDP による配送はベストエフォートですが、TCP Output や Forward Output永続バッファリングなどを利用して、配送の信頼性を高めることもできます。ただし、これらはコンシューマーでの処理完了まで保証するものではありません。より高度な配送保証や動的な購読、イベントの保存・再生などが必要な場合は、MQTT や NATS なども含め、要件に合った仕組みを選ぶことが重要です。