feat(new source): Initial `mqtt` source #19931

mladedav · 2024-02-22T17:52:13Z

Closes #584

I made this initial implementation farily simple, one important point is that I have not implemented acknowledgements.

StephenWakely

Nice. This is looking good and will be great to have to complement the mqtt sink.

There's just a few fairly minor changes that I can see.

StephenWakely · 2024-02-28T14:51:03Z

src/sources/mqtt/config.rs

+use vector_lib::config::{LegacyKey, LogNamespace};
+use vector_lib::configurable::configurable_component;
+use vector_lib::lookup::owned_value_path;
+use vector_lib::tls::{MaybeTlsSettings, TlsEnableableConfig};


Can we group these imports.

Done.

For the record nightly rustfmt can do this automatically, I think it's worth it to consider it. We've been using it for some time now and it seems stable though we don't really use latest features.

Nice. We'd need to wait for that to move into stable since we have so many different contributors and we shouldn't dictate the nightly versions. But I'm definitely looking forward to that.

src/sources/mqtt/config.rs

StephenWakely · 2024-02-28T16:23:14Z

src/sources/mqtt/source.rs

+        })
+    }
+
+    pub async fn run(self, mut out: SourceSender, _shutdown: ShutdownSignal) -> Result<(), ()> {


We shouldn't ignore the ShutdownSignal otherwise Vector just gets stuck when it tries to shutdown and needs to be force killed.

Ok, fixed based on the suggestion you posted later. I tried to copy some other source and delete stuff that didn't seem needed for this and this was caught in that.

StephenWakely · 2024-02-28T16:25:29Z

src/sources/mqtt/source.rs

+            match connection.poll().await {
+                Ok(MqttEvent::Incoming(Incoming::Publish(publish))) => {
+                    self.process_message(publish, &mut out).await;
+                }
+                Ok(MqttEvent::Incoming(
+                    Incoming::PubAck(_) | Incoming::PubRec(_) | Incoming::PubComp(_),
+                )) => {
+                    // TODO Handle acknowledgement
+                }
+                _ => {}
+            }


In pseudo code something like:

tokio::select! { _ = shudown => break; poll = connection.poll() => { match poll { .... } } }

StephenWakely · 2024-02-29T12:07:17Z

/ci-run-component-features

Co-authored-by: Stephen Wakely <[email protected]>

neuronull

🎉

neuronull · 2024-03-01T19:42:06Z

.vscode/settings.json

@@ -0,0 +1,6 @@
+{


🔨 warning: This config shouldn't be committed, it can instead be added to your global gitignore setting.‏

Sorry, my mistake. I fixed my gitignore, thanks for the suggestion.

neuronull · 2024-03-01T19:43:40Z

src/lib.rs

@@ -4,7 +4,6 @@
 #![deny(unused_allocation)]
 #![deny(unused_assignments)]
 #![deny(unused_comparisons)]
-#![deny(warnings)]


❓ question: ‏is there a good reason to delete this? We want to keep a clean code base.

No, sorry. I didn't mean to commit this.

It was just more convenient for me when warnings and clippy pedantic such as unused import did not abort compilation when I just wanted to check tests after some changes that I didn't properly clean up yet. I put it back.

neuronull · 2024-03-01T19:45:14Z

src/sources/mqtt/config.rs

+    InvalidClientId,
+
+    #[snafu(display("Username and password must be either both or neither provided."))]
+    BadCredentials,


🥜 nitpick: IncompleteCredentials is a bit more specific

neuronull · 2024-03-01T19:47:54Z

src/sources/mqtt/config.rs

+    /// MQTT server address (The broker’s domain name or IP address).
+    #[configurable(metadata(docs::examples = "mqtt.example.com", docs::examples = "127.0.0.1"))]
+    pub host: String,
+
+    /// TCP port of the MQTT server to connect to.
+    #[configurable(derived)]
+    #[serde(default = "default_port")]
+    #[derivative(Default(value = "default_port()"))]
+    pub port: u16,
+
+    /// MQTT username.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub user: Option<String>,
+
+    /// MQTT password.
+    #[configurable(derived)]
+    #[serde(default)]
+    pub password: Option<String>,
+
+    /// MQTT client ID. If there are multiple
+    #[configurable(derived)]
+    #[serde(default)]
+    pub client_id: Option<String>,
+
+    /// Connection keep-alive interval.
+    #[configurable(derived)]
+    #[serde(default = "default_keep_alive")]
+    #[derivative(Default(value = "default_keep_alive()"))]
+    pub keep_alive: u16,
+
+    /// MQTT topic from which messages are to be read.
+    #[configurable(derived)]
+    #[serde(default = "default_topic")]
+    #[derivative(Default(value = "default_topic()"))]
+    pub topic: String,


💭 thought: ‏These appear to be exact replicas of the same settings for the sink. We could consider a common struct they both use with serde_flatten. This could be located in a shared space like src/common/mqtt.rs

Yes, there's a lot of duplication here.

I'm also in favor of creating a shared module, I just didn't get around to do it yet.

neuronull · 2024-03-01T19:48:34Z

src/sources/mqtt/integration_tests.rs

@@ -0,0 +1,111 @@
+#![cfg(feature = "mqtt-integration-tests")]


👏 praise: ‏thanks for adding integration tests 🎉

neuronull · 2024-03-01T19:51:34Z

src/sources/mqtt/source.rs

+#[derive(Debug, Snafu)]
+#[snafu(visibility(pub))]
+pub enum MqttError {
+    #[snafu(display("TLS error: {}", source))]
+    Tls { source: TlsError },
+    #[snafu(display("MQTT configuration error: {}", source))]
+    Configuration { source: ConfigurationError },
+}
+
+#[derive(Clone)]
+pub struct MqttConnector {
+    options: MqttOptions,
+    topic: String,
+}
+
+impl MqttConnector {
+    pub const fn new(options: MqttOptions, topic: String) -> Result<Self, MqttError> {
+        Ok(Self { options, topic })
+    }
+
+    async fn connect(&self) -> Result<(AsyncClient, EventLoop), ClientError> {
+        let (client, eventloop) = AsyncClient::new(self.options.clone(), 1024);
+        client.subscribe(&self.topic, QoS::AtLeastOnce).await?;
+        Ok((client, eventloop))
+    }
+}


💬 suggestion: ‏this also looks like high overlap with the same objects in the sink, could it be in the common mqtt module?

src/sources/mqtt/source.rs

neuronull · 2024-03-01T20:24:38Z

src/sources/mqtt/source.rs

+                        )) => {
+                            // TODO Handle acknowledgement
+                        }
+                        _ => {}


