# Telegraf amqp\_consumer **URL:** \ **Category:** Telegraf\ **Tags:** telegraf\ **Created:** [August 15, 2017, 1:18pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046 "2017-08-15T13:18:14Z")\ **Posts on this page:** 9\ **Page:** 1
**Author:** ![aimdev](https://avatars.discourse-cdn.com/v4/letter/a/e95f7d/32.png) [@aimdev](https://community.influxdata.com/u/aimdev)\ **Post date:** [August 15, 2017, 1:18pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/1 "2017-08-15T13:18:14Z")
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 ---
**Author:** ![aimdev](https://avatars.discourse-cdn.com/v4/letter/a/e95f7d/32.png) [@aimdev](https://community.influxdata.com/u/aimdev)\ **Post date:** [August 15, 2017, 5:33pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/2 "2017-08-15T17:33:56Z")
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 ☹ Aimee ---
**Author:** ![daniel](https://sea1.discourse-cdn.com/flex023/user_avatar/community.influxdata.com/daniel/32/142_2.png) [@daniel](https://community.influxdata.com/u/daniel)\ **Post date:** [August 15, 2017, 5:50pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/3 "2017-08-15T17:50:40Z")
Here is a related feature request you may want to watch: > > > \## Feature Request > Allow different types of exchanges to be used instead of jus…t topic. > > \### Proposal: > We mainly use the x-consistent-hash exchange type to load balance between queues across a cluster. It would be great if we could specify the exchange type in the config. > > \### Current behavior: > Topic is hardcoded in the plugin > > \### Desired behavior: > Configurable exchange type > > \### Use case: \[Why is this important (helps with prioritizing requests)\] > Using an exchange type like x-consistent-hash allows you to spread the messages out across all of the queues in the cluster evenly instead of having them all pinned to a single queue. ---
**Author:** ![aimdev](https://avatars.discourse-cdn.com/v4/letter/a/e95f7d/32.png) [@aimdev](https://community.influxdata.com/u/aimdev)\ **Post date:** [August 18, 2017, 5:55am UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/4 "2017-08-18T05:55:46Z")
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 ---
**Author:** ![daniel](https://sea1.discourse-cdn.com/flex023/user_avatar/community.influxdata.com/daniel/32/142_2.png) [@daniel](https://community.influxdata.com/u/daniel)\ **Post date:** [August 18, 2017, 5:49pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/5 "2017-08-18T17:49:34Z")
You can do some cleanup with the [measurement filtering](https://github.com/influxdata/telegraf/blob/master/docs/CONFIGURATION.md#measurement-filtering) options. More advanced processing can be done with Kapacitor. ---
**Author:** ![aimdev](https://avatars.discourse-cdn.com/v4/letter/a/e95f7d/32.png) [@aimdev](https://community.influxdata.com/u/aimdev)\ **Post date:** [August 18, 2017, 6:12pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/6 "2017-08-18T18:12:20Z")
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 ---
**Author:** ![daniel](https://sea1.discourse-cdn.com/flex023/user_avatar/community.influxdata.com/daniel/32/142_2.png) [@daniel](https://community.influxdata.com/u/daniel)\ **Post date:** [August 18, 2017, 6:48pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/7 "2017-08-18T18:48:53Z")
You can use `data_format="json"`, more info about this data format is [here](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md#json). ---
**Author:** ![aimdev](https://avatars.discourse-cdn.com/v4/letter/a/e95f7d/32.png) [@aimdev](https://community.influxdata.com/u/aimdev)\ **Post date:** [August 19, 2017, 10:42am UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/8 "2017-08-19T10:42:18Z")
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 ---
**Author:** ![daniel](https://sea1.discourse-cdn.com/flex023/user_avatar/community.influxdata.com/daniel/32/142_2.png) [@daniel](https://community.influxdata.com/u/daniel)\ **Post date:** [August 21, 2017, 5:55pm UTC](https://community.influxdata.com/t/telegraf-amqp-consumer/2046/9 "2017-08-21T17:55:36Z")
This is an option on the amqp\_consumer input in your Telegraf configuration: ```auto [[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" ```