メインコンテンツまでスキップ

Sparkplug B

PlantPulse Edge の Sparkplug B パブリッシャー(outbound) + コマンドレシーバー(inbound) マニュアル。標準 SPB 1.0 トピックで NBIRTH / DBIRTH / DDATA / NDEATH をパブリッシュし、ホスト(SCADA) が送信する NCMD / DCMD を処理してタグ書き込みおよび rebirth までサポートします。SCADA / HiveMQ / Ignition などと互換性があります。

項目値
ステータスPhase 1 (BIRTH/DATA/DEATH) + Phase 3 (NCMD/DCMD インバウンド) 有効。Phase 2 (alias) はトグルでオプション
コントローラパッケージplantpulse.app.edge.component.queue.transfer.sparkplug
デフォルトsparkplug.enable=true (運用有効)
コマンド受信sparkplug.cmd.enable=true のとき NCMD/DCMD 購読 + 処理
ブローカーエッジの HiveMQ (mqtt.server.* 再利用)
認証MQTT username/password (app.properties の mqtt.server.user / mqtt.server.password)。2026.05+ install.sh がボックスごとにランダムパスワードを自動生成。HiveMQ broker の auth.properties は container-entrypoint が毎回起動時に app.properties の mqtt.* と自動 sync — オペレーターは app.properties のみ変更
2026.05+ コンテナモードのシークレット

MQTT アカウントは MQTT_USER / MQTT_PASSWORD_FILE または MQTT_PASSWORD で注入します。 外部 SCADA (Ignition / HiveMQ Edge など) が Sparkplug でこのボックスに接続する場合は、オペレーターが管理する secret ファイルからパスワードを確認して使用します。

トピックスキーマ​

spBv1.0/{group_id}/(N|D){BIRTH|DEATH|DATA|CMD}/{edge_node_id}[/{device_id}]
項目マッピング
group_idEdgeContext.site_id (例: SITE_00001)。sparkplug.group.id で override 可能
edge_node_idEdgeContext.id (例: EDGE_00303)。sparkplug.edge.node.id で override 可能
device_idopc_id (例: OPC_UA_Kepware) — sparkplug.device.strategy=per-opc
metric nametag_id

メッセージ種類​

メッセージトピックペイロードの核心パブリッシュタイミング
NBIRTHspBv1.0/{g}/NBIRTH/{n}bdSeq metric、seq=0接続直後
DBIRTHspBv1.0/{g}/DBIRTH/{n}/{d}該当 OPC の全タグ metric (initial null)、seq 増加接続直後 OPC ごと
DDATAspBv1.0/{g}/DDATA/{n}/{d}単一 metric (tag_id、value、datatype、ts)、seq 増加キューの Point 1件到着時
NDEATHspBv1.0/{g}/NDEATH/{n}bdSeq metric、seq なし (スペック)MQTT will (異常終了) / disconnect 時に明示パブリッシュ
NCMDspBv1.0/{g}/NCMD/{n}host → edge ノードコマンド (例: Node Control/Rebirth=true)host のユーザーアクション / タイムアウト / state sync
DCMDspBv1.0/{g}/DCMD/{n}/{d}host → edge デバイス metric write (target metric name または alias + new value)host が OPC の特定タグ値を setpoint として変更するとき

ライフサイクル / シーケンス​

  1. bdSeq — 0 ~ 255 wrap、ディスク永続化 (./data/sparkplug/bdseq デフォルト)。セッション開始ごとに +1。
  2. seq — 0 ~ 255 wrap。NBIRTH で 0 から開始、DBIRTH/DDATA ともにパブリッシュ時に 1 ずつ増加。
  3. NDEATH = MQTT Will — CONNECT 時に will として登録され、異常終了時に broker が自動パブリッシュ。正常終了時は disconnect() が明示パブリッシュ。

データタイプマッピング​

APP_TAG.data_type (自由文字列) → MetricDataType (大文字小文字無視、trim)。

edgeSPB
Float / SingleFloat
DoubleDouble
Byte / Int8Int8
Short / Int16Int16
Int / Integer / Int32Int32
Long / Int64Int64
UInt8/16/32/64UInt8/16/32/64
Boolean / BoolBoolean
String / TextString
DateTime / DateDateTime
その他 / nullString (fallback、log.warn)

Point.value は String で保存されているため、SPB metric value に入れる前に上記タイプにキャストします。 キャスト失敗時は metric 値を null でパブリッシュします。

設定 (src/app.properties)​

# --------------------------------------------------------------------------------------
# SPARKPLUG B
# 브로커 접속은 mqtt.server.* 를 그대로 재사용한다.
# --------------------------------------------------------------------------------------
sparkplug.enable=false
sparkplug.group.id=                  # 비우면 site_id
sparkplug.edge.node.id=              # 비우면 EdgeContext.id
sparkplug.device.strategy=per-opc    # per-opc | single (현재 per-opc 만 구현)
sparkplug.qos=0                      # DDATA 빈번 → 0 권장
sparkplug.bdseq.path=                # 비우면 ./data/sparkplug/bdseq
sparkplug.alias.enable=true          # Phase 2 토글 (현 무영향)
sparkplug.cmd.enable=false           # Phase 3 토글 (현 무영향)

構成要素​

