oslo.messagingを使った通知処理の実装
ここからはoslo.messagingが提供するもう一つの「通知（Notification）」機能について具体的な使い方について詳しく解説していきます。
通知機能の用途についてはOpenStackプロジェクトにおいてどのように使われているかの実例を紹介しますが、一般的にどういったケースで使われるかについて、ここで簡単に紹介します。
oslo.messagingの通知機能はもともと、サードパーティのシステムに対して非同期に通知を送る仕組みとして提供されていました。先に紹介したRPCとここで紹介する通知の使い分けとしては、RPCが特定のサービス（OpenStackの場合はコンポーネント）に閉じたケースで使われるのに対して、通知はシステムワイドに共通のイベントを発行し、システム全体における状態を把握する目的で主に使われます。
具体的に、以下のようなECサービスの例でRPCと通知機能のユースケースについて考えてみます。
ここでは各サービスが提供するRPC APIを通してサービス同士が連携し、一つのECサービスを提供していると仮定します。通知機能は、各サービスが利用履歴や検索キーワードなどの統計情報を集積したい場合に役立ちます。
それではサンプルを通じて、通知処理の実装方法について解説していきます。
実装
クライアント実装
まずはクライアントの実装について解説します。oslo.messagingでは、通知を送る主体のことを通知人（Notifier）と呼びますが、ここでは便宜上クライアントと呼びます。以下に、クライアント処理を抜粋します。
7 class NotifyClient(object):< 8 def __init__(self, topic='test_topic', driver='messaging', pub_id='', url=''): 9 transport = oslo_messaging.get_notification_transport(cfg.CONF, url=url) 10 self.notifier = oslo_messaging.Notifier(transport, driver=driver, publisher_id=pub_id, topic=topic)
NotifyClientはoslo.messagingの通知クライアントオブジェクトNotifierのラッパークラスです。ここではRPC clientと同様にTransportオブジェクトを生成した後Notifierオブジェクトを生成します。
以下ではNotifyClientオブジェクトの生成と、メッセージの通知処理を実行しています。
25 client = NotifyClient()
26 client.info({'context': 'foo'}, 'event-hoge', {'hoge': 'abcd'})
NotifyClientオブジェクトのinfoメソッドはoslo_messaging.Notifyオブジェクトのinfoメソッドにパススルーされ、サーバに対して通知が送信されます。その際3つの引数を取り、それぞれ左から順番にcontext、event-typeおよびpayloadを表しています。各パラメータのデータ型とOpenStackで主に設定されている値を以下に示します。
| パラメータ | 型 | 主な設定値 |
|---|---|---|
| context | dict | oslo.contextのRequestContextオブジェクト |
| event-type | string | 通知名を表す文字列 |
| payload | dict | 通知の詳細情報を表すデータ構造 |
contextとpayloadパラメータはそれぞれ同じ型ですが、OpenStackではcontextパラメータには、データをoslo.contextのReqeustContextオブジェクトでラッピングした値を設定する傾向があります。というのも、先述したoslo.messagingのRPC機能は基本的にコンポーネント内のサービス間通信で利用されているのに対して、通知機能は別のコンポーネントから参照される場合があるため、oslo.contextを利用してデータ構造を共通化させています。こうすることで、さまざまなコンポーネントから送信された通知を共通のアルゴリズムで処理することができます。ただサンプルプログラムでは説明の都合上oslo.contextを使わず、単にdict型の値を指定しています。
サーバ実装
次にサーバの実装について解説します。oslo.messagingでは、通知を取得する主体のことを通知リスナ（Notification Listener）と呼んでいますが、ここでは便宜上サーバと呼びます。以下は、サーバ処理の抜粋になります。
26 def start_server(endpoints=[], topic='test_topic', url=''): 27 transport = oslo_messaging.get_notification_transport(cfg.CONF, url=url) 28 targets = [oslo_messaging.Target(topic=topic)] 29 server = oslo_messaging.get_notification_listener(transport, targets, endpoints) 30 31 try: 32 server.start() 33 server.wait()
29行目でget_notification_listerメソッドの呼び出しでサーバオブジェクトを生成し、その際に27～28行目で生成したTransportオブジェクトとTargetオブジェクトをひも付けます。
サーバ側ではRPC同様に通知を受け取るエンドポイントを定義し、サーバオブジェクトにひも付けます。RPCの場合と異なり、通知のエンドポイントはクライアントに値を返却しません。以下にエンドポイントの定義を抜粋します。
6 class HogeEndpoint(object):
7 filter_rule = oslo_messaging.NotificationFilter(event_type='event-hoge')
8
9 def info(self, ctxt, publisher_id, event_type, payload, metadata):
10 print("[HogeEndpoint] ctxt: %s" % (ctxt))
11 print("[HogeEndpoint] publisher_id: %s" % (publisher_id))
12 print("[HogeEndpoint] event_type: %s" % (event_type))
13 print("[HogeEndpoint] payload: %s" % (payload))
14 print("[HogeEndpoint] metadata: %s" % (metadata))
15
16
17 class FugaEndpoint(object):
18 def info(self, ctxt, publisher_id, event_type, payload, metadata):
19 print("[FugaEndpoint] ctxt: %s" % (ctxt))
20 print("[FugaEndpoint] publisher_id: %s" % (publisher_id))
21 print("[FugaEndpoint] event_type: %s" % (event_type))
22 print("[FugaEndpoint] payload: %s" % (payload))
23 print("[FugaEndpoint] metadata: %s" % (metadata))
RPCのサンプルと異なり、通知では2つのエンドポイント（HogeEndpoint、FugaEndpoint）を定義しています。それぞれ、受け取った通知内容について出力に表示するだけのものですが、HogeEndpointにはfilter_ruleが設定されています。これによってエンドポイントが受け取る通知を取捨選択できるようになります。これがどのように機能するかについては、次の節で詳しく解説します。
動作確認
ここでは通知サーバ／クライアントスクリプトを実際に動かし、通知処理の動作を確認していきます。
まずは次のコマンドで通知サーバを起動します。TransportドライバはデフォルトのRabbitMQドライバを利用するため、停止している場合には起動させてから通知サーバを実行してください。
vagrant@vagrant:~$ sudo service rabbitmq-server start vagrant@vagrant:~$ cd oslo-messaging-examples/ vagrant@vagrant:~/oslo-messaging-examples$ src/notifier_server.py
続いて通知クライアントを実行します。別のターミナルを開き、次のコマンドを実行してください。
vagrant@vagrant:~$ cd oslo-messaging-examples/ vagrant@vagrant:~/oslo-messaging-examples$ src/notifier_client.py
すると、サーバ側のターミナルでクライアントが送った通知の中身が表示されます。以下はサーバを実行したターミナルの出力結果になります。
[HogeEndpoint] ctxt: {u'context': u'foo'}
[HogeEndpoint] publisher_id:
[HogeEndpoint] event_type: event-type: hoge
[HogeEndpoint] payload: {u'hoge': u'abcd'}
[HogeEndpoint] metadata: {'timestamp': u'2016-09-26 09:09:50.130070', 'message_id': u'51c23030-8ba8-48e8-bde4-05c21bd9833f'}
[FugaEndpoint] ctxt: {u'context': u'foo'}
[FugaEndpoint] publisher_id:
[FugaEndpoint] event_type: event-type: hoge
[FugaEndpoint] payload: {u'hoge': u'abcd'}
[FugaEndpoint] metadata: {'timestamp': u'2016-09-26 09:09:50.130070', 'message_id': u'51c23030-8ba8-48e8-bde4-05c21bd9833f'}
[FugaEndpoint] ctxt: {u'context': u'bar'}
[FugaEndpoint] publisher_id:
[FugaEndpoint] event_type: event-type: fuga
[FugaEndpoint] payload: {u'fuga': u'efgh'}
[FugaEndpoint] metadata: {'timestamp': u'2016-09-26 09:09:50.150497', 'message_id': u'187c4cd7-3102-48c9-b712-fa05ff67b306'}
HogeEndpointが1度しかよばれていないのに対して、FugaEndpointが2回呼ばれています。これはHogeEndpointに設定したフィルタ（NotificationFilter）によるものです。以下にHogeEndpointで設定されているフィルタの設定値を改めて以下に抜粋します。
6 class HogeEndpoint(object): 7 filter_rule = oslo_messaging.NotificationFilter(event_type='event-hoge')
HogeEndpointでは、送られてくる通知のうちevent_typeパラメータにevent-hogeという値が設定されている通知のみを受け取るようフィルタ設定を行っています。サンプルコードでは、フィルタの設定値に完全一致の文字列を指定しましたが、NotificationFilter内部のフィルタリング処理では正規表現によるマッチング処理を行うためevent_typeパラメータに.*-hogeと指定しても同様の結果が得られます。さらにNotificationFilterではevent_typeパラメータだけでなく、通知のすべてのパラメータ（context, publisher_id, event_type, payload, metadata）に対して同様のフィルタルールを設定できます。
設定したフィルタは、各エンドポイントクラス（HogeEndpointやFugaEndpoint）のクラス変数filter_ruleに設定することでoslo.messagingから参照されます。filter_ruleが設定されていない場合、当該エンドポイントは（FugaEndpointのように）すべての通知を受け付けます。
OpenStack（Ceilometer）の利用例
ここまでoslo.messagingの通知機能の実装方法について見てきました。ここではOpenStackがこの通知機能をどのように利用しているかについて簡単に紹介します。
例えばOpenStackの管理リソースのリアルタイム検索機能を提供しているSearchlightでは、リソースの状態の変更を検知するためにoslo.messagingを利用して、各コンポーネントからの通知を取得します。
Ceilometerでもoslo.messagingの通知機能を利用しています。Ceilometerはリソースの課金を行うための計量（Metering）機能、つまり『だれが』『いつ』『どの』リソースを『どれだけ』利用したかの情報を蓄積し、それらを標準化する機能を提供しています。
具体的には、以下に示すNotification Agent(s)によって通知を集積し、蓄積した通知を整形して外部システムと連携できる仕組みを提供しています。
ここでは、これらの内部の詳しい仕組みの解説は割愛させていただきます。
Ceilometerのアーキテクチャ（出典：System Architecture - Ceilometer）
その他のosloプロジェクト
ここまでoslo.messagingが提供するRPCと通知機能について詳しく解説してきました。ここまでの内容を駆使することでoslo.messagingを十分に使いこなせると思います。これによって、皆さんが開発するアプリケーションのサステーナビリティの向上に寄与できれば幸いです。
最後にoslo.messaging以外のosloプロジェクトについてピックアップしたものを以下に示します。
| プロジェクト名 | 概要 |
|---|---|
| oslo.log | ログの整形、出力先設定などロギング周り全般の機能を提供。OpenStackの各コンポーネント全般で利用されている。 |
| oslo.db | MySQLやPostgreSQLなどのRDBMSに対して共通のインターフェースからアクセスできる仕組みを提供。 |
| cliff | CLIアプリケーションのフレームワーク。コマンドライン引数の解析に加えgitやsvnなどのような多数のサブコマンドを持つCLIアプリケーションを開発する上で役立つ。 |
| stevedore | setuptoolsのentry_pointsで指定されたモジュールの動的な呼び出しをサポートするライブラリ。実行時にロードするライブラリを選択するといったプラグイン機構を提供する多くのOpenStackコンポーネントで利用されている。 |
皆さんもぜひOpenStackプロジェクトの成果物を活用してみてください。