💭 thought: ‏Am I reading this correctly that we are going to silently drop on errors?

If by drop you mean drop incoming messages, than no, that's should not be the error case here.

The poll method returns all incoming packets of which we only care about publishes (and hopefully later also about acks and such), but also completely irrelevant stuff like pong. It also returns all our outgoing packets like ping, acks, etc. This all could be handled by Ok(_).

The Err variant here is error of the protocol. In real life it's going to be most likely network timeout, pong not arriving in a timely manner and possibly misbehaving broker. But continuing to call poll should make the underlying library reconnect and everything should continue.

Now that I describe it, this should probably be logged as a warning/error as disconnection can eventually lead to lost messages.

Now that I describe it, this should probably be logged as a warning/error as disconnection can eventually lead to lost messages.

Sounds good to me. I think adding some more in code comments to what you just mentioned also would be helpful.

neuronull · 2024-03-01T20:26:01Z

src/sources/mqtt/source.rs

+            protocol: "mqtt",
+            endpoint: &self.connector.options.broker_address().0,
+        });
+        let events_received = register!(EventsReceived);


🔨 warning: ‏this should be saved as state in the struct, only registering it once on initialization.

Ok, fixed. I've taken most inspiration from the sqs source (for no specific reason) and there they seem to clone the handle even though it's used only as a shared reference. I don't clone it as much but maybe there's a reason to do so?

Just glancing at the aws_sqs source it looks to me like they're only cloning it once per the client_concurrency setting (which defaults to the number of threads on the system) , at initialization of the source.

Since the mqtt source isn't managing concurrency directly there shouldn't be a need to clone the EventsReceived registry.

neuronull · 2024-03-01T20:27:10Z

website/content/en/docs/reference/configuration/sources/mqtt.md

