How to write data to multiple buckets

Telegraf
, , ,
1

Hi there!

I recieve different messages (jsonv2) from one sensor and would like to put different messages into different buckets.

This is my config at the moment:

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "untouchedBucket"

##(1065)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "msg_id"
      data_type = "int"
    [[inputs.mqtt_consumer.json_v2.tag]]
      rename = "device_id"
      path = "device_info.device_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      rename = "mac"
      path = "device_info.mac"
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "switch_state"
      path = "data.switch_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "load_state"
      path = "data.load_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "overload_state"
      path = "data.overload_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "overcurrent_state"
      path = "data.overcurrent_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "overvoltage_state"
      path = "data.overvoltage_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "time_plug"
      path = "data.timestamp"
      data_type = "string"


##(1066)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "msg_id"
      data_type = "int"
    [[inputs.mqtt_consumer.json_v2.tag]]
      rename = "device_id"
      path = "device_info.device_id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      rename = "mac"
      path = "device_info.mac"
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "switch_state"
      path = "data.switch_state"
      data_type = "bool"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "countdown"
      path = "data.countdown"
      data_type = "int"
      optional = true
    [[inputs.mqtt_consumer.json_v2.field]]
      rename = "time_plug"
      path = "data.timestamp"
      data_type = "string"

I would like to have message 1065 still being written to “untouchedbucket”, while message 1066 should go to “touchedbucket”.

How should my config look like?
Can somebody give me an example?
Do I have to run a second Telegraf instance to do this?

2

@Nils_Franke You should be able to do it with a single agent, but you need to do the following:

  1. Update each set of metrics so Telegraf can differentiate them. You could add distinct measurement names or a distinct tag to each. For example:

  • Use distinct measurement names

    ##(1065)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data_1065"

# ...

##(1066)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data_1066"

  • Add distinct tags (my recommendation)

    ##(1065)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      message = "1065"

# ...

##(1066)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      message = "1066"

  1. Add a second influxdb_v2 output plugin that outputs to the “touchedbucket.”

  2. Add filters to your two output plugins to only write points based on the message differentiator. For example, if you use the distinct tags:

    [[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "untouchedBucket"
  [outputs.influxdb_v2.tagpass]
    message = ["1065"]

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "touchedBucket"
  [outputs.influxdb_v2.tagpass]
    message = ["1066"]

There’s more information about Telegraf metric filtering in the documentation.

3

Thank you for your quick reply!

Unfortunately, going with the tagpass option Telegraf fails to start:

##Telegraf fails to start

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "untouchedBucket"
  [outputs.influxdb_v2.tagpass]
    message = ["1065"]

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "touchedBucket"
  [outputs.influxdb_v2.tagpass]
    message = ["1066"]

##(1065)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      message = "1065"

# ...

##(1066)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data"
    [[inputs.mqtt_consumer.json_v2.tag]]
      message = "1066"

Going with the namepass/nameoverride option, Telegraf starts without any errors but both messages get into both buckets:

##Starts without errors, sends everything round robin

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "untouchedBucket"
  [outputs.influxdb_v2.namepass]
    namepass = ["package_data_1065"]

[[outputs.influxdb_v2]]
  urls = ["http://127.0.0.1:8086"]
  token = "mytoken"
  organization = "myorg"
  bucket = "touchedBucket"
  [outputs.influxdb_v2.namepass]
    namepass = ["package_data_1066"]

##(1065)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data_1065"

# ...

##(1066)
  [[inputs.mqtt_consumer.json_v2]]
    name_override = "package_data_1066"

Did I miss something out?

Btw I am using InfluxDB 2.7.11 and Telegraf 1.35.3

4

Does Telegraf output an error message when it fails to start?

5

The telegraf.log itself only gave the two shutdown messages and then stayed empty until
the restart few minutes later. Without failing config it started up normal:

2025-08-12T11:02:02Z I! [processors.execd] Process /usr/bin/python3 shut down
2025-08-12T11:02:02Z I! [agent] Hang on, flushing any cached metrics before shutdown
2025-08-12T11:02:02Z I! [agent] Stopping running outputs
2025-08-12T11:06:15Z I! Loading config: /etc/telegraf/telegraf.conf

this is what systemd gave me:

× telegraf.service - Telegraf
     Loaded: loaded (/lib/systemd/system/telegraf.service; enabled; vendor preset: enabled)
     Active: failed (Result: exit-code) since Tue 2025-08-12 13:02:04 CEST; 1min 48s ago
       Docs: https://github.com/influxdata/telegraf
    Process: 53019 ExecStart=/usr/bin/telegraf -config /etc/telegraf/telegraf.conf -config-directory /etc/telegraf/tele>
   Main PID: 53019 (code=exited, status=1/FAILURE)
        CPU: 57ms

Aug 12 13:02:04 Fred systemd[1]: telegraf.service: Scheduled restart job, restart counter is at 5.
Aug 12 13:02:04 Fred systemd[1]: Stopped Telegraf.
Aug 12 13:02:04 Fred systemd[1]: telegraf.service: Start request repeated too quickly.
Aug 12 13:02:04 Fred systemd[1]: telegraf.service: Failed with result 'exit-code'.
Aug 12 13:02:04 Fred systemd[1]: Failed to start Telegraf.
6

Hi @Nils_Franke ,

If creating a duplicate of the data is not a requirement and you only want to send the different metrics to a different bucket, then using bucket_tag in the influxdb configuration might be a better route.

With that you can define the influxdb bucket as a tag. You can use the tag method that @scott described.