OpenTelemetryトレーシングの統合

OpenTelemetryトレーシングは、分散システム内でのリクエストの流れをトレースするための仕様であり、リクエストが分散システムをどのように流れるかを追跡し、リクエストのパフォーマンスや動作を可視化・分析することができます。MQTTのシナリオでは、この概念を利用してMQTTメッセージ送信の異なる参加者（パブリッシャー - MQTTサーバー - サブスクライバー）間のリクエストをトレースできます。

「トレースコンテキスト」は、複数のシステムやサービスにまたがるリクエストやトランザクションを追跡・識別するために分散トレーシングで使用される仕組みです。W3C Trace Context MQTTドキュメントでは、この概念がMQTTプロトコルに適用され、MQTTメッセージ送信の異なる参加者間でリクエストを追跡可能にしています。これにより、システム管理者や開発者はメッセージがシステム内をどのように流れるかを理解できます。

EMQXはトレースコンテキストを伝播する機能を内蔵しており、分散トレーシングシステムにシームレスに参加できます。この伝播は、メッセージのパブリッシャーからサブスクライバーへtraceparentおよびtracestateのユーザープロパティを単に転送することで実現されます。EMQXがアプリケーションメッセージをクライアントに転送する際、トレースコンテキストの整合性を保持し、変更なく送信します。この方法はMQTT仕様3.3.2.3.7に完全準拠しており、トレースデータの送信における一貫性と信頼性を保証します。

注意

User-PropertyはMQTT 5.0で導入されたため、EMQXはMQTT 5.0を使用している場合にのみトレースコンテキストを抽出・伝播できます。

MQTT 5.0未対応のクライアントの場合、EMQXのTraces All Messagesオプションを有効にする必要があります。EMQXは内部の分散トレーシングのためにメッセージに自動的にトレースIDを追加します。

OpenTelemetry分散トレーシングを利用することで、EMQXのシステム管理者や開発者はIoTアプリケーションのパフォーマンスや動作をリアルタイムで監視・分析できます。問題発生時の迅速な検出と解決が可能です。

本ページでは、EMQXとOpenTelemetryトレーシングの統合方法を紹介し、OpenTelemetry Collectorのセットアップ、EMQXでのOpenTelemetryトレース統合の有効化・設定、トレーシングスパンの過負荷管理について説明します。

OpenTelemetryトレースを直接Dynatraceにエクスポートする方法は、Integrate OpenTelemetry with Dynatraceをご覧ください。

OpenTelemetry Collectorのセットアップ

EMQXとOpenTelemetryトレースを統合する前に、OpenTelemetry Collectorをデプロイ・設定し、可能であればOpenTelemetry対応のオブザーバビリティプラットフォーム（例：Jaeger）を用意します。以下はデプロイと設定の手順です。

1. OpenTelemetry Collectorの設定ファイルotel-trace-collector-config.yamlを作成します。

   receivers:
     otlp:
       protocols:
         grpc:
   
   exporters:
     otlp:
       endpoint: jaeger:4317
       tls:
         insecure: true
   
   processors:
     batch:
   
   extensions:
     health_check:
   
   service:
     extensions: [health_check]
     pipelines:
       traces:
         receivers: [otlp]
         processors: [batch]
         exporters: [otlp]

2. 同じディレクトリにDocker Composeファイルdocker-compose-otel-trace.yamlを作成します。

   version: '3.9'
   services:
     jaeger:
       image: jaegertracing/all-in-one:1.51.0
       restart: always
       ports:
         - "16686:16686"
   
     otel-collector:
       image: otel/opentelemetry-collector:0.90.0
       restart: always
       command: ["--config=/etc/otel-collector-config.yaml", "${OTELCOL_ARGS}"]
       volumes:
         - ./otel-trace-collector-config.yaml:/etc/otel-collector-config.yaml
       ports:
         - "13133:13133" # Health check extension
         - "4317:4317"   # OTLP gRPC receiver
       depends_on:
         - jaeger

3. Docker Composeでサービスを起動します。

   docker compose -f docker-compose-otel-trace.yaml up

4. 起動後、OpenTelemetry CollectorはホストのデフォルトgRPCポート（4317）で待機し、JaegerのWEB UIは http://localhost:16686 でアクセス可能です。

