Relevant telegraf.conf:

[agent]
debug=true

# Output Configuration for telegraf agent
[[outputs.influxdb_v2]]
  ## Point to your influxdb container
  urls = ["http://influxdb:8086"]
  ## insecure_skip_verify = true
  ## Token for authentication.
  token = "[redacted]"
  ## Organization is the name of the organization you wish to write to; must exist.
  organization = "myorg"
  ## Destination bucket to write into.
  bucket = "telegraf"
  
[[inputs.ping]]
  ## Hosts to send ping packets to.
  urls = ["influxdb"]
  
[[inputs.mqtt_consumer]]
  servers = ["tcp://mqtt:1883"]
  
  topics = [
    "spBv1.0/#",
  ]
  
  xpath_print_document = true
  xpath_protobuf_file = "/sparkplug_b.proto"
  xpath_protobuf_type = "org.eclipse.tahu.protobuf.Payload"

  client_id = "telegraf"
  data_format = "xpath_protobuf"
  
    [[inputs.mqtt_consumer.xpath_protobuf]]
        metric_selection = "//metrics"
        metric_name = "metrics"
        timestamp = "timestamp"

    [inputs.mqtt_consumer.xpath_protobuf.tags]
        dataType   = "dataType"

    [inputs.mqtt_consumer.xpath_protobuf.fields_int]
        value = "value"

System info:

Docker

Started as docker container.
Dockerfile:

FROM buildpack-deps:buster-curl

