Converting collecd binary stream to JSON with Telegraf or collecd forwarding?

Is it possible to use Telegraf or collectd to receive collectd binary streams and convert the data to JSON format?

It is possible yes but it would require soem custom coding. You could use the Telegraf Exec input plugin and define a script or program that received the input and performed the conversion. Then the output could be sent to Influx.

[[inputs.exec]]
commands = [“MyCustomTransformer”]

Timeout for each command to complete.

timeout = “5s”

Data format to consume.

data_format = “influx”

@sgreszcz Actually telegraf natively parses the collectd binary format. You can use the socket_listener to catch the collectd stream and then use the JSON data output format to output it to a queue.

[[inputs.socket_listener]]
...
    data_format = "collectd"

[[outputs.kafka]]
...
    data_format = "json"

Thanks @jackzampolin. What would be the most efficient way if we want to use inputs.socket_listener with data_format = “collectd” and then forward the metrics to an central influxDB server?

I’ve added:

# # Generic socket listener capable of handling multiple socket types.
[[inputs.socket_listener]]
  service_address = "udp://0.0.0.0:25826"
  name_prefix = "collectd_"

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "collectd"

  ## Authentication file for cryptographic security levels
  ## collectd_auth_file = "/etc/collectd/auth_file"
  ## One of none (default), sign, or encrypt
  ## collectd_security_level = "encrypt"
  ## Path of to TypesDB specifications
  collectd_typesdb = ["/usr/share/collectd/types.db"]

I want to output the collectd metrics to one DB name in the central influxdb, and I want the other Linux-based performance metrics to go to another database (“telegraf”) in the same central influxdb. Do I also have to set up some specific listening on the central InfluxDB?

Thank you!

You can use the namepass/namedrop or tagdrop/tagpass parameters on the output to select which metrics to output. Here is an example using namepass.

1 Like

Thanks @daniel ,

In the case where I want to pass only name_prefix = “uc-ops-duty-vcs” collectd metrics to a specific database, would I have to do an name_drop and a name_pass for each output? I pretty much only want the collectd metrics to go to a db called “uc-ops-duty-vcs” and the rest (default Linux sever telegraf collection - not shown) to the “telegraf” db.

Also, is there a way to label the data with a prefix to categorise it, but then not have that prefix show up as it will be a bit redundant as the database name would be the reference as to the data type?

Does this make sense?

[[inputs.socket_listener]]
#  Receive collectd stats from VCS devices
service_address = "udp://0.0.0.0:25826"
name_prefix = "uc-ops-duty-vcs"
data_format = "collectd"
collectd_typesdb = ["/usr/share/collectd/types.db"]

[[outputs.influxdb]]
urls = ["http://tick-alln-001:8086"] # required
## The target database for metrics (telegraf will create it if not exists).
database = "telegraf" # required
namedrop = [uc-ops-duty-vcs*]

urls = ["http://tick-alln-001:8086"] # required
## The target database for vcs metrics.
database = "uc-ops-duty-vcs" # required
namepass = [uc-ops-duty-vcs*]

@daniel

This sort of works, however I now have the issue where the measurements are all being prefixed with “uc-ops-duty-vcs” which messes up templating.

Is there another way to partition the Telegraf input data into particular outputs being forwarded to different InfluxDV database instances?

@daniel,

I tried assigning a tag to the collectd stream on input, but it still doesn’t differentiate on output. I still get the default Linux server telegraf stats and the collectd received stats in both databases in influxDB

 # # Generic socket listener capable of handling multiple socket types.
[[inputs.socket_listener]]
#  Receive collectd stats from VCS devices
service_address = "udp://0.0.0.0:25826"
data_format = "collectd"
collectd_typesdb = ["/usr/share/collectd/types.db"]
  [inputs.socket_listener.tags]
    tag = "prodvcs"

[[outputs.influxdb]]
  urls = ["http://tick-alln-001:8086"] # required
  ## The target database for metrics (telegraf will create it if not exists).
  database = "telegraf" # required
  tagdrop = ["prodvcs"]

# Define a database for vcs collectd metrics
[[outputs.influxdb]]
  urls = ["http://tick-alln-001:8086"]
  database = "vcs"
  # Only accept the vcs collecd metrics in this database
  tagpass = ["prodvcs"]

Sorry for all the noise, but I finally figured out how to use the tagpass/tagdrop to route the telegraf inputs to two separate InfluxDBs on the same server.

[[outputs.influxdb]]
  urls = ["http://tick-alln-001:8086"] # required
  ## The target database for metrics (telegraf will create it if not exists).
  database = "telegraf" # required
  [outputs.influxdb.tagdrop] 
    source = ["prodvcs"]

# Define a database for vcs collectd metrics
[[outputs.influxdb]]
  urls = ["http://tick-alln-001:8086"]
  database = "vcs"
  # Only accept the vcs collecd metrics in this database
  [outputs.influxdb.tagpass] 
    source = ["prodvcs"]

[[inputs.socket_listener]]
#  Receive collectd stats from VCS devices
service_address = "udp://0.0.0.0:25826"
data_format = "collectd"
#  ## Path of to TypesDB specifications
collectd_typesdb = ["/usr/share/collectd/types.db"]
  [inputs.socket_listener.tags]
    source = "prodvcs"
1 Like

I’m trying the same thing. but it doesn’t appear to work…

Line protocol output
ipcSession_25,hostname=,instance_number=1,component=ads,source=RMC uncompBytesSentRate=0
ipcSession_25,hostname=,instance_number=1,component=ads,source=RMC packetsSentRate=0
ipcSession_25,hostname=,instance_number=1,component=ads,source=RMC bytesReceivedRate=0
ipcSession_25,hostname=,instance_number=1,component=ads,source=RMC packetsReceivedRate=0
ipcSession_24,hostname=,instance_number=1,component=ads,source=RMC uncompBytesSentRate=0
ipcSession_24,hostname=,instance_number=1,component=ads,source=RMC packetsSentRate=0
ipcSession_24,hostname=,instance_number=1,component=ads,source=RMC bytesReceivedRate=0
ipcSession_24,hostname=,instance_number=1,component=ads,source=RMC packetsReceivedRate=0
ipcSession_23,hostname=,instance_number=1,component=ads,source=RMC uncompBytesSentRate=0
ipcSession_23,hostname=,instance_number=1,component=ads,source=RMC packetsSentRate=0
ipcSession_23,hostname=,instance_number=1,component=ads,source=RMC bytesReceivedRate=0
ipcSession_23,hostname=,instance_number=1,component=ads,source=RMC packetsReceivedRate=0

Telegraf Config

[[outputs.influxdb]]
retention_policy = ''
write_consistency = 'any’
timeout = '5s’
urls = [‘udp://:8089’]
database = ‘rmc’
[outputs.influxdb.tagpass]
source = [“RMC”]

[[outputs.influxdb]]
retention_policy = ''
write_consistency = 'any’
timeout = '5s’
urls = [‘udp://:8089’]
database = ‘telegraf’
[outputs.influxdb.tagdrop]
source = [“RMC”]