EMQXでのOpenTelemetryトレーシングの有効化

このセクションでは、EMQXでOpenTelemetryトレーシングを有効にし、マルチノード構成での分散トレーシング機能を確認する手順を説明します。

opentelemetry.exporter.endpointには1つのURLを指定します。URLはhttpまたはhttpsスキームで、明示的なポート番号を含む必要があります。例：http://localhost:4317は有効ですが、localhost:4317やhttp://localhostは無効です。

1. EMQXのcluster.hoconファイルに以下の設定を追加します（EMQXがローカルで動作している場合の例）。

   opentelemetry {
     exporter { endpoint = "http://localhost:4317" }
     traces {
      enable = true
      # 全メッセージをトレースするかどうか
      # メッセージからトレースIDが抽出できない場合は新しいトレースIDが生成されます。
      # filter.trace_all = true
    }
   }

   また、ダッシュボードのManagement -> MonitoringにあるIntegrationタブからOpenTelemetryトレース統合を設定することも可能です。

2. 例えば、ノード名がemqx@127.0.0.1とemqx1@127.0.0.1の2ノードクラスタを起動し、分散トレーシング機能を確認します。

3. 異なるノード・ポートでMQTTX CLIクライアントを使い、同じトピックにサブスクライブします。

   • emqx@127.0.0.1ノード（デフォルトMQTTリスナー1883ポート）で：

     mqttx sub -t t/trace/test -h localhost -p 1883

   • emqx1@127.0.0.1ノード（1884ポートリスナー）で：

     mqttx sub -t t/trace/test -h localhost -p 1884

4. 有効なtraceparentユーザープロパティを含むメッセージをトピックにパブリッシュしてトレースコンテキスト付きメッセージを送信します。

   mqttx pub -t t/trace/test -h localhost -p 1883 -up "traceparent: 00-cce3a024ca134a7cb4b41e048e8d98de-cef47eaa4ebc3fae-01"

5. 約5秒後（EMQXのトレースデータエクスポートのデフォルト間隔）、JaegerのWEB UI（http://localhost:16686）にアクセスしてトレースデータを確認します。

   • emqxサービスを選択し、Find Tracesをクリックします。emqxサービスがすぐに表示されない場合は数秒後にページをリロードしてください。メッセージのトレースが表示されます。

     

   • トレースをクリックすると詳細なスパン情報とトレースタイムラインが表示されます。

     

この例では、EMQXは2種類のスパンをトレースしています。

• process_messageスパンは、EMQXノードがPUBLISHパケットを受信・解析した時点で開始し、メッセージがローカルのサブスクライバーに配信されるか、アクティブなサブスクライバーがいる他ノードに転送されるまで継続します。各スパンは1つのトレースされたパブリッシュメッセージに対応します。

• send_published_messageスパンは、トレースされたメッセージがサブスクライバーの接続制御プロセスで受信され、送信パケットがシリアライズされて接続ソケットに送信されるまでの間に開始・終了します。アクティブなサブスクライバーごとに1つのsend_published_messageスパンが生成されます。

トレーシングスパンの過負荷管理

EMQXはトレーシングスパンを蓄積し、定期的にバッチでエクスポートします。エクスポート間隔はopentelemetry.trace.scheduled_delayパラメータで制御され、デフォルトは5秒です。

バッチ処理のトレーススパンプロセッサには過負荷保護機構が組み込まれており、蓄積可能なスパン数の上限（デフォルト2048スパン）を超えると新しいスパンは破棄されます。この上限は以下の設定で変更可能です。

opentelemetry {
  traces { max_queue_size = 2048 }
}

max_queue_sizeの上限に達すると、現在のキューがエクスポートされるまで新しいトレーシングスパンは破棄されます。

注意

トレースされたメッセージが非常に多くのサブスクライバーに配信される場合（max_queue_sizeの値を大幅に超える場合）、エクスポートされるスパンはごく一部に限られ、多くのスパンは過負荷保護により破棄されることが予想されます。

max_queue_sizeの増加はパフォーマンスやメモリ消費に影響を与えるため、慎重に行ってください。