チュートリアル: Dapr と MQTT ブローカーを使用してイベントドリブンアプリを構築する

[アーティクル]
11/20/2024

このチュートリアルでは、Dapr アプリケーションをクラスターにデプロイします。 Dapr アプリケーションは、MQTT ブローカーに発行されたシミュレートされた MQTT データを使用し、ウィンドウ関数を適用して、結果を MQTT ブローカーに発行します。公開された出力は、メッセージの頻度とサイズを減らすために、エッジで大量のデータを集計する方法を表します。 Dapr アプリケーションはステートレスであり、MQTT ブローカー状態ストアを使用して、ウィンドウの計算に必要な過去の値をキャッシュします。

Dapr アプリケーションは、次の手順を実行します:

センサーデータの sensor/data トピックをサブスクライブします。
このトピックのデータを受信すると、MQTT ブローカー状態ストアに転送されます。
10 秒ごとに、状態ストアからデータをフェッチし、過去 30 秒にタイムスタンプが付いたセンサーデータの最小、最大、平均、中央値、および 75 パーセンタイルの値を計算します。
30 秒を超えるデータは状態ストアから有効期限が切れています。
結果は JSON 形式で sensor/window_data トピックに発行されます。

Note

このチュートリアルでは、生の MQTT を使用して公開やサブスクライブできるようにする Dapr CloudEvents を無効にします。

前提条件

Azure IoT Operations のインストール - クイックスタート: K3s を使用して GitHub Codespaces で Azure IoT Operations を実行する
MQTT ブローカー Dapr コンポーネントのインストール - MQTT ブローカー Dapr コンポーネントをインストールする

Dapr アプリケーションをデプロイする

この時点で、Dapr アプリケーションをデプロイできます。コンポーネントを登録しても、パッケージ化された関連バイナリはコンテナーにデプロイされません。アプリケーションとともにバイナリをデプロイするには、[デプロイ] を使用して、コンテナー化された Dapr アプリケーションと 2 つのコンポーネントを一緒にグループ化します。

まず、次の定義を使用する yaml ファイルを作成します。

コンポーネント	説明
`volumes.mqtt-client-token`	MQTT ブローカーおよび状態ストアによる Dapr プラグイン可能なコンポーネントの認証に使用される SAT
`volumes.aio-internal-ca-cert-chain`	MQTT ブローカー TLS 証明書を検証するための信頼チェーン
`containers.mq-event-driven`	事前ビルド済みの Dapr アプリケーションコンテナー。

app.yaml という名前のファイルに次のデプロイ yaml を保存します。

apiVersion: v1
kind: ServiceAccount
metadata:
  name: dapr-client
  namespace: azure-iot-operations
  annotations:
    aio-broker-auth/group: dapr-workload
---    
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mq-event-driven-dapr
  namespace: azure-iot-operations
spec:
  selector:
    matchLabels:
      app: mq-event-driven-dapr
  template:
    metadata:
      labels:
        app: mq-event-driven-dapr
      annotations:
        dapr.io/enabled: "true"
        dapr.io/inject-pluggable-components: "true"
        dapr.io/app-id: "mq-event-driven-dapr"
        dapr.io/app-port: "6001"
        dapr.io/app-protocol: "grpc"
    spec:
      serviceAccountName: dapr-client

      volumes:
      # SAT token used to authenticate between Dapr and the MQTT broker
      - name: mqtt-client-token
        projected:
          sources:
            - serviceAccountToken:
                path: mqtt-client-token
                audience: aio-internal
                expirationSeconds: 86400

      # Certificate chain for Dapr to validate the MQTT broker
      - name: aio-ca-trust-bundle
        configMap:
          name: azure-iot-operations-aio-ca-trust-bundle

      containers:
      - name: mq-event-driven-dapr
        image: ghcr.io/azure-samples/explore-iot-operations/mq-event-driven-dapr:latest

次のコマンドを実行して、アプリケーションをデプロイします。
```
kubectl apply -f app.yaml
```
アプリケーションが正しくデプロイされたことを確認します。ポッドは、次のコマンドで示すように、しばらくしてからすべてのコンテナーの準備ができたことを報告するはずです。
```
kubectl get pods -l app=mq-event-driven-dapr -n azure-iot-operations
```
次の出力を使用します。
```
NAME                         READY   STATUS              RESTARTS   AGE
mq-event-driven-dapr         3/3     Running             0          30s
```

シミュレーターをデプロイする

Kubernetes ワークロードをデプロイしてテストデータをシミュレートします。 sensor/data トピックの MQTT クライアントを使用して、サンプル温度、振動、圧力の読み取り値を定期的に MQTT ブローカーに送信することで、センサーをシミュレートします。

Explore IoT Operations リポジトリからシミュレーターをデプロイします。

kubectl apply -f https://raw.githubusercontent.com/Azure-Samples/explore-iot-operations/main/tutorials/mq-event-driven-dapr/simulate-data.yaml

シミュレーターが正常に実行中であることを確認します:

kubectl logs deployment/mqtt-publisher-deployment -n azure-iot-operations -f

次の出力を使用します。

Get:1 http://deb.debian.org/debian stable InRelease [151 kB]
Get:2 http://deb.debian.org/debian stable-updates InRelease [52.1 kB]
Get:3 http://deb.debian.org/debian-security stable-security InRelease [48.0 kB]
Get:4 http://deb.debian.org/debian stable/main amd64 Packages [8780 kB]
Get:5 http://deb.debian.org/debian stable-updates/main amd64 Packages [6668 B]
Get:6 http://deb.debian.org/debian-security stable-security/main amd64 Packages [101 kB]
Fetched 9139 kB in 3s (3570 kB/s)
...
Messages published in the last 10 seconds: 10
Messages published in the last 10 seconds: 10
Messages published in the last 10 seconds: 10

