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

* wip

* add mqtt source cue reference

* add changelog

* Update src/sources/mqtt/config.rs

Co-authored-by: Stephen Wakely <stephen@lisp.space>

* 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 <david.mladek.cz@gmail.com>
Co-authored-by: Stephen Wakely <stephen@lisp.space>
Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: 4 people

Cargo.toml

@@ -580,6 +580,7 @@ sources-logs = [
   "sources-kafka",
   "sources-kubernetes_logs",
   "sources-logstash",
+  "sources-mqtt",
   "sources-nats",
   "sources-opentelemetry",
   "sources-pulsar",
@@ -635,6 +636,7 @@ sources-kafka = ["dep:rdkafka"]
 sources-kubernetes_logs = ["vector-lib/file-source", "kubernetes", "transforms-reduce"]
 sources-logstash = ["sources-utils-net-tcp", "tokio-util/net"]
 sources-mongodb_metrics = ["dep:mongodb"]
+sources-mqtt = ["dep:rumqttc"]
 sources-nats = ["dep:async-nats", "dep:nkeys"]
 sources-nginx_metrics = ["dep:nom"]
 sources-opentelemetry = ["dep:hex", "vector-lib/opentelemetry", "dep:prost", "dep:prost-types", "sources-http_server", "sources-utils-http", "sources-utils-http-headers", "sources-vector"]
@@ -934,7 +936,7 @@ kafka-integration-tests = ["sinks-kafka", "sources-kafka"]
 logstash-integration-tests = ["docker", "sources-logstash"]
 loki-integration-tests = ["sinks-loki"]
 mongodb_metrics-integration-tests = ["sources-mongodb_metrics"]
-mqtt-integration-tests = ["sinks-mqtt"]
+mqtt-integration-tests = ["sinks-mqtt", "sources-mqtt"]
 nats-integration-tests = ["sinks-nats", "sources-nats"]
 nginx-integration-tests = ["sources-nginx_metrics"]
 opentelemetry-integration-tests = ["sources-opentelemetry", "dep:prost"]

changelog.d/19931_add_mqtt_source.feature.md

@@ -0,0 +1,3 @@
+Adds a new `mqtt` source enabling Vector to receive logs from a MQTT broker.
+
+authors: mladedav pront StormStake

scripts/integration/mqtt/test.yaml

@@ -10,3 +10,4 @@ paths:
 - "src/internal_events/mqtt.rs"
 - "src/sinks/mqtt/**"
 - "src/sinks/util/**"
+- "src/sources/mqtt/**"

src/common/mod.rs

@@ -18,6 +18,10 @@ pub(crate) mod sqs;
 #[cfg(any(feature = "sources-aws_s3", feature = "sinks-aws_s3"))]
 pub(crate) mod s3;
 
+#[cfg(any(feature = "sources-mqtt", feature = "sinks-mqtt",))]
+/// Common MQTT configuration shared by MQTT components.
+pub mod mqtt;
+
 #[cfg(any(feature = "transforms-log_to_metric", feature = "sinks-loki"))]
 pub(crate) mod expansion;
 

src/common/mqtt.rs

@@ -0,0 +1,123 @@
+use rumqttc::{AsyncClient, EventLoop, MqttOptions};
+use snafu::Snafu;
+use vector_config_macros::configurable_component;
+use vector_lib::tls::{TlsEnableableConfig, TlsError};
+
+use crate::template::TemplateParseError;
+
+/// Shared MQTT configuration for sources and sinks.
+#[configurable_component]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(deny_unknown_fields)]
+pub struct MqttCommonConfig {
+    /// 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.
+    #[serde(default)]
+    #[configurable(derived)]
+    pub user: Option<String>,
+
+    /// MQTT password.
+    #[serde(default)]
+    #[configurable(derived)]
+    pub password: Option<String>,
+
+    /// MQTT client ID.
+    #[serde(default)]
+    #[configurable(derived)]
+    pub client_id: Option<String>,
+
+    /// Connection keep-alive interval.
+    #[serde(default = "default_keep_alive")]
+    #[derivative(Default(value = "default_keep_alive()"))]
+    pub keep_alive: u16,
+
+    /// TLS configuration.
+    #[configurable(derived)]
+    pub tls: Option<TlsEnableableConfig>,
+}
+
+const fn default_port() -> u16 {
+    1883
+}
+
+const fn default_keep_alive() -> u16 {
+    60
+}
+
+/// MQTT Error Types
+#[derive(Debug, Snafu)]
+#[snafu(visibility(pub))]
+pub enum MqttError {
+    /// Topic template parsing failed
+    #[snafu(display("invalid topic template: {source}"))]
+    TopicTemplate {
+        /// Source of error
+        source: TemplateParseError,
+    },
+    /// TLS error
+    #[snafu(display("TLS error: {source}"))]
+    Tls {
+        /// Source of error
+        source: TlsError,
+    },
+    /// Configuration error
+    #[snafu(display("MQTT configuration error: {source}"))]
+    Configuration {
+        /// Source of error
+        source: ConfigurationError,
+    },
+}
+
+/// MQTT Configuration error types
+#[derive(Clone, Debug, Eq, PartialEq, Snafu)]
+pub enum ConfigurationError {
+    /// Empty client ID error
+    #[snafu(display("Client ID is not allowed to be empty."))]
+    EmptyClientId,
+    /// Invalid credentials provided error
+    #[snafu(display("Username and password must be either both provided or both missing."))]
+    InvalidCredentials,
+    /// Invalid client ID provied error
+    #[snafu(display(
+        "Client ID must be 1-23 characters long and must consist of only alphanumeric characters."
+    ))]
+    InvalidClientId,
+    /// Credentials provided were incomplete
+    #[snafu(display("Username and password must be either both or neither provided."))]
+    IncompleteCredentials,
+}
+
+#[derive(Clone)]
+/// Mqtt connector wrapper
+pub struct MqttConnector {
+    /// Mqtt connection options
+    pub options: MqttOptions,
+}
+
+impl MqttConnector {
+    /// Creates a new MqttConnector
+    pub const fn new(options: MqttOptions) -> Self {
+        Self { options }
+    }
+
+    /// Connects the connector and generates a client and eventloop
+    pub fn connect(&self) -> (AsyncClient, EventLoop) {
+        let (client, eventloop) = AsyncClient::new(self.options.clone(), 1024);
+        (client, eventloop)
+    }
+
+    /// TODO: Right now there is no way to implement the healthcheck properly: <https://github.com/bytebeamio/rumqtt/issues/562>
+    pub async fn healthcheck(&self) -> crate::Result<()> {
+        Ok(())
+    }
+}

src/sinks/mqtt/config.rs

@@ -2,53 +2,32 @@ use std::time::Duration;
 
 use rand::Rng;
 use rumqttc::{MqttOptions, QoS, TlsConfiguration, Transport};
-use snafu::{ResultExt, Snafu};
+use snafu::ResultExt;
 use vector_lib::codecs::JsonSerializerConfig;
 
 use crate::template::Template;
 use crate::{
     codecs::EncodingConfig,
-    config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext},
-    sinks::{
-        mqtt::sink::{ConfigurationSnafu, MqttConnector, MqttError, MqttSink, TlsSnafu},
-        prelude::*,
-        Healthcheck, VectorSink,
+    common::mqtt::{
+        ConfigurationError, ConfigurationSnafu, MqttCommonConfig, MqttConnector, MqttError,
+        TlsSnafu,
     },
-    tls::{MaybeTlsSettings, TlsEnableableConfig},
+    config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext},
+    sinks::{mqtt::sink::MqttSink, prelude::*, Healthcheck, VectorSink},
+    tls::MaybeTlsSettings,
 };
 
 /// Configuration for the `mqtt` sink
 #[configurable_component(sink("mqtt"))]
 #[derive(Clone, Debug)]
 pub struct MqttSinkConfig {
-    /// 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.
-    #[serde(default = "default_port")]
-    pub port: u16,
-
-    /// MQTT username.
-    pub user: Option<String>,
-
-    /// MQTT password.
-    pub password: Option<String>,
-
-    /// MQTT client ID.
-    pub client_id: Option<String>,
-
-    /// Connection keep-alive interval.
-    #[serde(default = "default_keep_alive")]
-    pub keep_alive: u16,
+    #[serde(flatten)]
+    pub common: MqttCommonConfig,
 
     /// If set to true, the MQTT session is cleaned on login.
     #[serde(default = "default_clean_session")]
     pub clean_session: bool,
 
-    #[configurable(derived)]
-    pub tls: Option<TlsEnableableConfig>,
-
     /// MQTT publish topic (templates allowed)
     pub topic: Template,
 
@@ -100,14 +79,6 @@ impl From<MqttQoS> for QoS {
     }
 }
 
