AMQP Consumer plugin #1678

jackzampolin · 2016-08-29T18:42:48Z

Required for all PRs:

CHANGELOG.md updated (we recommend not updating this until the PR has been approved by a maintainer)
Sign CLA (if not already signed)
README.md updated (if adding a new plugin)

hanej · 2016-10-10T14:38:32Z

Really looking forward to this PR so I can get the data back out of Rabbit

nhaugo · 2016-10-13T17:26:00Z

@hanej we are implementing a plugin system, this should get merged once that is complete.
see #1717 and #1718.

jackzampolin · 2017-01-05T17:14:02Z

@sparrc Any way we could get this in 1.2?

sparrc · 2017-01-06T10:09:58Z

Sorry bur I think 1.2 is too soon as I haven't reviewed this yet and there is a large backlog. We can try for 1.3.

sparrc

See comments in review.

One other thing: I think we might want to change the AMQP output plugin to default to an exchange of "". We should also verify that both plugins can talk to each other with their default configurations.

sparrc · 2017-01-13T15:46:36Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+func (rmq *AMQPConsumer) SampleConfig() string {
+	return `
+  # The following options form a connection string to amqp:
+  # amqp://{username}:{password}@{amqp_host}:{amqp_port}


change this so that users just need to specify they're server in this format

sparrc · 2017-01-13T15:46:47Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+  amqp_port = "5672"
+  # name of the queue to consume from
+  queue = "task_queue"
+	prefetch = 1000


spacing and what does this option do?

sparrc · 2017-01-13T15:47:48Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+}
+
+// listen(s) for new messages coming in from AMQP and pawns them off to handleMessage
+func (rmq *AMQPConsumer) listen(msgs <-chan amqp.Delivery, acc telegraf.Accumulator) {


we should probably wait for this function to exit when calling Stop()

sparrc · 2017-01-13T15:48:00Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+func (rmq *AMQPConsumer) registerConsumer() <-chan amqp.Delivery {
+	messages, err := rmq.ch.Consume(rmq.Queue, "", false, false, false, false, nil)
+	if err != nil {
+		panic(fmt.Errorf("%v: failed establishing connection to queue", err))


don't panic

sparrc · 2017-01-13T15:49:28Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+// listen(s) for new messages coming in from AMQP and pawns them off to handleMessage
+func (rmq *AMQPConsumer) listen(msgs <-chan amqp.Delivery, acc telegraf.Accumulator) {
+	for d := range msgs {
+		go handleMessage(d, acc, rmq.parser)


a go-routine per message is not a good idea, this will be too much in high-traffic scenarios. Would be better to just handle messages serially.

sparrc · 2017-01-13T15:51:02Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+func handleMessage(d amqp.Delivery, acc telegraf.Accumulator, parser parsers.Parser) {
+	metric, err := parser.Parse(d.Body)
+	if err != nil {
+		log.Fatalf("%v: error parsing metric - %v", err, string(d.Body))


dont use log.Fatal, this should be a log.Print and the message should start with E! ...

sparrc · 2017-01-13T15:52:13Z

plugins/inputs/rabbitmq_consumer/rabbitmq_consumer.go

@@ -0,0 +1,142 @@
+package rabbitmqConsumer


why are there two of the same plugin? couldn't we only do an amqp consumer?

sparrc · 2017-01-13T15:53:58Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+  queue = "task_queue"
+	prefetch = 1000
+
+  data_format = "influx"


you should put the standard comment in here that is in other plugins

sparrc · 2017-01-13T16:40:23Z

last thing is that we should probably mention in the documentation, default config, and README (including the main repo README) that AMQP == RabbitMQ.

danielnelson · 2017-03-01T22:41:46Z

I added support for a topic exchange, since that is what the output plugin uses, as well as the authentication options that the output plugin has (custom TLS and SASL)

goller

Minor changes in formatting and such.

goller · 2017-03-02T04:33:11Z

plugins/outputs/amqp/amqp.go

@@ -205,6 +209,7 @@ func (q *AMQP) Write(metrics []telegraf.Metric) error {
 				Headers:     q.headers,
 				ContentType: "text/plain",
 				Body:        buf,
+				//				DeliveryMode: amqp.Persistent,


Should this comment be removed?

goller · 2017-03-02T04:35:18Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+		nil,        // arguments
+	)
+	if err != nil {
+		return fmt.Errorf("Failed to declare an exchange: %s", err)


This error message is inconsistent with the rest. Should be:

return fmt.Errorf("%v: Failed to declare an exchange", err)

I usually use %s for errors, is there a benefit to using %v? if err != nil then the string format will always work properly, I believe it will just print whatever err.Error() returns

I'll stick to the current convention of "Failed to blah blah: %s"

goller · 2017-03-02T04:35:47Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+		a.Prefetch,
+		0,     // prefetch-size
+		false, // global
+	)
 	if err != nil {
 		return fmt.Errorf("%v: failed to set Qos", err)


Should be QoS in error message.

goller · 2017-03-02T04:37:54Z

plugins/outputs/amqp/amqp.go

@@ -160,7 +163,8 @@ func (q *AMQP) Connect() error {
 }

 func (q *AMQP) Close() error {
-	return q.channel.Close()
+	q.channel.Close()


Check for close error and return

goller · 2017-03-02T04:38:29Z

plugins/outputs/amqp/amqp.go

@@ -205,6 +209,7 @@ func (q *AMQP) Write(metrics []telegraf.Metric) error {
 				Headers:     q.headers,
 				ContentType: "text/plain",
 				Body:        buf,
+				//				DeliveryMode: amqp.Persistent,
 			})
 		if err != nil {
 			return fmt.Errorf("FAILED to send amqp message: %s", err)


Error formatting inconsistent here are compared to others

goller · 2017-03-02T04:38:47Z

plugins/outputs/amqp/amqp.go

@@ -129,6 +130,8 @@ func (q *AMQP) Connect() error {
 	if err != nil {
 		return err
 	}
+	q.conn = connection
+
 	channel, err := connection.Channel()
 	if err != nil {
 		return fmt.Errorf("Failed to open a channel: %s", err)


Error message inconsistent here as compared to others.

sparrc · 2017-03-02T10:52:45Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+  ## Binding Key
+  binding_key = "#"
+
+  ## Maximum number of messages server should give to the worker.


I would expand on this a bit, using language directly from either the streadway AMQP library or from the RabbitMQ documentation.

You can also link to external documentation here.

(especially mention that this is a channel QoS setting)

sparrc · 2017-03-02T10:55:44Z

plugins/inputs/amqp_consumer/README.md

@@ -0,0 +1,35 @@
+# AMQP Consumer Input Plugin
+
+This plugin reads data from an AMQP Queue ([RabbitMQ](https://www.rabbitmq.com/) being an example) formatted in one of


Here I think we should mention that this plugin is for the AMQP 0.9.1 protocol, and link to the RabbitMQ doc about it: https://www.rabbitmq.com/amqp-0-9-1-reference.html & https://www.rabbitmq.com/tutorials/amqp-concepts.html

sparrc · 2017-03-02T10:56:48Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+
+	// Declare a queue and assign it to AMQPConsumer
+	q, err := ch.QueueDeclare(
+		"telegraf", // queue


should the queue name be configurable? what if a user wanted to consume from two different telegraf instances?

I added support for this.

sparrc · 2017-03-02T11:01:09Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+  url = "amqp://localhost:5672/influxdb"
+  ## AMQP exchange
+  exchange = "telegraf"
+  ## Auth method. PLAIN and EXTERNAL are supported


Does EXTERNAL require the AMQP instance to have a specific plugin enabled? could you link to a doc about this? (I could be wrong, but https://www.rabbitmq.com/authentication.html seems to suggest it requires a plugin)

Yes it is required, I added a link in the latest push.

sparrc · 2017-03-02T11:06:20Z

plugins/inputs/amqp_consumer/README.md

+
+The following defaults are set to work with RabbitMQ:
+
+```


put toml after the ```, like:

```toml ...

this will tell github to colorize the readme

sparrc · 2017-03-02T11:15:33Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+func (a *AMQPConsumer) process(msgs <-chan amqp.Delivery, acc telegraf.Accumulator) {
+	defer a.wg.Done()
+	for d := range msgs {
+		metric, err := a.parser.Parse(d.Body)


nit: metric -> metrics

sparrc · 2017-03-02T11:18:49Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+
+func (a *AMQPConsumer) Stop() {
+	a.Lock()
+	defer a.Unlock()


should we call a.ch.Cancel()? https://godoc.org/github.com/streadway/amqp#Channel.Cancel

Could be no, TBH I don't really understand the difference between "Channel.Cancel" and "Channel.Close"

FWIW, there's probably nothing to change here, this example code from the library itself simply closes the connection, which "takes the channel with it": https://github.com/streadway/amqp/blob/master/_examples/pubsub/pubsub.go#L37

(note that the session struct in that example has an s.Channel object defined, but doesn't get used in the Close() func)

and with that being said, another example code calls Cancel and then Close: https://github.com/streadway/amqp/blob/master/_examples/simple-consumer/consumer.go#L141

but it calls Cancel with noWait=true, which is completely pointless AFAICT

I think it only matters if you will use multiple consumers on a single channel. We should be able to just close the connection as well.

sparrc · 2017-03-02T11:30:35Z

plugins/inputs/amqp_consumer/amqp_consumer.go

+// Read messages from queue and add them to the Accumulator
+func (a *AMQPConsumer) process(msgs <-chan amqp.Delivery, acc telegraf.Accumulator) {
+	defer a.wg.Done()
+	for d := range msgs {


what happens here when the channel is closed? does the loop just cleanly exit? how does that happen?

I would have thought it'd start sending nils?

"When the channel or connection closes, all delivery chans will also close. " - https://godoc.org/github.com/streadway/amqp#Channel.Consume

danielnelson · 2017-03-02T23:13:46Z

I believe all the concerns above are addressed, and additionally I added support for reconnecting. So there are quite a few changes, can you re-review?

sparrc · 2017-03-03T13:38:41Z

LGTM, nice work!

jwilder added the feature request Requests for new plugin and for new features to existing plugins label Sep 1, 2016

nhaugo self-assigned this Sep 1, 2016

jackzampolin force-pushed the jz-rabbitmq-input branch 3 times, most recently from 411e325 to 79f071b Compare October 7, 2016 19:49

nhaugo removed their assignment Oct 13, 2016

sparrc added the plugin request label Nov 26, 2016

sparrc added this to the 1.3.0 milestone Jan 6, 2017

sparrc suggested changes Jan 13, 2017

View reviewed changes

jackzampolin and others added 6 commits March 1, 2017 14:32

Add rabbitmq consumer input plugin

3f6a98f

Add README.md to RabbitMQConsumer

640b984

Change from RabbitMQ -> AMQP

c4e609c

make batch size configurable

863eea7

Fix varibale naming and add amqp_consumer to all/all.go

a321953

Add support for topic exchange to match amqp output

b7f217c

danielnelson force-pushed the jz-rabbitmq-input branch from cde1a7c to b7f217c Compare March 1, 2017 22:35

danielnelson requested a review from goller March 1, 2017 22:43

goller suggested changes Mar 2, 2017

View reviewed changes

sparrc reviewed Mar 2, 2017

View reviewed changes

danielnelson force-pushed the jz-rabbitmq-input branch from 205d328 to d2c903d Compare March 2, 2017 23:16

Add queue name and reconnect support to amqp_consumer

d4693ee

danielnelson force-pushed the jz-rabbitmq-input branch from d2c903d to d4693ee Compare March 2, 2017 23:41

goller approved these changes Mar 3, 2017

View reviewed changes

sparrc approved these changes Mar 3, 2017

View reviewed changes

Update changelog

e30439b

danielnelson merged commit 1074464 into master Mar 3, 2017

danielnelson deleted the jz-rabbitmq-input branch March 3, 2017 18:24

ssorathia pushed a commit to ssorathia/telegraf that referenced this pull request Mar 25, 2017

AMQP Consumer plugin (influxdata#1678)

0a025a1

vlamug pushed a commit to vlamug/telegraf that referenced this pull request May 30, 2017

AMQP Consumer plugin (influxdata#1678)

e497c4a

maxunt pushed a commit that referenced this pull request Jun 26, 2018

AMQP Consumer plugin (#1678)

2fe1613

		@@ -0,0 +1,35 @@
		# AMQP Consumer Input Plugin

		This plugin reads data from an AMQP Queue ([RabbitMQ](https://www.rabbitmq.com/) being an example) formatted in one of

AMQP Consumer plugin #1678

AMQP Consumer plugin #1678

Conversation

jackzampolin commented Aug 29, 2016 • edited by danielnelson Loading

Required for all PRs:

hanej commented Oct 10, 2016

nhaugo commented Oct 13, 2016

jackzampolin commented Jan 5, 2017

sparrc commented Jan 6, 2017

sparrc left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sparrc commented Jan 13, 2017

danielnelson commented Mar 1, 2017

goller left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sparrc Mar 2, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sparrc Mar 2, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sparrc Mar 2, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

danielnelson commented Mar 2, 2017

sparrc commented Mar 3, 2017

jackzampolin commented Aug 29, 2016 •

edited by danielnelson

Loading

sparrc Mar 2, 2017 •

edited

Loading

sparrc Mar 2, 2017 •

edited

Loading

sparrc Mar 2, 2017 •

edited

Loading