ファイル役割
SparkPlugQueueTransfer.javaQueueTransfer 実装。init/trans/close のみ公開
SparkPlugSession.javaライフサイクル (CONNECT+will / NBIRTH / DBIRTH / DDATA / disconnect-NDEATH / seq / bdSeq)
SparkPlugConfig.javaプロパティ → 設定オブジェクト (ブローカーは mqtt.server.* 再利用)
SparkPlugTopicBuilder.javaトピック文字列ビルダー
SparkPlugDataTypeMapper.javadata_type → MetricDataType + value coerce
SparkPlugBdSeqStore.javabdSeq ディスク永続化 (0 ~ 255 wrap)
PointQueueProcessorFactory.javatransfer_list に SparkPlugQueueTransfer 追加

依存 jar​

WebContent/WEB-INF/lib/ に追加:

jar役割
tahu-core-1.0.14.jarEclipse Tahu Sparkplug B ペイロードモデル + エンコーダー
protobuf-java-3.25.5.jartahu の protobuf 依存
commons-compress-1.27.1.jartahu の commons-compress 依存
lib 同期

WEB-INF/lib の jar はビルド時に Nexus から materialize されるためコミットしません。新規依存は build.gradle に座標を追加 必要。

検証 (dev サーバー / 2026-05-06)​

paho-mqtt subscriber で spBv1.0/# をキャプチャ:

[SUMMARY] total=511
  NDEATH: 1   ← 재시작 시 will 자동 발행
  NBIRTH: 1   ← seq=0
  DBIRTH: 10  ← OPC 10개 모두, seq=1..10
  DDATA: 499  ← 13초간 ≈ 38 msgs/s

DBIRTH メトリック数 = APP_TAG の該当 OPC タグ数と正確に一致 (OPC_UA_Kepware=11、OPC_00303=56、 OPC_LS_XBM_0001=28 など)。

DDATA トピック例:

spBv1.0/SITE_00001/DDATA/EDGE_00303/OPC_UA_Kepware  bytes=44  seq=125  ts=...

ペイロードデコード確認ツール:

MQTT_USER="${MQTT_USER:-edge}" \
MQTT_PASSWORD="$(tr -d '\r\n' < /run/secrets/mqtt-password)" \
python3 /tmp/spb_sub.py 100.106.92.21 1883 "$MQTT_USER" "$MQTT_PASSWORD" 60

(このスクリプトは一時ファイルであり、ビルド成果物には含まれません。)

Phase 3 — コマンド(NCMD/DCMD)受信動作​

sparkplug.cmd.enable=true のとき SparkPlugSession が次のトピックを自動購読します。

spBv1.0/{group}/NCMD/{edge_node}
spBv1.0/{group}/DCMD/{edge_node}/+        # 모든 device 와일드카드

受信メッセージは静的メソッド parseCmdTopic(String) → CmdTopic で検証後、分岐処理されます。

コマンド処理
NCMD Node Control/Rebirth=truerebirthAsync() — NBIRTH 再パブリッシュ + 全 OPC の DBIRTH 再パブリッシュ + (alias 有効時) aliasReverseMap 更新 (refreshAliasReverseMap())
DCMD metric writemetric name または alias で tag_id 逆検索 → 該当 OPC の driver write(tagId, value) 呼び出し → 結果を次の DDATA サイクルに反映

トピック検証 (parseCmdTopic)​

入力結果
spBv1.0/Plant1/NCMD/EDGE_00303OK (NCMD、deviceId=null)
spBv1.0/Plant1/DCMD/EDGE_00303/OPC_LSOK (DCMD、deviceId=OPC_LS)
null / 空文字列 / 4 segment 未満reject (null)
spBv1.0/Plant1/NDATA/EDGE_00303reject (outbound タイプ)
別 namespace (spBv2.0/...)reject
spBv1.0/Plant1/DCMD/EDGE_00303 (4 seg、deviceId 欠落)reject
6+ segment追加 segment 無視 (4-5 のみ使用)

→ 単体テスト SparkPlugCommandTopicTest が上記 10 ケースを保証。

rebirth 安全網​

rebirthAsync() が NBIRTH/DBIRTH 再パブリッシュ後に aliasReverseMap 更新を忘れ、次の DCMD が stale alias で lookup していた欠陥は修正済み — refreshAliasReverseMap() 呼び出し追加 (subscribeCommands() も同じヘルパーを使用、DRY)。

ロードマップ​

Phase内容ステータス
1 — BIRTH/DATA/DEATHNBIRTH / DBIRTH / DDATA / NDEATH(will)✅ 完了
2 — alias 最適化NBIRTH/DBIRTH 時に tag_id → alias(Long) テーブル生成、DDATA は alias のみパブリッシュ (ペイロード 30 ~ 40% 削減)。トグル: sparkplug.alias.enableオプション (toggle)
3 — NCMD/DCMD インバウンドhost が送信する metric write コマンド処理 + Node Control/Rebirth✅ 完了 (sparkplug.cmd.enable=true)
4 — STATE / 運用可視性STATE/{primary_host_id} 購読 → primary OFFLINE 時 DDATA 保留。/api/v1/edge 応答に sparkplug セクション追加 (connected / bdSeq / seq / devices)TBD

既知の制限​

  • DDATA value=null 時 isNull フラグ未設定: 現在 Metric.setValue(null) のみ呼び出し。スペック上 isNull=true 明示がより安全 → 後続補強候補。
  • 単一 metric per DDATA: キューが 1件ずつ trans するためバッチパブリッシュ最適化は未実施。頻度が高くなったら batched DDATA へ発展可能。
  • bdSeq path デフォルトが作業ディレクトリ相対パス: 実サーバーでは明示絶対パス (/opt/kopens/.../data/sparkplug/bdseq) 推奨。
  • DCMD write 失敗応答未実装: 現在 driver write() が失敗しても host に NDATA/DDATA で結果通知を 行いません。今後 result_status metric 追加検討。