Azure Event Hubs で受信したメッセージがその後ちゃんと処理されているかチェックしたい

# はじめに

リアルタイムに大量のデータを受け取って処理できる Azure Event Hubs を触る機会がありました。

Azure Event Hubs で受け取ったメッセージを取得してその情報を基に処理するアプリケーションを作りましたが、すべてのメッセージを取得して処理しているかチェックする術がないかなーと思いました。アプリケーションが出力するログをチェックするのはとてもシンドイので、もっとカンタンな方法はないものかと。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/230264/a252f986-8e23-eb94-60b0-9c381074f81a.png)


# Azure Event Hubs とは

Azure Event Hubs を扱う際は「用語」と「機能」を理解しておく必要があります。理解できていないといろいろ悩んでしまうことがありますので、まずは以下のドキュメントをしっかり読み込みましょう。

https://learn.microsoft.com/ja-jp/azure/event-hubs/event-hubs-about?WT.mc_id=AZ-MVP-5003113

https://learn.microsoft.com/ja-jp/azure/event-hubs/event-hubs-features?WT.mc_id=AZ-MVP-5003113

んー...ムズカシイデス。こんな時は先人たちのチカラを借りましょう。

私は以下のブログと上のドキュメントを行ったり来たり繰り返し読んで、やっと理解しました。ありがとう！

https://blog.nextscape.net/archives/Date/2015/08/eventhubs

https://blog.nextscape.net/archives/Date/2015/08/eventhubs2


# カンタンにチェックするには

Azure Monitor でメトリックやログでチェックできると良いですね。
ということで、以下のドキュメントでどのような監視データがあるのか確認します。

https://learn.microsoft.com/ja-jp/azure/event-hubs/monitor-event-hubs-reference?WT.mc_id=AZ-MVP-5003113

(英語のほうも見てみる)

https://learn.microsoft.com/en-us/azure/event-hubs/monitor-event-hubs-reference?WT.mc_id=AZ-MVP-5003113

メッセージのメトリックの **「受信メッセージ (Incoming Messages)」** と **「送信メッセージ (Outgoing Messages)」** が良さげです。これらは、以下のような情報が収集されます。

| メトリック名 | 説明 | 
|:-|:-|
| 受信メッセージ<br />(Incoming Messages) | 指定された期間に Event Hubs に受信したイベントまたはメッセージの数。 |
| 送信メッセージ<br />(Outgoing Messages) | 指定した期間に Event Hubs から送信されたイベントまたはメッセージの数。 |


## Azure Event Hubs の診断ログを設定

それでは、Azure Monitor にログとメトリックをストリーミングするために、診断ログを設定します。

Azure ポータルで、Azure Event Hubs 名前空間リソースの [監視] カテゴリの [診断設定] を開き、「メトリック」の [AllMetrics] を選択し、「宛先の詳細」で [Log Analytics ワークスペースへの送信] を選択して [Log Analytics ワークスペース] を指定します。そして、[保存] します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/230264/4459d656-46c4-c00f-9ab0-c125e1f6365f.png)


## メトリックログを検索

先ほど指定した [AllMetrics] は「AzureMetrics」テーブルに収集されます。

https://learn.microsoft.com/ja-jp/azure/azure-monitor/reference/tables/azuremetrics?WT.mc_id=AZ-MVP-5003113

以下のようなクエリで、Azure Event Hubs の「受信メッセージ」と「送信メッセージ」を確認します。

```:受信メッセージと送信メッセージを確認
AzureMetrics 
| where MetricName == "OutgoingMessages" or MetricName == "IncomingMessages"
| project TimeGenerated, Resource, ResourceProvider, MetricName, Total
| sort by TimeGenerated desc nulls last 
```

結果は、以下のように出力されます。
受信したメッセージ分、送信されていることがわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/230264/9e358f79-e31d-28df-0488-b015942883cb.png)

ただ、この結果でチェックするには、受信メッセージ、送信メッセージそれぞれの数を合計して比較しなければならないので、まだまだメンドクサイです。

以上のことを踏まえて、チェックするためのクエリをもっと工夫が必要そうです。


## 送受信メッセージの数を比較するためのクエリ

ここでは、指定した期間内の受信メッセージ数の合計と送信メッセージ数の合計を比較する、以下のクエリを用意しました。

