Telegraf解析MQTT JSON数据时出现invalid character 'c'错误求助

错误信息 #

Error in plugin: invalid character 'c' looking for beginning of value

环境信息 #

• InfluxDB v2.7.10 (OSS)
• Telegraf 1.32.0
• Ubuntu 20.04.6 LTS

相关数据与测试输出 #

MQTT发布数据示例 #

msg send {"msg":"Hello","device_id":1234567890,"inputA":148,"inputB":-2167,"inputC":818,"inputD":488,"inputE":46,"inputF":179,"inputG":1,"inputH":1,"protocol":"tcp","date":"2024-10-12 13:41:28"}

Telegraf --test输出 #

test,device_id=1234567890,topic=test inputA=1.48,inputB=-21.67,inputC=8.18,inputD=4.88,inputE=0.46,inputF=1.79,inputG=0.01,inputH=0.01 1728733410993721776

运行telegraf.conf时的终端输出 #

当前Telegraf配置 #

[agent]
  interval = "10s"
  round_interval = true
  metric_batch_size = 2500 #1000
  metric_buffer_limit = 100000 #10000
  collection_jitter = "0s"
  flush_interval = "10s"
  flush_jitter = "0s"
  precision = ""
  hostname = ""
  omit_hostname = true
 
[[inputs.mqtt_consumer]]
  servers = ["tcp://*****:1883"]
  topics = ["emqx/#"]
  username = "****"
  password = "****"
  client_id = "*****"
  data_format = "json"
  tag_keys = ["device_id"]
  qos = 0
[[processors.regex]]
  [[processors.regex.tags]]
    key = "topic"
    pattern = "^emqx/(.*)$"
    replacement = "${1}"
[[processors.converter]]
  [processors.converter.fields]
    float = ["inputA", "inputB", "inputC", "inputD", "inputE", "inputF", "inputG", "inputH"]
[[processors.starlark]]
  source = '''
def apply(metric):
    for field in ["inputA", "inputB", "inputC", "inputD", "inputE", "inputF", "inputG", "inputH"]:
        if field in metric.fields:
            metric.fields[field] = float(metric.fields[field]) / 100  # Remove this line to keep original values
    metric.name = metric.tags["topic"]
    return metric
'''
[[outputs.influxdb_v2]]
  urls = ["http://*****:8086"]
  token = "$INFLUX_TOKEN"
  organization = "hivagroup"
  bucket = "test"
  timeout = "5s"

需求说明 #

希望实现特定的InfluxDB输出格式，当前问题出在Telegraf的数据解析环节，请求帮助定位并解决。

内容的提问来源于stack exchange，提问作者arj