feat(mqtt): add MQTT sink #16065
feat(mqtt): add MQTT sink #16065
Conversation
Co-authored-by: Kyle Criddle <kyle.criddle@datadoghq.com>
Co-authored-by: Kyle Criddle <kyle.criddle@datadoghq.com>
Co-authored-by: Kyle Criddle <kyle.criddle@datadoghq.com>
- update to the newer metrics infrastructure - update rumqttc version - fix compilation errors - fix formatting Tested: - Local build
- add quality_of_service option - add mqtt.org link - update documentation Tested: - Local build
✅ Deploy Preview for vrl-playground canceled.
|
✅ Deploy Preview for vector-project canceled.
|
github-actions
bot
added
domain: external docs
Regression Test Results
Run ID: 28afd420-80cc-47d2-850e-9ef7a085578f Explanation
A regression test is an integrated performance test for
The table below, if present, lists those experiments that have experienced a
statistically significant change in their Changes in
Fine details of change detection per experiment.
|
|
Thanks for reviving this @zamazan4ik ! Stephen and myself will be reviewing this shortly. |
|
@jszwedko well, I still cannot resolve the issue, of why the integration tests cannot receive the events from the broker. When I test the sink locally with mqttx, it works fine and I see my events. But integration tests cannot receive events from the broker. |
src/sinks/mqtt/integration_tests.rs
Outdated
| let mut failures = 0; | ||
| let mut message_count = 0; | ||
| while failures < 5 && message_count < input.len() { | ||
| if let Ok(try_msg) = tokio::time::timeout(Duration::from_secs(1), eventloop.poll()).await { |
There was a problem hiding this comment.
If it helps, I can make the test pass by increasing the timeout.
| if let Ok(try_msg) = tokio::time::timeout(Duration::from_secs(1), eventloop.poll()).await { | |
| if let Ok(try_msg) = tokio::time::timeout(Duration::from_secs(10), eventloop.poll()).await { |
I'm not sure if this is a cause for concern. Does it imply that this sink has really low throughput?
|
I wonder if the integration tests needs some modifications. The |
neuronull
added
the
meta: awaiting author
|
@neuronull @zamazan4ik Any updates on this? There hasn't been updates in months. |
It is frustratingly close, but (apart from the merge conflicts now) we just have that failing integration test that needs digging into. We would very happily welcome any contributions if someone wanted to take this on. |
|
@gaby no updates from my side. I still couldn't resolve the integration test issues on my local dev machine and I don't work on this PR anymore. If you want to continue work on it - feel free to do it. |
|
I'd love to have MQTT as a sink! Unfortunately, I have no experience with Rust whatsoever, otherwise I could have given it a shot. |
What better way to learn Rust than to solve a complex concurrency timing issue on a large codebase? |
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
|
Ok, it looks like the issue was caused because when you create a client in
The original code handled both publishing a message and polling the event loop in a single loop. Essentially: loop {
tokio::select! {
connection.poll() => {}
input.next() => { client.publish(..).await }
}
}The docs for poll do state:
Due to the way The downside to this is that now the actual sending of the messages to mqtt is separated from the code that publishes the message. We are unable to determine exactly when or even if that message is successfully sent. Because of this issue we don't get an |
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>
|
So that wasn't the full fix. It turned out the tests were testing everything that came from I'm fairly sure the problem is to do with the integration test and not the main sink code. I can create a new executable with this code and it picks up the integration test messages just fine and without delay. use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, Publish, QoS};
use std::time::Duration;
use tokio::{task, time};
#[tokio::main]
async fn main() {
let mut mqttoptions = MqttOptions::new("rumqtt-async", "localhost", 1883);
mqttoptions.set_keep_alive(Duration::from_secs(5));
let (client, mut eventloop) = AsyncClient::new(mqttoptions, 10);
client.subscribe("test", QoS::AtMostOnce).await.unwrap();
loop {
let notification = eventloop.poll().await.unwrap();
if let Event::Incoming(Incoming::Publish(publish)) = notification {
let message = String::from_utf8_lossy(&publish.payload);
dbg!(message);
}
}
}So I wonder if maybe there are issues with running two EventLoop instances in the same application. I've tried running the client in a |
|
I think that the issue with the integration test is just a race between the subscriber and publisher. When I run the tests as they are, they more often than not first publish all 10 messages and only then gets the client that wants to subscribe a CONNACK message. When it then subscribes, the messages are already handled and since there was no session the messages were not stored for the client. The error I'm seeing with the current code is I think this is also why it works for you when you make another binary. You first run the subscriber and only then start the test so the subscriber gets the messages fast. I tried moving code calling By the way I've added three comments in the code. Feel free to ignore and resolve any or all of them. |
|
|
||
| impl MqttSinkConfig { | ||
| fn build_connector(&self) -> Result<MqttConnector, MqttError> { | ||
| if self.client_id.is_empty() { |
There was a problem hiding this comment.
This should probably also check thath the client_id is 1-23 characters long and only contains alphanumeric characters [a-zA-Z0-9]{1,23} [MQTT-3.1.3-5].|
Also note, that as per [MQTT-3.1.3-6]:
A Server MAY allow a Client to supply a ClientId that has a length of zero bytes
so empty client ID may be valid although not for every server. I think it's reasonable to disallow it here though for simplicity and compatibility concerns along with having static client ID for reconnections.
It seems that rumqttc panics with empty string, but also for strings starging with whitespace, so if that was the concern, I think the check should cover at least that.
But I would personally expect a lot of people try to use either a dash or an underscore so it might be better UX to reject all invalid client IDs early.
There was a problem hiding this comment.
I take this back. The server MUST allow alphanumeric client ID shorter than 24 bytes, but may also allow other IDs.
| let mut options = MqttOptions::new(&self.client_id, &self.host, self.port); | ||
| options.set_keep_alive(Duration::from_secs(self.keep_alive.into())); | ||
| options.set_clean_session(self.clean_session); | ||
| if let (Some(user), Some(password)) = (&self.user, &self.password) { |
There was a problem hiding this comment.
This is ignorable, but I believe mqtt allows for sending only password. rumqttc does not seem to allow that though.
In any case if a vector user provides either a username or password and it is ignored because the other part was not provided, I think it should be at least logged as a warning if not outright considered a bad configuration. Even more so since sending only a password may be what they were trying to do.
| AtMostOnce, | ||
|
|
||
| /// ExactlyOnce. | ||
| #[derivative(Default)] |
There was a problem hiding this comment.
TILI, but I think AtLeastOnce is a better default because that is the guarantee that Vector itself provides. Using exactly once will add two more MQTT messages for each sent event while not really providing the guarantee since messages can still be duplicated.
| } | ||
|
|
||
| fn default_client_id() -> String { | ||
| "vector".to_string() |
There was a problem hiding this comment.
This can be dangerous since two different clients cannot use the same client ID. So if you have two vector instances that should send mqtt messages to the same sink, this could lead to the clients fighting over the same ID disconnecting each other in an endless loop.
I think there must have been the same issue with other sources or sinks, is there some kind of vector function to create a unique client ID? Ideally, it should be stable between restarts but that's more important for a source than a sink.
Otherwise, I think we should use at least something like VectorSink<short hash> and document that multiple mqtt sinks (and sources) connecting to the same broker must not use the same client ID.
|
@StephenWakely @zamazan4ik Would it be fine by you if I took over and reordered the stuff inside the integration test so that this can be merged? I have pretty much working mqtt source which just needs a bit of cleanup but I'd like to piggy back on some of the stuff introduced here such as the integration tests infra. |
Yes please! That would be amazing. |
|
Closing since this is continued in #19813. |
Continues work from #13587
@neuronull I suggest continuing work on the MQTT sink here since the original author is not against it.
I have tried to fix your comments (at least that I am able to fix). Feel free to continue your review.