-const fn default_port() -> u16 {
-    1883
-}
-
-const fn default_keep_alive() -> u16 {
-    60
-}
-
 const fn default_clean_session() -> bool {
     false
 }
@@ -123,14 +94,9 @@ const fn default_retain() -> bool {
 impl Default for MqttSinkConfig {
     fn default() -> Self {
         Self {
-            host: "localhost".into(),
-            port: default_port(),
-            user: None,
-            password: None,
-            client_id: None,
-            keep_alive: default_keep_alive(),
+            common: MqttCommonConfig::default(),
             clean_session: default_clean_session(),
-            tls: None,
+
             topic: Template::try_from("vector").expect("Cannot parse as a template"),
             retain: default_retain(),
             encoding: JsonSerializerConfig::default().into(),
@@ -164,17 +130,9 @@ impl SinkConfig for MqttSinkConfig {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, Snafu)]
-pub enum ConfigurationError {
-    #[snafu(display("Client ID is not allowed to be empty."))]
-    EmptyClientId,
-    #[snafu(display("Username and password must be either both provided or both missing."))]
-    InvalidCredentials,
-}
-
 impl MqttSinkConfig {
     fn build_connector(&self) -> Result<MqttConnector, MqttError> {
-        let client_id = self.client_id.clone().unwrap_or_else(|| {
+        let client_id = self.common.client_id.clone().unwrap_or_else(|| {
             let hash = rand::rng()
                 .sample_iter(&rand_distr::Alphanumeric)
                 .take(6)
@@ -186,11 +144,12 @@ impl MqttSinkConfig {
         if client_id.is_empty() {
             return Err(ConfigurationError::EmptyClientId).context(ConfigurationSnafu);
         }
-        let tls = MaybeTlsSettings::from_config(self.tls.as_ref(), false).context(TlsSnafu)?;
-        let mut options = MqttOptions::new(&client_id, &self.host, self.port);
-        options.set_keep_alive(Duration::from_secs(self.keep_alive.into()));
+        let tls =
+            MaybeTlsSettings::from_config(self.common.tls.as_ref(), false).context(TlsSnafu)?;
+        let mut options = MqttOptions::new(&client_id, &self.common.host, self.common.port);
+        options.set_keep_alive(Duration::from_secs(self.common.keep_alive.into()));
         options.set_clean_session(self.clean_session);
-        match (&self.user, &self.password) {
+        match (&self.common.user, &self.common.password) {
             (Some(user), Some(password)) => {
                 options.set_credentials(user, password);
             }
@@ -211,7 +170,7 @@ impl MqttSinkConfig {
                 alpn,
             }));
         }
-        MqttConnector::new(options, self.topic.to_string())
+        Ok(MqttConnector::new(options))
     }
 }
 

src/sinks/mqtt/integration_tests.rs

@@ -1,3 +1,4 @@
+use crate::common::mqtt::MqttCommonConfig;
 use crate::config::{SinkConfig, SinkContext};
 use crate::sinks::mqtt::config::MqttQoS;
 use crate::sinks::mqtt::MqttSinkConfig;
@@ -25,16 +26,21 @@ async fn mqtt_happy() {
     trace_init();
 
     let topic = "test";
-    let cnf = MqttSinkConfig {
+    let common = MqttCommonConfig {
         host: mqtt_broker_address(),
         port: mqtt_broker_port(),
+        ..Default::default()
+    };
+
+    let config = MqttSinkConfig {
+        common,
         topic: Template::try_from(topic).expect("Cannot parse the topic template"),
         quality_of_service: MqttQoS::AtLeastOnce,
         ..Default::default()
     };
 
     let cx = SinkContext::default();
-    let (sink, healthcheck) = cnf.build(cx).await.expect("Cannot build the sink");
+    let (sink, healthcheck) = config.build(cx).await.expect("Cannot build the sink");
     healthcheck.await.expect("Health check failed");
 
     // prepare consumer