```:送受信メッセージの数を比較
AzureMetrics
| where MetricName == "OutgoingMessages" or MetricName == "IncomingMessages"
| extend Total_Incoming_Messages = iif(MetricName == "IncomingMessages", Total, 0.00)
| extend Total_Outgoing_Messages = iif(MetricName == "OutgoingMessages", Total, 0.00)
| summarize sum(Total_Incoming_Messages), sum(Total_Outgoing_Messages)
| extend delta_messages = sum_Total_Incoming_Messages - sum_Total_Outgoing_Messages
```


### クエリの解説

以下より、各行を解説します。
※1 ～ 2 行目は省略

```:3 ～ 4 行目
| extend Total_Incoming_Messages = iif(MetricName == "IncomingMessages", Total, 0.00)
| extend Total_Outgoing_Messages = iif(MetricName == "OutgoingMessages", Total, 0.00)
```

```Total_Incoming_Messages``` (受信メッセージの件数) に、```MetricName``` が ```"IncomingMessages"``` の場合は ```Total``` の値を、そうでない場合は ```0``` をセットします。

(上記と同様に) ```Total_Outgoing_Messages``` (送信メッセージの件数) に、```MetricName``` が ```"OutgoingMessages"``` の場合は ```Total``` の値を、そうでない場合は ```0``` をセットします。

https://learn.microsoft.com/ja-jp/azure/data-explorer/kusto/query/extendoperator?WT.mc_id=AZ-MVP-5003113

https://learn.microsoft.com/ja-jp/azure/data-explorer/kusto/query/iiffunction?WT.mc_id=AZ-MVP-5003113

```:5 行目
| summarize sum(Total_Incoming_Messages), sum(Total_Outgoing_Messages)
```

```Total_Incoming_Messages``` 列、```Total_Outgoing_Messages``` 列、それぞれの合計を算出します。

https://learn.microsoft.com/ja-jp/azure/data-explorer/kusto/query/summarizeoperator?WT.mc_id=AZ-MVP-5003113

https://learn.microsoft.com/ja-jp/azure/data-explorer/kusto/query/sum-aggfunction?WT.mc_id=AZ-MVP-5003113

```:6 行目
| extend delta_messages = sum_Total_Incoming_Messages - sum_Total_Outgoing_Messages
```

最後に、```Total_Incoming_Messages``` 列の合計から ```Total_Outgoing_Messages``` 列の合計を引いた結果を ```delta_messages``` 列にセットします。

期間を指定してこのクエリを実行すると、以下のような結果となります。

これで、アプリケーションが Azure Event Hubs で受け取ったメッセージすべて取得していることがわかりますね。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/230264/3d61a55c-01e5-6a5d-3973-440f12b86db8.png)


## ただし注意点がありました

[前述のメトリックログを検索](#%E3%83%A1%E3%83%88%E3%83%AA%E3%83%83%E3%82%AF%E3%83%AD%E3%82%B0%E3%82%92%E6%A4%9C%E7%B4%A2)するクエリの結果を見てみると、下図のようにたまに送信メッセージの件数が大きいログが検出されることがあります。アプリケーションが出力するログをチェックしてもこのような件数を処理した形跡もありません。

さて、これはなんでしょう？

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/230264/0a806c51-60b8-56ef-a339-19c4f0478ace.png)

そもそも、Azure Event Hubs からメッセージを取得する際、コンシューマーグループを介して配信されます。コンシューマーグループは、Azure Event Hubs インスタンスのパーティション内の古いメッセージから順に読み取り、配信します。

Azure Event Hubs からメッセージを取得するアプリケーションを起動するたびにこのような挙動されると困るので、メッセージを取得する側はどこまで読み取ったか記録するためオフセット値 (パーティション内の位置を示す値) を保持します。これは、アプリケーションが利用する以下のライブラリ (クライアント SDK) が担ってくれています。

https://learn.microsoft.com/ja-jp/azure/event-hubs/sdks?WT.mc_id=AZ-MVP-5003113#client-sdks

これにより、アプリケーションを再起動しても読み取り済み (処理済み) のメッセージを再び処理することなく、処理を再開することができます。

さて、話を戻そう。

上図の **「送信メッセージの件数が大きいログ」** は何かというと、「コンシューマーグループは、Azure Event Hubs インスタンスのパーティション内の古いメッセージから順に読み取り、配信」するので、アプリケーションが利用する **ライブラリが保持しているオフセット値まで読み飛ばしている (加えて、オフセット値後に新たに取得したメッセージ) 分** が「送信メッセージ」としてカウントされたものと思われます。

些細なことの発端から、いろいろ考えたり、調べたりすることによって、より多くの気づきや発見がありました。面白い！