feat(new sink): Initial rabbitmq sink implementation

[AlyHKafoury]

Signed-off-by: AlyHKafoury aly.kafoury@gmail.com

[AlyHKafoury]

@LucioFranco I've created this as a draft and for you to follow the progress from early stages if it's ok to guide me early for a better path

[LucioFranco]

This is looking like a great start!! Let me know if you have any further questions.

[AlyHKafoury]

@LucioFranco the acker in the integration test gets out a strange number:

thread 'sinks::rabbitmq::integration_test::publish_messages' panicked at 'assertion failed: (left == right)
left: 499500,
right: 1000', src/sinks/rabbitmq.rs:329:5

assert_eq!(
            ack_counter.load(std::sync::atomic::Ordering::Relaxed),
            number_of_events
        );

I don't understand where this value comes from as the sequence number is correct in the sink

[LucioFranco]

@AlyHKafoury so I think there might be something wrong when you decide to ack. It seems that you may be calling https://github.com/timberio/vector/blob/master/src/buffers/mod.rs#L143 more than you think you are? I would assume the code where you call ack just might be getting called a lot more than expected?

[LucioFranco]

@AlyHKafoury were you able to figure it out?

[AlyHKafoury]

@AlyHKafoury were you able to figure it out?

@LucioFranco yea, I was acking the sequence number of the messages instead the count of the messages that was the wrong issue the number 499500 is the summation of all the numbers from 0 to 999 :D :D !!

[LucioFranco]

@AlyHKafoury so I think the code mostly looks good! I think the next step is to add a docker container that we can run the tests aginst. We have a docker feature flag you can add your feature flag to that. Then we just need to add that docker contianer to docker-compose and to circleci test-stable job.

Let me know if you have any trouble, I am happy to help!

[AlyHKafoury]

@LucioFranco Can you please check the latest commit is there something I am missing yet ?

[binarylogic]

@Jeffail tagging you since you have experience with RabbitMQ. 😄

[Jeffail]

Hey @AlyHKafoury, I think it might be worth refactoring the configuration parameters here slightly.

When defining an AMQP sink I would only expect to specify an exchange and a routing key. It's often the case that the routing key will be the same as the target queue, but sometimes the queue bindings will be defined separately and only known to consumers.

I would therefore not expect to see any queue based fields in a sink config, including the ability to declare a queue. However, I would want to be able to declare the target exchange, but this should be optional.

[binarylogic]

@AlyHKafoury thanks again for doing this. Do you plan on making the suggested changes? We'd really appreciate it. If not, we can assign someone to finish this off.

[AlyHKafoury]

@binarylogic Sure I will I just missed the comments.

[binarylogic]

@AlyHKafoury thanks, and no problem if you can't finish this off. We're planning to have @Jeffail work on it this week, so feel free to chime if you plan to do it yourself.

[Jeffail]

Hey @AlyHKafoury, got the changes we need on this branch: https://github.com/timberio/vector/tree/AlyHKafoury-rabbitmq-sink, I'm just finishing up (adding some extra integration tests) and then I'll get it merged.

[Jeffail]

Superseded by: #1376