MQTT クライアントをデプロイする

MQTT ブリッジが動作していることを確認するには、クラスターに MQTT クライアントをデプロイします。

次の client.yaml という名前の新しいファイルで、クライアントのデプロイを指定します。

apiVersion: v1
kind: ServiceAccount
metadata:
  name: mqtt-client
  namespace: azure-iot-operations
---
apiVersion: v1
kind: Pod
metadata:
  name: mqtt-client
  namespace: azure-iot-operations
spec:
  serviceAccountName: mqtt-client
  containers:
  - image: alpine
    name: mqtt-client
    command: ["sh", "-c"]
    args: ["apk add mosquitto-clients mqttui && sleep infinity"]
    volumeMounts:
    - name: mqtt-client-token
      mountPath: /var/run/secrets/tokens
    - name: aio-ca-trust-bundle
      mountPath: /var/run/certs/aio-internal-ca-cert/
  volumes:
  - name: mqtt-client-token
    projected:
      sources:
      - serviceAccountToken:
          path: mqtt-client-token
          audience: aio-internal
          expirationSeconds: 86400
  - name: aio-ca-trust-bundle
    configMap:
      name: azure-iot-operations-aio-ca-trust-bundle

kubectl を使用してデプロイファイルを適用します。
```
kubectl apply -f client.yaml
```
次の出力を確認します。
```
pod/mqtt-client created
```

Dapr アプリケーション出力を確認する

Mosquitto クライアントポッドに対してシェルを開きます。
```
kubectl exec --stdin --tty mqtt-client -n azure-iot-operations -- sh
```
Dapr アプリケーションからの公開出力を確認するには、sensor/window_data トピックをサブスクリプションしてください。
```
mosquitto_sub -L mqtt://aio-broker/sensor/window_data
```

アプリケーションが次のようなセンサーのスライディングウィンドウの計算を出力していることを 10 秒ごとに確認します。

{
    "timestamp": "2023-11-16T21:59:53.939690+00:00",
    "window_size": 30,
    "temperature": {
        "min": 553.024,
        "max": 598.907,
        "mean": 576.4647857142858,
        "median": 577.4905,
        "75_per": 585.96125,
        "count": 28
    },
    "pressure": {
        "min": 290.605,
        "max": 299.781,
        "mean": 295.521,
        "median": 295.648,
        "75_per": 297.64050000000003,
        "count": 28
    },
    "vibration": {
        "min": 0.00124192,
        "max": 0.00491257,
        "mean": 0.0031171810714285715,
        "median": 0.003199235,
        "75_per": 0.0038769150000000003,
        "count": 28
    }
}

オプション ‐ Dapr アプリケーションを作成する

このチュートリアルでは、Dapr アプリケーションの事前構築済みコンテナーを使用します。コードを自分で変更してビルドする場合は、次の手順に従います:

前提条件

Docker - アプリケーションコンテナーのビルド用
コンテナーレジストリ - アプリケーションコンテナーのホスト用

アプリケーションのビルド

Explore IoT Operations リポジトリを複製します。

git clone https://github.com/Azure-Samples/explore-iot-operations

Dapr チュートリアルディレクトリに移動します。
```
cd explore-iot-operations/tutorials/mq-event-driven-dapr/src
```

Docker イメージをビルドします:

docker build docker build . -t mq-event-driven-dapr

Kubernetes クラスターでアプリケーションを使用するには、Azure Container Registry などのコンテナーレジストリにイメージをプッシュする必要があります。 minikube や Docker などのローカルコンテナーレジストリにプッシュすることもできます。
```
docker tag mq-event-driven-dapr <container-alias>
docker push <container-alias>
```
app.yaml を更新して、新しく作成したイメージをプルします。

トラブルシューティング

アプリケーションが起動しない場合や、CrashLoopBackoff にコンテナーがある場合は、多くの場合、daprd コンテナーログに役立つ情報が含まれます。

次のコマンドを実行して、daprd コンポーネントのログを表示します。

kubectl logs -l app=mq-event-driven-dapr -n azure-iot-operations -c daprd

次のステップ

MQTT ブローカーと Azure Event Grid の間で MQTT データをブリッジする

次の方法で共有

チュートリアル: Dapr と MQTT ブローカーを使用してイベントドリブンアプリを構築する

前提条件

Dapr アプリケーションをデプロイする

シミュレーターをデプロイする

MQTT クライアントをデプロイする

Dapr アプリケーション出力を確認する

オプション ‐ Dapr アプリケーションを作成する

前提条件

アプリケーションのビルド

トラブルシューティング

次のステップ

フィードバック

その他のリソース

次の方法で共有

チュートリアル: Dapr と MQTT ブローカーを使用してイベントドリブン アプリを構築する

前提条件

Dapr アプリケーションをデプロイする

シミュレーターをデプロイする

MQTT クライアントをデプロイする

Dapr アプリケーション出力を確認する

オプション ‐ Dapr アプリケーションを作成する

前提条件

アプリケーションのビルド

トラブルシューティング

次のステップ

フィードバック

その他のリソース

チュートリアル: Dapr と MQTT ブローカーを使用してイベントドリブンアプリを構築する