AMQP Support

[jaxxstorm]

I know this is on the roadmap, but I wanted to create a ticket to track it.

Kafka is an interesting solution, but ideally I'd like AMQP support back ASAP as we already have a RabbitMQ in use.

[henrikjohansen]

+1

[jaxxstorm]

I should add, ideally this would be broke agnostic. There is some talk of utilising ActiveMQ in our company, and I know it's a popular solution due to mcollective using it. Previously only RabbitMQ was available

[RabidDog]

+1

[lennartkoopmann]

Scheduled for v0.20.1 - We are aware that there is big demand for getting AMQP back into Graylog2 and will work on that right after 0.20.0 is out. Probably with a follow up version that brings only this rapidly after the release.

[henrikjohansen]

@lennartkoopmann will this resurrected AMQP support include the ability to let inputs route messages to different queues (like the first invocation of graylog-radio) ?

[lennartkoopmann]

:+1: yes

[henrikjohansen]

<3

[tommymonk]

+1

[h0nIg]

+1

[bonJoeV]

+1

[d-ulyanov]

+1

[ezmorph]

+1

[lennartkoopmann]

Great news everybody! I am now starting to implement this for v0.20.1, which is a small milestone that will come out really soon.

[kroepke]

[h0nIg]

it can happen that the amqp process will flood the queue's after a long period of shutdown. so there are no cpu resources left for the processing, because all resources are used for managing buffers and getting amqp messages and pushing them back. this problem will likely occur for hardware configuration where graylog2/amqp-broker are bundled on one machine. on this issue the qos feature of amqp has to be used and maybe a configuration option should be added. this feature will only get X messages for a request for new messages

channel.basicQos(1);

[02strich]

Is this now part of v0.20.1?

[kroepke]

No, it will be in 0.20.2. 0.20.1 had to be released quickly to fix some bugs.

Best, Kay On Feb 25, 2014 11:13 AM, "Stefan Richter" notifications@github.com wrote:

Is this now part of v0.20.1?

— Reply to this email directly or view it on GitHubhttps://github.com/Graylog2/graylog2-server/issues/385#issuecomment-35992964 .

[jaxxstorm]

Do we have any update? :(

[lennartkoopmann]

yepp, work is going on with this! :)

[agentgt]

I couldn't wait for graylog2 to re-support AMQP so I wrote a small daemon that takes messages off of AMQP and just forwards it to graylog2 via UDP.

https://github.com/agentgt/gelf-amqp-bridge

[lennartkoopmann]

Sorry that this took so much longer than expected.

[h0nIg]

why you are going to ack the message if a failure occured? personally i would prefer that messages are not lost if they are incomplete

[kroepke]

Well, I agree that they shouldn't be lost, but not ack'ing would mean they get redelivered with the same result. Constantly bouncing between the broker and graylog2.

We definitely need a better way of handling incomplete messages, but I think nack is not the correct approach either. :(

On Fri, Apr 4, 2014 at 4:06 PM, Hans-Joachim Kliemeck < notifications@github.com> wrote:

why you are going to ack the message if a failure occured? personally i would prefer that messages are not lost if they are incomplete

— Reply to this email directly or view it on GitHubhttps://github.com/Graylog2/graylog2-server/issues/385#issuecomment-39568179 .

[henrikjohansen]

Incomplete messages should be shipped to a failure queue that can be inspected from the UI - that would make it consistent with the current way of handling input / extractor errors ?

[lennartkoopmann]

Yepp, that's a follow-up issue!

[h0nIg]

btw: is there a reason that the current version is inserting the entries through insertCached instead of the old one that was insertFailFast ? the messages that are inserted are lost on a server restart / crash. instead of inserting them into the cache, they should added via insertFailFast and be returned to the queue if there is no space left on the queue

[dennisoelkers]

h0nig, the idea is that the process/output buffers are limited in size and therefore create backpressure on other parts of the system if there are load spikes. Calling insertCached() means that the message is inserted into a Cache (whose size is limited only by heap size) when the process buffer's capacity is reached. When the server is shut down, it will process all messages in the process/output buffers and caches before it exits, so no messages are lost in this case. You are right about message loss in case of a sudden crash, but this is also the case for all messages in the process/output buffers.

[h0nIg]

You are right about message loss in case of a sudden crash, but this is also the case for all messages in the process/output buffers.

yeah, but why provoking a loss of all messages on the cache instead of only loosing those messages that are on the buffer? i think amqp got a character of NOT loosing a message / loosing only a little bit. (e.g. on high thoughput systems the buffer is much less than the heap limited cache). what do you think about a configuration option that gives more reliability?

[dennisoelkers]

You are right. Actually it does not make sense to do cache-backed inserts for protocols that are queued externally. Thanks much for the hint!

[lennartkoopmann]

There is a scheduled follow-up ticket for this already. One needs to put some thought into the acking and nacking mechanisms for different use cases and we chose the easiest way for now.

Thanks! :)

[henrikjohansen]

@lennartkoopmann @kroepke even thought this issue has been closed we're still lacking the ability to route messages from different radio-inputs to different AMQP queues ...

See my comment from Jan 23 2014 above :)