RUN DEBIAN_FRONTEND=noninteractive apt-get update &&
DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends iputils-ping snmp procps lm-sensors &&
rm -rf /var/lib/apt/lists/*

RUN set -ex &&
mkdir ~/.gnupg;
echo "disable-ipv6" >> ~/.gnupg/dirmngr.conf;
for key in
05CE15085FC09D18E99EFB22684A14CF2582E0C5 ;
do
gpg --keyserver hkp://keyserver.ubuntu.com --recv-keys "$key" ;
done

ENV TELEGRAF_VERSION 1.20.0
RUN ARCH= && dpkgArch="$(dpkg --print-architecture)" &&
case "${dpkgArch##*-}" in
amd64) ARCH='amd64';;
arm64) ARCH='arm64';;
armhf) ARCH='armhf';;
armel) ARCH='armel';;
) echo "Unsupported architecture: ${dpkgArch}"; exit 1;;
esac &&
wget --no-verbose https://dl.influxdata.com/telegraf/nightlies/telegraf_nightly_amd64.deb && \
dpkg -i telegraf_nightly_amd64.deb &&
rm -f telegraf_nightly_amd64.deb

EXPOSE 8125/udp 8092/udp 8094

COPY entrypoint.sh /entrypoint.sh
COPY sparkplug_b.proto /
CMD ["/usr/bin/telegraf"]

Docker compose file:
version: "3.9"
services:

mqtt:
container_name: mqtt
#image: emqx/emqx:latest
image: hivemq/hivemq4:latest
ports:
- "18083:18083"
- "1883:1883"
- "8080:8080"
restart: unless-stopped

nodered:
container_name: nodered
build: nodered/
ports:
- "1880:1880"
restart: unless-stopped
volumes:
- nodered_data:/data

telegraf:
container_name: telegraf
build: telegraf/
volumes:
- ./telegraf/mytelegraf.conf:/etc/telegraf/telegraf.conf

influxdb:
container_name: influxdb
image: influxdb
volumes:
- influxdbv2:/root/.influxdbv2
ports:
- "8086:8086"

volumes:
nodered_data:
influxdbv2:

Node Red exported flow:

[
{
"id": "f6f2187d.f17ca8",
"type": "tab",
"label": "Flow 1",
"disabled": false,
"info": ""
},
{
"id": "3cc11d24.ff01a2",
"type": "comment",
"z": "f6f2187d.f17ca8",
"name": "WARNING: please check you have started this container with a volume that is mounted to /data\n otherwise any flow changes are lost when you redeploy or upgrade the container\n (e.g. upgrade to a more recent node-red docker image).\n If you are using named volumes you can ignore this warning.\n Double click or see info side panel to learn how to start Node-RED in Docker to save your work",
"info": "\nTo start docker with a bind mount volume (-v option), for example:\n\n\ndocker run -it -p 1880:1880 -v /home/user/node_red_data:/data --name mynodered nodered/node-red\n\n\nwhere /home/user/node_red_data is a directory on your host machine where you want to store your flows.\n\nIf you do not do this then you can experiment and redploy flows, but if you restart or upgrade the container the flows will be disconnected and lost. \n\nThey will still exist in a hidden data volume, which can be recovered using standard docker techniques, but that is much more complex than just starting with a named volume as described above.",
"x": 350,
"y": 80,
"wires": []
},
{
"id": "a969a93b5b45440e",
"type": "inject",
"z": "f6f2187d.f17ca8",
"name": "dbirth",
"props": [
{
"p": "payload"
},
{
"p": "topic",
"vt": "str"
}
],
"repeat": "",
"crontab": "",
"once": false,
"onceDelay": 0.1,
"topic": "my_sensor/DBIRTH",
"payload": "{"timestamp":1465577611580,"metrics":[{"name":"my_int","value":456,"type":"int32"},{"name":"my_float","value":456,"type":"float"}]}",
"payloadType": "json",
"x": 390,
"y": 500,
"wires": [
[
"1cc90b63a305142f"
]
]
},
{
"id": "1cc90b63a305142f",
"type": "sparkplug",
"z": "f6f2187d.f17ca8",
"name": "",
"broker": "mqtt://mqtt",
"port": "1883",
"clientid": "NodeREDSimpleEdgeNode",
"groupid": "sparkplug_demo",
"edgenode": "nodered_edgenode",
"version": "spBv1.0",
"enablecache": "false",
"publishdeath": "true",
"user": "user",
"password": "PWRD",
"x": 600,
"y": 500,
"wires": [
[]
],
"info": "my sparkplug publisher"
},
{
"id": "08b5250dec3a77ea",
"type": "inject",
"z": "f6f2187d.f17ca8",
"name": "ddata",
"props": [
{
"p": "payload"
},
{
"p": "topic",
"vt": "str"
}
],
"repeat": "",
"crontab": "",
"once": false,
"onceDelay": 0.1,
"topic": "my_sensor/DDATA",
"payload": "{"timestamp":1465577611580,"metrics":[{"name":"my_int","value":456,"type":"int32"}]}",
"payloadType": "json",
"x": 390,
"y": 580,
"wires": [
[
"1cc90b63a305142f"
]
]
},
{
"id": "bfbc5e71405e5db3",
"type": "inject",
"z": "f6f2187d.f17ca8",
"name": "ddeath",
"props": [
{
"p": "payload"
},
{
"p": "topic",
"vt": "str"
}
],
"repeat": "",
"crontab": "",
"once": false,
"onceDelay": 0.1,
"topic": "my_sensor/DDEATH",
"payload": "{"timestamp":1465577611580}",
"payloadType": "json",
"x": 390,
"y": 660,
"wires": [
[
"1cc90b63a305142f"
]
]
}
]

Steps to reproduce:

1. Start up telegraf, mqtt, influxdb
2. Node Red has module node-red-contrib-sparkplug installed.
3. Import flow into node red and click on DBIRTH node. This should send a sparkplug b message to MQTT broker.

Expected behavior:

Telegraf should decode the sparkplug b message and send it to influxdb output.

Actual behavior:

Telegraf receives the message and prints error message:

C:\dev\iiot>docker-compose up telegraf
Starting telegraf ... done Attaching to telegraf
telegraf | 2021-09-14T10:50:13Z I! Starting Telegraf
telegraf | 2021-09-14T10:50:13Z I! Using config file: /etc/telegraf/telegraf.conf
telegraf | 2021-09-14T10:50:13Z I! Loaded inputs: mqtt_consumer ping
telegraf | 2021-09-14T10:50:13Z I! Loaded aggregators:
telegraf | 2021-09-14T10:50:13Z I! Loaded processors:
telegraf | 2021-09-14T10:50:13Z I! Loaded outputs: influxdb_v2
telegraf | 2021-09-14T10:50:13Z I! Tags enabled: host=db631e8a1a03
telegraf | 2021-09-14T10:50:13Z I! [agent] Config: Interval:10s, Quiet:false, Hostname:"db631e8a1a03", Flush Interval:10s
telegraf | 2021-09-14T10:50:13Z D! [agent] Initializing plugins
telegraf | 2021-09-14T10:50:13Z D! [agent] Connecting outputs
telegraf | 2021-09-14T10:50:13Z D! [agent] Attempting connection to [outputs.influxdb_v2]
telegraf | 2021-09-14T10:50:13Z D! [agent] Successfully connected to outputs.influxdb_v2
telegraf | 2021-09-14T10:50:13Z D! [agent] Starting service inputs
telegraf | 2021-09-14T10:50:13Z I! [inputs.mqtt_consumer] Connected [tcp://mqtt:1883]
telegraf | 2021-09-14T10:50:23Z D! [outputs.influxdb_v2] Wrote batch of 1 metrics in 14.0735ms
telegraf | 2021-09-14T10:50:23Z D! [outputs.influxdb_v2] Buffer fullness: 0 / 10000 metrics
telegraf | 2021-09-14T10:50:29Z D! [parsers.xpath_protobuf::mqtt_consumer] XML document equivalent: "21465577611580my_int3<int_value>456</int_value>my_float9<float_value>456</float_value>"
telegraf | 2021-09-14T10:50:29Z D! [parsers.xpath_protobuf::mqtt_consumer] Number of selected metric nodes: 2
telegraf | panic: interface conversion: interface {} is nil, not string
telegraf |
telegraf | goroutine 16 [running]:
telegraf | github.com/influxdata/telegraf/plugins/parsers/xpath.(*Parser).parseQuery(0xc000186080, {0x7f53cf23c100, 0x6, 0x7d394e0}, {0x43e7a20, 0xc000f0b540}, {0x43e7a20, 0xc000f0b720}, {{0xc000d30260, 0xd}, ...})
telegraf | /go/src/github.com/influxdata/telegraf/plugins/parsers/xpath/parser.go:169 +0x1a6d
telegraf | github.com/influxdata/telegraf/plugins/parsers/xpath.(*Parser).Parse(0xc000186080, {0xc000f92630, 0x2b, 0x2b})
telegraf | /go/src/github.com/influxdata/telegraf/plugins/parsers/xpath/parser.go:107 +0x4d0
telegraf | github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer.(*MQTTConsumer).onMessage(0xc00087e680, {0x5428848, 0xc000f5e900}, {0x53f5060, 0xc000f0b4a0})
telegraf | /go/src/github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer/mqtt_consumer.go:282 +0x74
telegraf | github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer.(*MQTTConsumer).recvMessage(0xc00087e680, {0x0, 0x2}, {0x53f5060, 0xc000f0b4a0})
telegraf | /go/src/github.com/influxdata/telegraf/plugins/inputs/mqtt_consumer/mqtt_consumer.go:271 +0x210
telegraf | github.com/eclipse/paho%2emqtt%2egolang.(*router).matchAndDispatch(0xc000f97aa0, 0xc000088800, 0x1, 0xc000185600)
telegraf | /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.0/router.go:171 +0x76d
telegraf | github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers.func1()
telegraf | /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.0/client.go:504 +0x31
telegraf | created by github.com/eclipse/paho%2emqtt%2egolang.(*client).startCommsWorkers
telegraf | /go/pkg/mod/github.com/eclipse/paho.mqtt.golang@v1.3.0/client.go:503 +0x3bd
telegraf exited with code 2

Additional info: