https://www.umh.app logo
Join Discord
Powered by
Downsampler
# umh-support
r
Hanno wrote: I have a support question but I see the channels for that is achived so asking here if it is ok. I am trying to setup a basic modbus-> Sparkplug B flow using benthos-umh Here is the config:
Copy code
input:
  modbus:
    controller: "tcp://192.168.0.1:502"
    slaveIDs: [1]
    timeBetweenReads: "1s"
    unifiedAddresses:
      - "output_frequency.input.24.INT16:scale=0.01"
      - "output_voltage.input.25.INT16"
      - "dc_bus_voltage.input.26.INT16:scale=0.1"
      - "output_power.input.27.INT16" 

pipeline:
  processors:
    - mapping: |
        if meta("modbus_tag_name") == "heartbeat" {
          root = deleted()
        } else {
          # Payload
          root.value = content().string().number()
          root.timestamp_ms = timestamp_unix_milli()

          # Metadata
          meta location_path = "None"
          meta virtual_path = "None"
          meta tag_name = meta("modbus_tag_name")
          meta data_contract = "_sparkplug"
        }

output:
  sparkplug_b:
    mqtt:
      urls:
        - "tcp://localhost:1883"
    identity:
      group_id: "Edge"
      edge_node_id: "benthos-umh"
Modbus reads great, pipeline works. MQTT connection is made to my broker and node data is sent. But no reading are sent every second to the broker? I am thinking I did something that does not meet the requirements for the sparkplug_b output i.t.o DDATA or something?
This is probably the downsampler, check the documentation here: https://docs.umh.app/benthos-umh/processing/downsampler you can set as default: msg.meta.ds_ignore = "true"
h
Hi @runeston3 thanks for taking the time to support me with this. I did what you said and also stopped using the "mapping" processor and rather used the "tag_processor" I am still not coming right I added the ds_ignore as you suggested. I am definitely getting values on the output side (see srdout that is commented out). But as soon as I switch to the actual MQTT broker, I see it makes a connection to the mqtt broker but the tag values are not being sent every 1 sec to the mqtt broker? I am not sure what I am missing. Here is my YAML file:
Copy code
input:
  modbus:
    controller: "tcp://192.168.0.1:502"
    slaveIDs: [1]
    timeBetweenReads: "1s"
    unifiedAddresses:
      - "output_frequency.input.24.INT16:scale=0.01"
      - "output_voltage.input.25.INT16"
      - "dc_bus_voltage.input.26.INT16:scale=0.1"
      - "output_power.input.27.INT16" 

pipeline:
  processors:
    - tag_processor:
        defaults: |
          // 1. Drop the heartbeat message
          if (msg.meta.modbus_tag_name === "heartbeat") {
            return null;
          }

          // 2. Set the required metadata dynamically
          msg.meta.location_path = "EnergyDrive";
          msg.meta.data_contract = "_historian";
          msg.meta.tag_name = msg.meta.modbus_tag_name;
          msg.meta.ds_ignore = "true"

          return msg;

output:
  sparkplug_b:
    mqtt:
      urls:
        - "ssl://mqtt.energydrivecloud.com:8883"
    identity:
      group_id: "Edge"
      edge_node_id: "benthos-umh"

# output:
#   stdout: {}
Here is my log if I use the stdout (Prt1): > time="2026-06-08T08:06:54Z" level=info msg="Published retained BIRTH message on topic: spBv1.0/Edge/NBIRTH/benthos-umh" @service=benthos label="" path=root.output > time="2026-06-08T08:06:54Z" level=info msg="Successfully published BIRTH message" @service=benthos label="" path=root.output > time="2026-06-08T08:06:54Z" level=info msg="Subscribing to node rebirth commands on topic: spBv1.0/Edge/NCMD/benthos-umh" @service=benthos label="" path=root.output > time="2026-06-08T08:06:55Z" level=info msg="Successfully subscribed to node rebirth commands" @service=benthos label="" path=root.output > time="2026-06-08T08:08:57Z" level=info msg="Received SIGTERM, the service is closing" @service=benthos > time="2026-06-08T08:08:57Z" level=error msg="slave 1 encountered an error: modbus: connect: dial tcp 192.168.0.1:502: operation was canceled" @service=benthos label="" path=root.input
Prt2: > time="2026-06-08T08:08:57Z" level=error msg="Modbus transport error for slave 1, requesting reconnect: modbus: connect: dial tcp 192.168.0.1:502: operation was canceled" @service=benthos label="" path=root.input > time="2026-06-08T08:08:58Z" level=info msg="Published retained DEATH message before disconnect" @service=benthos label="" path=root.output > time="2026-06-08T08:08:58Z" level=info msg="Sparkplug output closed" @service=benthos label="" path=root.output > time="2026-06-08T08:09:00Z" level=info msg="Running main config from specified file" @service=benthos benthos_version="" path=/benthos.yaml > time="2026-06-08T08:09:00Z" level=info msg="Listening for HTTP requests at: http://0.0.0.0:4195" @service=benthos > time="2026-06-08T08:09:00Z" level=info msg="Slave 1: 0 holding request(s) (0 regs, 0 fields), 1 input (4 regs, 4 fields), 0 discrete (0 regs, 0 fields), 0 coil (0 regs, 0 fields)" @service=benthos label="" path=root.input > time="2026-06-08T08:09:00Z" level=info msg="Launching a Benthos instance, use CTRL+C to close" @service=benthos > time="2026-06-08T08:09:00Z" level=info msg="Output type stdout is now active" @service=benthos label="" path=root.output > time="2026-06-08T08:09:00Z" level=info msg="Successfully connected to Modbus device at tcp://192.168.0.1:502" @service=benthos label="" path=root.input > time="2026-06-08T08:09:00Z" level=info msg="Input type modbus is now active" @service=benthos label="" path=root.input
Prt3: > {"timestamp_ms":1780906141630,"value":21.46} > {"timestamp_ms":1780906141630,"value":9133} > {"timestamp_ms":1780906141630,"value":23.700000000000003} > {"timestamp_ms":1780906141630,"value":902} > {"timestamp_ms":1780906142637,"value":21.46} > {"timestamp_ms":1780906142637,"value":9133} > {"timestamp_ms":1780906142637,"value":23.700000000000003} > {"timestamp_ms":1780906142637,"value":912} > {"timestamp_ms":1780906143645,"value":21.46} > {"timestamp_ms":1780906143646,"value":9133} > {"timestamp_ms":1780906143646,"value":23.700000000000003} > {"timestamp_ms":1780906143646,"value":912} > {"timestamp_ms":1780906144658,"value":21.44} > {"timestamp_ms":1780906144658,"value":9133} > {"timestamp_ms":1780906144658,"value":23.700000000000003} > {"timestamp_ms":1780906144658,"value":912}
j
hey @hanno23 , do the tag-values even appear in the mqtt-broker and how often if so ?
h
Hi @janikkn no, they never appear in the mqtt broker. Here is a screenshot to prove that:
Again from the logs and my MQTT.fx session I can see that benthos did successfully do a NBIRTH.
a
Hello @hanno23 , I think I found the issue. Seems like our Sparkplug B output plugin is reading the payload[tag_name] while the tag_processor puts the value in payload.value. We will implement a solve for this, but as a workaround you could try this code snippet in your processor so the value is moved to the place where the output plugin expects it:
Copy code
- nodered_js:
        code: |
          var out = {};
          out[msg.meta.tag_name] = msg.payload.value;
          msg.payload = out;
          return msg;
