Telegraf amqp_consumer

Telegraf
1

Hi
I am using the amqp_consumer to collect data from rabbitmq, which is receiving data from pmacct, in json format.
However no data is getting into InfluxDb, and the telegraf log is showing the following error’s
eg
2017-08-15T08:39:31Z E! metric parsing error, reason: [invalid field format], buffer: [{“event_type”: “purge”, “ip_src”: “192.168.0.3”, “ip_dst”: “192.168.0.1”, “port_src”: 44962, “port_dst”: 161, “ip_proto”: “udp”, “tos”: 0, “stamp_inserted”: “2017-08-15 09:35:00”, “stamp_updated”: “2017-08-15 09:39:31”, “packets”: 2, “bytes”: 146, “writer_id”: “default_amqp/2616”}], index: [23]: error parsing metric - {“event_type”: “purge”, “ip_src”: “192.168.0.3”, “ip_dst”: “192.168.0.1”, “port_src”: 44962, “port_dst”: 161, “ip_proto”: “udp”, “tos”: 0, “stamp_inserted”: “2017-08-15 09:35:00”, “stamp_updated”: “2017-08-15 09:39:31”, “packets”: 2, “bytes”: 146, “writer_id”: “default_amqp/2616”}

Telegraf v1.3.5 (git: release-1.3 7192e68b2423997177692834f53cdf171aee1a88)

Please can some one assist me in resolving this

Thanks

Aimee

2

Hi

To answer own question, missed the fact that telegraf requires a topic exhange, while pmacct only supports a direct exchange. Back to the drawing board :frowning:

Aimee

3

Here is a related feature request you may want to watch:

4

Hi
the topic issue is now resolved, but still getting the same error.
Is there any way to pre-process (cleanup) the data in telegraf, or will it need a specific plugin?

Thanks

Aimee

5

You can do some cleanup with the measurement filtering options. More advanced processing can be done with Kapacitor.

6

Hi
This is the data coming from telegraf
metric parsing error, reason: [invalid field format], buffer: [{“event_type”: “purge”, “ip_src”: “192.168.0.1”, “ip_dst”: “192.168.0.3”, “stamp_inserted”: “1503066600”, “stamp_updated”: “1503066811”, “packets”: 564, “bytes”: 50850, “writer_id”: “default_amqp/8809”}], index: [23]: error parsing metric - {“event_type
”: “purge”, “ip_src”: “192.168.0.1”, “ip_dst”: “192.168.0.3”, “stamp_inserted”: “1503066600”, “stamp_updated”: “1503066811”, “packets”: 564, “bytes”: 50850, “writer_id”: “default_amqp/8809”}

which is not compatible with influx’s Line Protocol.

Is there a way to reformat this using any of the existing TIKS products, or does it require a new plugin & parser?

Thanks

Aimee

7

You can use data_format="json", more info about this data format is here.

8

Hi

bit confused about where this would fit into my proposed solution…

pmacctd—>rabbitmq—>telegraf—>influxdb—>grafana

can I process the stream at rabbitmq—>telegraf or selegraf—>influxdb?

Please note its all streams, no databases (apart from influx) or file storage involved.

Confused

Aimee

9

This is an option on the amqp_consumer input in your Telegraf configuration:

[[inputs.amqp_consumer]]
  ## Data format to output.
  ## Each data format has it's own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
  data_format = "json"