Mapping mqtt Topic as measurament name

Telegraf
1

Hi all,
i have a broker tath receive the data from remote site like this “RemoteSite/name_of_site/ProcessValues”. i would like to set the measurament name to topic part “name_of_site” but i do not understand how i can do it with topi parser.
Is there somebody can help me?

thanks in advance!

2

You can do this using Telegraf’s MQTT Consumer plugin along with the topic_parsing feature. Here’s how to configure it:

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker-address:1883"]
  topics = ["RemoteSite/+/ProcessValues"]
  username = "your-username"
  password = "your-password"
  
  # Set the topic as a tag
  topic_tag = "topic"
  
  # Configure topic parsing
  topic_parsing = [
    "RemoteSite/measurement/ProcessValues"
  ]
  
  # Additional data format settings
  data_format = "json" # Adjust based on your actual data format

The key part is the topic_parsing section. By using “measurement” in the position where “name_of_site” appears in your topic, Telegraf will automatically use that segment as the measurement name.

The + wildcard in the topics line allows it to subscribe to all topics that match the pattern, regardless of what “name_of_site” is.

Take a look at the Topic Parsing Example in the docs.

1 Like
3

Thanks!
I have another question..
Mqtt topic nomrally receive an array of metrics value as

{
  "metrics": [
    {
      "alias": 2,
      "datatype": 9,
      "name": "VAR2",
      "timestamp": 1744963440000,
      "value": 46.102432250976562
    },
    {
      "alias": 3,
      "datatype": 9,
      "name": "VAR1",
      "timestamp": 1744963440000,
      "value": 3.3405671119689941
    },

  ],
  "seq": 99,
  "timestamp": 1744963440033
}

and with pivot i trasform it into a metrics like
name_of_site , [VAR1] = 3.3405671119689941 , [VAR2]=46.102432250976562 1744963440033
and with output sql i wirte it into a table.

Some time i receive the below topic :

{
  "metrics": [
    {
      "alias": 1,
      "datatype": 12,
      "name": "CloudLogger Error Message",
      "timestamp": 1744964501933,
      "value": ""
    }
  ],
  "seq": 1,
  "timestamp": 1744964503834
}

i would like to discard this payload to create a metrics because it dosent match a sql table and sql output hang on writing a row..

How i can drop this metrics?

should i filter on mqtt, pivot or sql?

[agent]
  interval = "10s"
  round_interval = true
  flush_interval = "5s"
  metric_batch_size = 1000
  metric_buffer_limit = 10000
  collection_jitter = "0s"

[[inputs.mqtt_consumer]]
 
  servers = ["xxx"]
  topics =[
	'SolarFields/#'
  ]
  qos = 1	
  topic_tag =""
  
  ## Connection timeout for initial connection in seconds
  connection_timeout = "180s"
  persistent_session = true
  client_id = ""
  username = ""
  password = ""


  ## Use TLS but skip chain & host verification
  insecure_skip_verify = true
  data_format = "json_v2"
  [[inputs.mqtt_consumer.topic_parsing]]
        topic = "SolarFields/+/ProcessValues"
        measurement = "_/measurament/_"

       
  [[inputs.mqtt_consumer.json_v2]]

	[[inputs.mqtt_consumer.json_v2.object]]
		path = "metrics"
		disable_prepend_keys=true
		included_keys=["name","value"]
		timestamp_key='timestamp'
		timestamp_format='unix'
		timestamp_timezone="Europe/Rome"
		tags = ["name"]

[[processors.regex]]
[[processors.regex.tags]]
      key="name"
      pattern = '(\w+)$'
      replacement = '[${1}]'

[[processors.pivot]]
	tag_key="name"
	value_key="value"
	
[[aggregators.merge]]
	drop_original= true
	
[[outputs.file]]
files=["stdout"]

[[outputs.sql]]
        log_level="error"
		driver="mssql"
        data_source_name = "sqlserver://ccccccc/?database=test"
        timestamp_column = "timestamp"
		table_exists_template="SELECT TOP(1) 1 FROM {TABLE}"


  [outputs.sql.convert]
   integer              = "INT"
    real                 = "FLOAT"
    text                 = "TEXT"
    timestamp            = "datetimeoffset"
    defaultvalue         = "TEXT"
    unsigned             = "UNSIGNED"
    bool                 = "BOOL"