Could you try this out and confirm that it is working for you?
h
Thanks @Aaron for the help, I did what you said and I could see in my stdout that the tags were now named their unique names instead of "value" but it still did not work 0.0
I did the following and it seems to have solved the problem:
Copy code
input:
  modbus:
    controller: "tcp://192.168.0.1:502"
    slaveIDs: [1]
    timeBetweenReads: "1s"
    unifiedAddresses:
      - "output_frequency.input.24.INT16:scale=0.01"
      - "output_voltage.input.25.INT16"
      - "dc_bus_voltage.input.26.INT16:scale=0.1"
      - "output_power.input.27.INT16" 

pipeline:
  processors:
    - mapping: |
        # 1. Drop the heartbeat message
        if meta("modbus_tag_name") == "heartbeat" {
          root = deleted()
        } else {
          # 2. Extract the actual tag name and the numeric value
          let tag = meta("modbus_tag_name")
          let val = content().string().number()
          
          # 3. Create a unique JSON object: e.g., {"output_voltage": 22.05}
          root = {
            $tag: $val
          }

          # 4. Set the metadata
          meta location_path = "EnergyDrive"
          meta data_contract = "_sparkplug"
          meta tag_name = $tag
          meta ds_ignore = "true"
        }

output:
  broker:
    pattern: fan_out
    outputs:
      - stdout: {}
      - sparkplug_b:
          mqtt:
            urls:
              - "ssl://mqtt.energydrivecloud.com:8883"
          identity:
            group_id: "Edge"
            edge_node_id: "benthos-umh"
          # 5. Map each metric to its UNIQUE payload key
          metrics:
            - name: "output_frequency"
              alias: 1
              type: "Float"
              value_from: "output_frequency"
            - name: "output_voltage"
              alias: 2
              type: "Float"
              value_from: "output_voltage"
            - name: "dc_bus_voltage"
              alias: 3
              type: "Float"
              value_from: "dc_bus_voltage"
            - name: "output_power"
              alias: 4
              type: "Float"
              value_from: "output_power"
As you can see, the only way I got my sparkplug_b output to actually publish the DDATA was to change the payload to the tag name as you suggested and then manually create a metric for each unique tag. Ideally I would just want to create a unqiue tag at the start where I do my modbus reading but I guess if this is the only way then it is the only way?
Maybe you guys can give me a clear example of what the exact payload and metadata is the sparkplug_b output is expecting?
r
check the documentation. https://docs.umh.app/benthos-umh/output/sparkplug-b-output also see if you can show the full output so validate all nedded meta data is correct.
j
a quick question: would it make sense for you to use umh-core so you can use the store and forward capabilitites there? so modbus to UNS, and then from UNS to spb?
h
Thanks @runeston3 , I have been following the docs religiously the last couple of days but I will check again and make sure I am not missing somehting. Thanks for the help so far.
Hi Jeremy, unfortunately I don't think that is possible. We have a bunch of small edge devices that have such bad connection they can barely download 100mb within an hour. I originally wanted to see if I can install Ignition Edge on them but that failed. So I started playing around with much smaller pieces of software (e.g. emqx neuron, hiveMQ edge, Node-Red, Benthos-umh-edge). As far as I understand umh-core is a bit more beafy, ment for factory edge installations and not for my case where I have unstable networks, low resource devices.
a
Hi @hanno23 we just merged a PR that should fix your issue. With the PR you also shouldn't have to define the metric list anymore. Feel free to try it out and report if everything is fixed for you. It is on staging now and I expect it to release next week.
j
ah ok, if downlaoding 300mb is an issue, then umh-core is still too beefy. but you would get automatic buffering on the edge with it, which could be nice.
a
The new benthos release is out now, so the bug should be fixed
j
h
Thanks guys, just away commissioning a project the next couple of weeks but will be sure to test it out once I am back👍 💪 Really impressed with UMH's level of assistance and update rollout!
4 Views
PreviousNext
Powered by