feat(new source): Initial MQTT Source, #19931

[StormStake]

Summary

The base for this PR is #19931, I have put some work into de-duplicating the code between the MQTT source and sink, I also made a few changes to update some of the older documentation to get it to pass tests. I tried to make as few changes to the overall logic of the source/sink as I could while refactoring but I do need some review and guidance on how the source looks now.

I tried to work on some of the unresolved comments from the original PR but I would like a fresh review so that I can work on anything remaining that the source needs to get merged. Feel free to provide any advice on what I have done, I only really have an okay understanding of Rust and not much in the way of knowledge of Vector's internals but I am prepared to learn if that is needed to get this merged.

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

How did you test this PR?

I used Vector's builtin MQTT sink and the Eclipse Mosquitto MQTT broker along the mosquitto_sub and mosquitto_pub commands provided by Mosquitto to test base functionality. I used the configuration defaults for the mosquitto broker, (No authentication, No TLS, localhost:1883)

Vector MQTT source configuration

[sources.mqtt]
type = "mqtt"
host = "localhost"
topic = "testvectortopic"

[sinks.stdout]
inputs = ["mqtt"]
type = "console"
encoding.codec = "json"

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the "no-changelog" label to this PR.

Checklist

• Please read our Vector contributor resources.
  • make check-all is a good command to run locally. This check is
    defined here. Some of these
    checks might not be relevant to your PR. For Rust changes, at the very least you should run:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run dd-rust-license-tool write to regenerate the license inventory and commit the changes (if any). More details here.

References

• Forked from feat(new source): Initial mqtt source #19931
• Closes: New mqtt source #584

[pront]

Thank you @StormStake! I will take a look.

[ahsandar]

is it due to release in upcoming version ?

[pront]

This looks great. We can release it as a beta component and see if any issues come up.

[pront]

is it due to release in upcoming version ?

Looks likely based on the current state of the PR.

[StormStake]

I noticed that, from what I can tell you can't get what topic the message was sent to from the log event in this version of the source. In my head I think that is pretty fundamental for this type of application, so I think it would make sense to have it added on first release. But I don't really work with MQTT data, so I could be wrong. Otherwise it could always be added in future releases.

Does this look like it's headed in the right
direction for adding that metadata? Or is there a simpler way to manage this?

https://github.com/StormStake/vector/blob/e09bb91a2637be153d888a08fa37b0c08514ce74/src/sources/mqtt/source.rs#L85-L111

[pront]

Thanks @StormStake!

[pront]

I noticed that, from what I can tell you can't get what topic the message was sent to from the log event in this version of the source. In my head I think that is pretty fundamental for this type of application, so I think it would make sense to have it added on first release. But I don't really work with MQTT data, so I could be wrong. Otherwise it could always be added in future releases.

Does this look like it's headed in the right direction for adding that metadata? Or is there a simpler way to manage this?

https://github.com/StormStake/vector/blob/e09bb91a2637be153d888a08fa37b0c08514ce74/src/sources/mqtt/source.rs#L85-L111

Yeah, let's add a topic field. We can draw inspiration from other sources:
https://github.com/vectordotdev/vector/blob/master/src/sources/kafka.rs#L361-L416

[pront]

Happy to see this merged, thanks all!