Sending to direct exchange not working in amqp output

[keitalbame]

Relevant telegraf.conf:

[[inputs.logparser]]
  files = ["/var/log/nginx/access.log"]
  [inputs.logparser.grok]
    patterns = ["%{COMMON_LOG_FORMAT}"]
    measurement = "nginx_access_log"

[[outputs.amqp]]
  brokers = ["amqp://rabbitmq:5672"]
  exchange = "log-messages"
  exchange_type = "direct"
  exchange_durability = "durable"
  username = "xxx"
  password = "xxx"
  routing_tag = "log-messages"

  # routing_key = "log-messages"

  delivery_mode = "persistent"
  data_format = "json"
[[outputs.file]]
  files = ["stdout"]
  data_format = "json"

System info:

docker image telegraf:1.10.1

Steps to reproduce:

1. Configure rabbitmq with exchange, queue and routing key from docker image rabbitmq:3.7-management
2. Run telegraf with above configuration

Expected behavior:

Messages arriving on rabbitmq

Actual behavior:

Messages not arriving to the queue in rabbitmq but data is seen arriving to the exchange.
Same behaviour when also setting up routing_key = "log-messages" or without specifiyng data_format.

from rabbitmq:

rabbitmq_1          | 2019-04-01 14:38:40.006 [debug] <0.859.0> Supervisor {<0.859.0>,rabbit_connection_sup} started rabbit_reader:start_link(<0.860.0>, {acceptor,{0,0,0,0,0,0,0,0},5672}) at pid <0.861.0>
rabbitmq_1          | 2019-04-01 14:38:40.006 [info] <0.861.0> accepting AMQP connection <0.861.0> (172.28.0.4:53736 -> 172.28.0.2:5672)
rabbitmq_1          | 2019-04-01 14:38:40.007 [debug] <0.861.0> User 'guest' authenticated successfully by backend rabbit_auth_backend_internal
rabbitmq_1          | 2019-04-01 14:38:40.007 [debug] <0.860.0> Supervisor {<0.860.0>,rabbit_connection_helper_sup} started rabbit_queue_collector:start_link(<<"172.28.0.4:53736 -> 172.28.0.2:5672">>) at pid <0.862.0>
rabbitmq_1          | 2019-04-01 14:38:40.007 [debug] <0.860.0> Supervisor {<0.860.0>,rabbit_connection_helper_sup} started rabbit_heartbeat:start_heartbeat_sender(#Port<0.62>, 60, #Fun<rabbit_reader.34.52813811>, {heartbeat_sender,<<"172.28.0.4:53736 -> 172.28.0.2:5672">>}) at pid <0.863.0>
rabbitmq_1          | 2019-04-01 14:38:40.008 [info] <0.861.0> connection <0.861.0> (172.28.0.4:53736 -> 172.28.0.2:5672): user 'guest' authenticated and granted access to vhost '/'
rabbitmq_1          | 2019-04-01 14:38:40.008 [debug] <0.860.0> Supervisor {<0.860.0>,rabbit_connection_helper_sup} started rabbit_heartbeat:start_heartbeat_receiver(#Port<0.62>, 60, #Fun<rabbit_reader.35.52813811>, {heartbeat_receiver,<<"172.28.0.4:53736 -> 172.28.0.2:5672">>}) at pid <0.864.0>
rabbitmq_1          | 2019-04-01 14:38:40.008 [debug] <0.860.0> Supervisor {<0.860.0>,rabbit_connection_helper_sup} started rabbit_channel_sup_sup:start_link() at pid <0.865.0>
rabbitmq_1          | 2019-04-01 14:38:40.008 [debug] <0.866.0> Supervisor {<0.866.0>,rabbit_channel_sup} started rabbit_writer:start_link(#Port<0.62>, 1, 131072, rabbit_framing_amqp_0_9_1, <0.861.0>, {<<"172.28.0.4:53736 -> 172.28.0.2:5672">>,1}, true) at pid <0.868.0>
rabbitmq_1          | 2019-04-01 14:38:40.008 [debug] <0.866.0> Supervisor {<0.866.0>,rabbit_channel_sup} started rabbit_limiter:start_link({<<"172.28.0.4:53736 -> 172.28.0.2:5672">>,1}) at pid <0.869.0>
rabbitmq_1          | 2019-04-01 14:38:40.009 [debug] <0.866.0> Supervisor {<0.866.0>,rabbit_channel_sup} started rabbit_channel:start_link(1, <0.861.0>, <0.868.0>, <0.861.0>, <<"172.28.0.4:53736 -> 172.28.0.2:5672">>, rabbit_framing_amqp_0_9_1, {user,<<"guest">>,[administrator],[{rabbit_auth_backend_internal,none}]}, <<"/">>, [{<<"connection.blocked">>,bool,true},{<<"consumer_cancel_notify">>,bool,true}], <0.862.0>, <0.869.0>) at pid <0.870.0>

Additional info:

from telegraf logs:

{"fields":{"auth":"-","client_ip":"127.0.0.1","http_version":1.1,"ident":"-","request":"/poweredby.png","resp_bytes":0},"name":"nginx_access_log","tags":{"host":"b85ac6870c4b","path":"/var/log/nginx/access.log","resp_code":"304","verb":"GET"},"timestamp":1554129512}
{"fields":{"auth":"-","client_ip":"127.0.0.1","http_version":1.1,"ident":"-","request":"/nginx-logo.png","resp_bytes":0},"name":"nginx_access_log","tags":{"host":"b85ac6870c4b","path":"/var/log/nginx/access.log","resp_code":"304","verb":"GET"},"timestamp":1554129512}
{"fields":"auth":"-","client_ip":"127.0.0.1","http_version":1.1,"ident":"-","request":"/","resp_bytes":0},"name":"nginx_access_log","tags":{"host":"b85ac6870c4b","path":"/var/log/nginx/access.log","resp_code":"304","verb":"GET"},"timestamp":1554129512}
2019-04-01T14:38:40Z D! Output [amqp] connecting to "amqp://rabbitmq:5672"
2019-04-01T14:38:40Z D! [outputs.file] wrote batch of 3 metrics in 107.812µs
2019-04-01T14:38:40Z D! [outputs.file] buffer fullness: 0 / 10000 metrics. 
2019-04-01T14:38:40Z D! Output [amqp] connected to "amqp://rabbitmq:5672"
2019-04-01T14:38:40Z D! [outputs.amqp] wrote batch of 3 metrics in 4.298431ms

[danielnelson]

It looks like Telegraf received confirmation of delivery to RabbitMQ:

D! [outputs.amqp] wrote batch of 3 metrics in 4.298431ms

Perhaps you don't have a queue bound to the routing key? Check the section on direct exchanges in the routing tutorial:

The routing algorithm behind a direct exchange is simple - a message goes to the queues whose binding key exactly matches the routing key of the message.

All other messages will be discarded.

[keitalbame]

@danielnelson already tried to create queues and routing keys with different names and issue still persists.
The routing keys were already setup.

$ rabbitmqadmin list exchanges
+--------------------+---------+
|        name        |  type   |
+--------------------+---------+
|                    | direct  |
| amq.direct         | direct  |
| amq.fanout         | fanout  |
| amq.headers        | headers |
| amq.match          | headers |
| amq.rabbitmq.log   | topic   |
| amq.rabbitmq.trace | topic   |
| amq.topic          | topic   |
| log-messages       | direct  |
+--------------------+---------+

$ rabbitmqadmin list queues
+--------------+----------+
|     name     | messages |
+--------------+----------+
| log-messages | 0        |
+--------------+----------+

$ rabbitmqadmin list bindings
+--------------+--------------+--------------+
|    source    | destination  | routing_key  |
+--------------+--------------+--------------+
|              | log-messages | log-messages |
| log-messages | log-messages | log-messages |
+--------------+--------------+--------------+

Do you have any additional pointers?

[danielnelson]

Try sending to the direct exchange using a command similar to this and verify that you receive the data:

rabbitmqadmin -u xxx -p xxx publish exchange=log-messages routing_key=log-messages payload="{}"

On the Telegraf side you should use the routing_key option instead of routing_tag.

[keitalbame]

@danielnelson using the rabbitmqadmin command from both rabbitmq and telegraf hosts, the messages were delivered to the queue.
Configuring telegraf to use routing_key instead of routing_tag did not make any change. I can see data arriving to the exchange but not passing to the queue.

The only thing I can see is that rabbitmqadmin connects to the management port (15672) and telegraf connects to 5672.
I can telnet from telegraf host to rabbitmq on both 15672 and 5672 ports.

[danielnelson]

Let's reopen the issue until the issue can be confirmed.

[danielnelson]

@keitalbame This is indeed a bug in Telegraf where we aren't sending the routing_key on direct exchanges, we'll have the fix in the next release.