problems with json_transformation in telegraf-outputs.mqtt

[iserranoe]

Relevant telegraf.conf

[[outputs.mqtt]]
  fieldpass = ["FeederOverride","PressSPM__value", "PressSPM__target"]             
  topic = "###"
  json_transformation = '''
$each($merge([metrics.fields]), function($value, $key) {{
    "key": $key,
    "value": $value
  }}) ~> $reduce(function($acc, $entry) {(
    $contains($entry.key, "__")
      ? (
        $outerLevelKey := $split($entry.key, "__")[0];
        $innerLevelKey := $split($entry.key, "__")[1];        
        $merge([$acc, { 
          $outerLevelKey: $merge([
            $lookup($acc, $outerLevelKey), 
            { $innerLevelKey:$entry.value } 
          ])
        }])
      )
      : $merge([$acc, { $entry.key: $entry.value }])
  )}, {})
'''
  batch = true
  data_format="json"

Logs from Telegraf

[outputs.mqtt] Could not serialize metric batch for topic "###": argument must be an object or an array of objects

System info

Telegraf 1.27, Ubuntu 20.04, Docker 4.13.0

Docker

FROM telegraf:1.27

USER root

WORKDIR /telegraf COPY ./telegraf.conf /telegraf/telegraf.conf COPY ./OPCUA.key /telegraf/OPCUA.key COPY ./OPCUA.crt /telegraf/OPCUA.crt

RUN apt-get update

RUN apt-get install -y jq

RUN apt-get install -y nano

CMD ["sh","-c", "telegraf --config telegraf.conf"]

Steps to reproduce

1. We need to have input data with fields: [FeederOverride","PressSPM__value", "PressSPM__target"]
2. We need to install mosquitto
3. We send the date to que mosquitto with outputs.mqtt plugins

Expected behavior

I'm traying to transform the data from:

{
  "metrics": [
    {
      "fields": {
        "FeederOverride": 1
      },
      "name": "data_1",
      "tags": {
        "id": "ns=2;s=unit/OPCUA.FeederOverride"
      },
      "timestamp": 1696586985
    },
    {
      "fields": {
        "PressSPM__value": 10
      },
      "name": "data_1",
      "tags": {
        "id": "ns=2;s=unit/OPCUA.PressSPM"
      },
      "timestamp": 1696586985
    },
    {
      "fields": {
        "PressSPM__target": 2
      },
      "name": "data_1",
      "tags": {
        "id": "ns=2;s=unit/OPCUA.PressSPM_target"
      },
      "timestamp": 1696586985
    }
  ]
}

to

{
  "FeederOverride": 1,
  "PressSPM": {
    "value": 10,
    "target": 2
  }
}

Actual behavior

I get the error message:

[outputs.mqtt] Could not serialize metric batch for topic "###": argument must be an object or an array of objects

Additional info

The transformation is working in JSONata simulator: https://try.jsonata.org/5aN2vrMY3 but it's not working within Telegraf.

However, it is working when I change the json_transformation to:

$each($merge([metrics.fields]), function($value, $key) {{
    "key": $key,
    "value": $value
  }}) ~> $reduce(function($acc, $entry) {(
    $contains($entry.key, "__")
      ? (
        $outerLevelKey := $split($entry.key, "__")[0];
        $innerLevelKey := $split($entry.key, "__")[1];        
        $merge([$acc, { 
          $outerLevelKey: $merge([
            $acc.PressSPM, 
            { $innerLevelKey:$entry.value } 
          ])
        }])
      )
      : $merge([$acc, { $entry.key: $entry.value }])
  )}, {})

[srebhan]

Looking deeper into this, the issue seems to be in the upstream library we are using... I cannot fix it in Telegraf, it has to be corrected there: https://github.com/blues/jsonata-go

[iserranoe]

Thank you @srebhan, I will post an issue there

[srebhan]

I'm closing this issue as it seems like the upstream library won't get any fix for the matter and given that there seems to be a workaround. In case there is some movement upstream, please feel free to reopen the issue and/or bump the dependency to a fixed version...