@@ -0,0 +1,14 @@
+---
+title: MQTT
+description: Receive observability event data from an [MQTT](https://mqtt.org) broker


💬 suggestion: ‏Since we're being specific to log data, I think we should keep that consistent here.

dmvinson · 2024-07-31T17:15:29Z

Hi, is this PR still actively being worked on? Would love to see an MQTT sink shipped, happy to contribute to that effort

mladedav · 2024-07-31T21:36:56Z

Sorry, I forgot about it. I believe the only major thing I left unresolved is deduplication between the sink and source but don't quote me on that.

Feel free to push it over the finish line. I can't promise I'd have time to get back to this in the coming weeks and definitely not in the coming days.

yonas · 2024-09-28T11:59:57Z

Looking forward to having this feature. 🥳

Is this ready for merge?

aliciascott · 2024-11-07T19:25:26Z

Hi @mladedav I see this is quite stale, is this PR still being worked on or should it be closed ?

mladedav · 2024-11-10T23:19:33Z

I'm obviously biased but I think it would be a shame to close this.

IIRC the code works (unless there are conflicts now with the main branch) and the only thing I didn't get around to was code deduplication between the mqtt source and sink.

pront · 2024-11-11T17:06:01Z

I'm obviously biased but I think it would be a shame to close this.

IIRC the code works (unless there are conflicts now with the main branch) and the only thing I didn't get around to was code deduplication between the mqtt source and sink.

Thanks @mladedav, are you motivated to take this over the finish line? I can help with fast reviews if so.

ahsandar · 2024-12-05T11:21:53Z

would be great to have this in vector in the next release

pront · 2024-12-05T16:07:01Z

I will (1) review, (2) if it's not too much effort address nits. If it turns out this need a lot of dev work, I will leave it up to community to take it over the finish line.

outdated

ahsandar · 2024-12-11T12:23:09Z

@pront can this be expected to be merged anytime soon ?

pront · 2024-12-11T15:13:25Z

@pront can this be expected to be merged anytime soon ?

The PR still needs some work. But if I recall correctly it's almost there.

ahsandar · 2025-03-05T04:21:02Z

Are we there yet ?

pront · 2025-03-05T18:40:19Z

Are we there yet ?

There hasn't been any activity on this PR for a while. If some community member wants to fork it or work on this directly, I am happy to review.

pront · 2025-03-31T16:52:52Z

Closed in favor of #22752

* wip * add mqtt source cue reference * add changelog * Update src/sources/mqtt/config.rs Co-authored-by: Stephen Wakely <[email protected]> * rustfmt * honor shutdown signa * git rm .vscode/settings.json * add deny warnings back * use new_maybe_logs * fix type, changelog, rename bad creds * add fowllow up github issue * common config WIP * Consolidated MQTT error types * More deduplication in MQTT error types * Deduplicated MQTT connector type * Formatting fixes for tests * Fixed docs tests * Fixed metrics and integration tests for MQTT * Added to changelog authorship * Fixed cue docs formating * Formatting fix on errors * Added topic metadata --------- Co-authored-by: David Mládek <[email protected]> Co-authored-by: Stephen Wakely <[email protected]> Co-authored-by: Pavlos Rontidis <[email protected]>

mladedav added 2 commits February 21, 2024 23:59

wip

004ca96

add mqtt source cue reference

8800f29

mladedav requested review from a team as code owners February 22, 2024 17:52

mladedav requested a review from a team February 22, 2024 17:52

github-actions bot added domain: sources Anything related to the Vector's sources domain: ci Anything related to Vector's CI environment domain: external docs Anything related to Vector's external, public documentation labels Feb 22, 2024

add changelog

09c17b8

aliciascott approved these changes Feb 22, 2024

View reviewed changes

jszwedko requested a review from StephenWakely February 26, 2024 16:52

sghall requested review from sghall and removed request for sghall February 26, 2024 22:17

neuronull self-requested a review February 26, 2024 23:35

StephenWakely previously requested changes Feb 29, 2024

View reviewed changes

mladedav and others added 3 commits February 29, 2024 19:05

Update src/sources/mqtt/config.rs

6531221

Co-authored-by: Stephen Wakely <[email protected]>

rustfmt

f56386b

honor shutdown signa

85f2c6c

neuronull previously requested changes Mar 1, 2024

View reviewed changes

jszwedko requested a review from a team as a code owner October 3, 2024 18:54

pront requested a review from neuronull December 5, 2024 16:52

pront requested a review from StephenWakely December 5, 2024 16:53

pront added 5 commits December 5, 2024 11:59

Merge remote-tracking branch 'origin' into dm/mqtt-source

0e2c1a0

git rm .vscode/settings.json

40a4f73

add deny warnings back

a39d115

use new_maybe_logs

2a49f4d

fix type, changelog, rename bad creds

b989b04

pront added the source: mqtt label Dec 5, 2024

pront added 2 commits December 5, 2024 12:11

add fowllow up github issue

da06b83

common config WIP

41d7c76

github-actions bot added the domain: sinks Anything related to the Vector's sinks label Dec 5, 2024

StormStake mentioned this pull request Mar 30, 2025

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

Merged

10 tasks

pront closed this Mar 31, 2025

		@@ -0,0 +1,111 @@
		#![cfg(feature = "mqtt-integration-tests")]

feat(new source): Initial mqtt source #19931

feat(new source): Initial mqtt source #19931

Uh oh!

Conversation

mladedav commented Feb 22, 2024

Uh oh!

StephenWakely left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

StephenWakely commented Feb 29, 2024

Uh oh!

neuronull left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dmvinson commented Jul 31, 2024

Uh oh!

mladedav commented Jul 31, 2024

Uh oh!

yonas commented Sep 28, 2024

Uh oh!

aliciascott commented Nov 7, 2024

Uh oh!

mladedav commented Nov 10, 2024

Uh oh!

pront commented Nov 11, 2024

Uh oh!

ahsandar commented Dec 5, 2024

Uh oh!

pront commented Dec 5, 2024

Uh oh!

feat(new source): Initial `mqtt` source #19931

feat(new source): Initial `mqtt` source #19931