diff --git a/Cargo.toml b/Cargo.toml index 5b0b7e3b43..3d304fff8d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ svi = "1.2.0" # ASYNC reqwest = { version = "0.13.3", default-features = false, features = ["json", "stream", "form", "query", "rustls"] } +rumqttc = "0.25.0" tokio = { version = "1.52.2", features = ["full"] } tokio-util = { version = "0.7.18", features = ["io", "codec"] } tokio-stream = { version = "0.1.18", features = ["sync"] } @@ -137,4 +138,4 @@ bytes = "1.11.1" regex = "1.12.3" [profile.release] -strip = "debuginfo" \ No newline at end of file +strip = "debuginfo" diff --git a/MQTT_ALERTER_PLAN.md b/MQTT_ALERTER_PLAN.md new file mode 100644 index 0000000000..f327f47aa7 --- /dev/null +++ b/MQTT_ALERTER_PLAN.md @@ -0,0 +1,39 @@ +# MQTT Alerter Plan + +## Minimum Files To Modify + +- `client/core/rs/src/entities/alerter.rs` + - Add a new `AlerterEndpoint::Mqtt` variant and `MqttAlerterEndpoint` config struct. + - Define defaults (topic `komodo/events`, QoS default) and optional auth/client fields. + - Keeps API/resource model aligned for backend + UI. + +- `client/core/ts/src/types.ts` + - Add generated-type equivalents for the new Rust endpoint variant and params. + - Ensures UI can type-check new MQTT config fields. + +- `bin/core/src/alert/mod.rs` + - Register `mod mqtt;`. + - Add endpoint dispatch arm for `AlerterEndpoint::Mqtt`. + +- `bin/core/src/alert/mqtt.rs` (new) + - Implement MQTT transport: connect, serialize existing `Alert` JSON unchanged, publish, and return success/failure. + +- `Cargo.toml` and `bin/core/Cargo.toml` + - Add MQTT client crate dependency needed by core alert transport. + +- `ui/src/resources/alerter/config/endpoint.tsx` + - Add `MQTT` to endpoint selector. + - Render MQTT-specific config fields: broker URL, topic, username, password, client ID, QoS, retain. + +- `docsite/docs/resources.md` + - Small docs update to include MQTT as a native Alerter destination. + +## Recommended Rust MQTT Library + +Use `rumqttc`. + +Why: +- Pure Rust + Tokio-native async flow (fits Komodo core runtime). +- Lightweight for one-shot publish operations. +- Actively used in Rust MQTT integrations. +- Supports auth, QoS levels, retain flag, and broker URL parsing patterns we need. diff --git a/bin/core/Cargo.toml b/bin/core/Cargo.toml index 19cd11b64e..9b46bb6409 100644 --- a/bin/core/Cargo.toml +++ b/bin/core/Cargo.toml @@ -64,6 +64,7 @@ serde_qs.workspace = true colored.workspace = true tracing.workspace = true reqwest.workspace = true +rumqttc.workspace = true dotenvy.workspace = true anyhow.workspace = true bcrypt.workspace = true @@ -83,4 +84,4 @@ envy.workspace = true hmac.workspace = true sha2.workspace = true hex.workspace = true -url.workspace = true \ No newline at end of file +url.workspace = true diff --git a/bin/core/src/alert/mod.rs b/bin/core/src/alert/mod.rs index c015c41d05..4b5ce1abf9 100644 --- a/bin/core/src/alert/mod.rs +++ b/bin/core/src/alert/mod.rs @@ -18,6 +18,7 @@ use crate::helpers::{ use crate::{config::core_config, state::db_client}; mod discord; +mod mqtt; mod ntfy; mod pushover; mod slack; @@ -148,6 +149,14 @@ pub async fn send_alert_to_alerter( ) }) } + AlerterEndpoint::Mqtt(mqtt_endpoint) => mqtt::send_alert( + mqtt_endpoint, + alert, + ) + .await + .with_context(|| { + format!("Failed to send alert to MQTT Alerter {}", alerter.name) + }), } } diff --git a/bin/core/src/alert/mqtt.rs b/bin/core/src/alert/mqtt.rs new file mode 100644 index 0000000000..09af294827 --- /dev/null +++ b/bin/core/src/alert/mqtt.rs @@ -0,0 +1,162 @@ +use std::time::Duration; + +use anyhow::{Context, anyhow, bail}; +use komodo_client::entities::alert::AlertDataVariant; +use rumqttc::{AsyncClient, MqttOptions, Packet, QoS}; +use url::Url; + +use super::*; + +pub async fn send_alert( + endpoint: &MqttAlerterEndpoint, + alert: &Alert, +) -> anyhow::Result<()> { + let VariablesAndSecrets { variables, secrets } = + get_variables_and_secrets().await?; + + let mut broker_url = endpoint.broker_url.clone(); + let mut topic = endpoint.topic.clone(); + let mut username = endpoint.username.clone(); + let mut password = endpoint.password.clone(); + let mut client_id = endpoint.client_id.clone(); + + let mut interpolator = + Interpolator::new(Some(&variables), &secrets); + + let res = async { + interpolator.interpolate_string(&mut broker_url)?; + interpolator.interpolate_string(&mut topic)?; + if let Some(value) = username.as_mut() { + interpolator.interpolate_string(value)?; + } + if let Some(value) = password.as_mut() { + interpolator.interpolate_string(value)?; + } + if let Some(value) = client_id.as_mut() { + interpolator.interpolate_string(value)?; + } + let topic = topic; + + send_message( + &broker_url, + &topic, + username.as_deref(), + password.as_deref(), + client_id.as_deref(), + endpoint.qos, + endpoint.retain, + alert, + ) + .await + } + .await; + + res.map_err(|e| { + let replacers = interpolator + .secret_replacers + .into_iter() + .collect::>(); + let sanitized_error = + svi::replace_in_string(&format!("{e:?}"), &replacers); + anyhow!("Error with publish to MQTT: {sanitized_error}") + }) +} + +async fn send_message( + broker_url: &str, + topic: &str, + username: Option<&str>, + password: Option<&str>, + client_id: Option<&str>, + qos: u8, + retain: bool, + alert: &Alert, +) -> anyhow::Result<()> { + if topic.trim().is_empty() { + bail!("MQTT topic cannot be empty"); + } + + let parsed = Url::parse(broker_url).with_context(|| { + format!("Invalid MQTT broker URL: {broker_url}") + })?; + let scheme = parsed.scheme(); + if !matches!(scheme, "mqtt" | "tcp") { + bail!( + "Unsupported MQTT broker URL scheme `{scheme}`. Use mqtt:// or tcp://" + ); + } + + let host = parsed + .host_str() + .context("MQTT broker URL must include a host")?; + let port = parsed.port().unwrap_or(1883); + + let client_id = client_id + .filter(|id| !id.trim().is_empty()) + .map(ToOwned::to_owned) + .unwrap_or_else(|| format!("komodo-alert-{}", uuid::Uuid::new_v4())); + + let mut options = MqttOptions::new(client_id, host, port); + options.set_keep_alive(Duration::from_secs(10)); + + let username = username.filter(|value| !value.trim().is_empty()); + let password = password.filter(|value| !value.trim().is_empty()); + let url_username = parsed.username(); + let url_password = parsed.password().filter(|value| !value.is_empty()); + match (username, password) { + (Some(user), Some(pass)) => { + options.set_credentials(user, pass); + } + (Some(user), None) => { + options.set_credentials(user, ""); + } + (None, Some(pass)) => { + options.set_credentials("", pass); + } + (None, None) if !url_username.is_empty() => { + options.set_credentials(url_username, url_password.unwrap_or("")); + } + _ => {} + } + + let payload = serde_json::to_vec(alert) + .context("Failed to serialize alert payload to JSON")?; + let qos = to_qos(qos)?; + + let (client, mut eventloop) = AsyncClient::new(options, 10); + client + .publish(topic, qos, retain, payload) + .await + .context("Failed queuing MQTT publish")?; + + let poll_count = if qos == QoS::AtMostOnce { 2 } else { 4 }; + for _ in 0..poll_count { + let event = tokio::time::timeout( + Duration::from_secs(5), + eventloop.poll(), + ) + .await + .context("Timed out waiting for MQTT broker response")? + .context("MQTT event loop failed while publishing")?; + + if matches!( + event, + rumqttc::Event::Incoming(Packet::PubAck(_)) + | rumqttc::Event::Incoming(Packet::PubComp(_)) + ) { + break; + } + } + + client.disconnect().await.ok(); + Ok(()) +} + +fn to_qos(qos: u8) -> anyhow::Result { + match qos { + 0 => Ok(QoS::AtMostOnce), + 1 => Ok(QoS::AtLeastOnce), + 2 => Ok(QoS::ExactlyOnce), + _ => bail!("Invalid MQTT QoS `{qos}`. Must be 0, 1, or 2"), + } +} diff --git a/client/core/rs/src/entities/alerter.rs b/client/core/rs/src/entities/alerter.rs index 7a4d9c2d6f..af40235a3d 100644 --- a/client/core/rs/src/entities/alerter.rs +++ b/client/core/rs/src/entities/alerter.rs @@ -170,6 +170,9 @@ pub enum AlerterEndpoint { /// Send alert to Pushover Pushover(PushoverAlerterEndpoint), + + /// Send alert JSON to an MQTT broker + Mqtt(MqttAlerterEndpoint), } impl Default for AlerterEndpoint { @@ -319,6 +322,66 @@ fn default_pushover_url() -> String { ) } +/// Configuration for an MQTT alerter. +#[typeshare] +#[derive( + Debug, Clone, PartialEq, Serialize, Deserialize, Builder, +)] +#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))] +#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))] +pub struct MqttAlerterEndpoint { + /// MQTT broker URL. Example: `mqtt://localhost:1883` + #[serde(default = "default_mqtt_broker_url")] + #[builder(default = "default_mqtt_broker_url()")] + pub broker_url: String, + + /// Topic to publish alerts to + #[serde(default = "default_mqtt_topic")] + #[builder(default = "default_mqtt_topic()")] + pub topic: String, + + /// Optional username for broker authentication + pub username: Option, + + /// Optional password for broker authentication + pub password: Option, + + /// Optional client identifier. If empty, core will generate one. + pub client_id: Option, + + /// MQTT QoS level: 0, 1, or 2 + #[serde(default)] + #[builder(default)] + pub qos: u8, + + /// Whether the broker should retain the message + #[serde(default)] + #[builder(default)] + pub retain: bool, +} + +impl Default for MqttAlerterEndpoint { + fn default() -> Self { + Self { + broker_url: default_mqtt_broker_url(), + topic: default_mqtt_topic(), + username: None, + password: None, + client_id: None, + qos: 0, + retain: false, + } + } +} + +fn default_mqtt_broker_url() -> String { + String::from("mqtt://localhost:1883") +} + +fn default_mqtt_topic() -> String { + String::from("komodo/events") +} + // QUERY #[typeshare] pub type AlerterQuery = ResourceQuery; diff --git a/client/core/ts/src/types.ts b/client/core/ts/src/types.ts index cb7f0c110f..9dfc52b45f 100644 --- a/client/core/ts/src/types.ts +++ b/client/core/ts/src/types.ts @@ -227,7 +227,9 @@ export type AlerterEndpoint = /** Send alert to Ntfy */ | { type: "Ntfy", params: NtfyAlerterEndpoint } /** Send alert to Pushover */ - | { type: "Pushover", params: PushoverAlerterEndpoint }; + | { type: "Pushover", params: PushoverAlerterEndpoint } + /** Send alert JSON to an MQTT broker */ + | { type: "Mqtt", params: MqttAlerterEndpoint }; /** Used to reference a specific resource across all resource types */ export type ResourceTarget = @@ -9150,6 +9152,24 @@ export interface NtfyAlerterEndpoint { email?: string; } +/** Configuration for an MQTT alerter. */ +export interface MqttAlerterEndpoint { + /** MQTT broker URL. Example: `mqtt://localhost:1883` */ + broker_url: string; + /** Topic to publish alerts to */ + topic: string; + /** Optional username for broker authentication */ + username?: string; + /** Optional password for broker authentication */ + password?: string; + /** Optional client identifier. If empty, core will generate one. */ + client_id?: string; + /** MQTT QoS level: 0, 1, or 2 */ + qos: number; + /** Whether the broker should retain the message */ + retain: boolean; +} + /** Pauses all containers on the target server. Response: [Update] */ export interface PauseAllContainers { /** Name or id */ diff --git a/docsite/docs/resources.md b/docsite/docs/resources.md index 24eb5a4794..2114434fff 100644 --- a/docsite/docs/resources.md +++ b/docsite/docs/resources.md @@ -70,4 +70,5 @@ All resources which depend on git repos / docker registries are able to use thes ## Alerter - Route alerts to various endpoints. +- Native endpoints include `Discord`, `Slack`, `Ntfy`, `Pushover`, `MQTT`, and `Custom`. - Can configure rules on each Alerter, such as resource whitelist, blacklist, or alert type filter. diff --git a/ui/src/resources/alerter/config/endpoint.tsx b/ui/src/resources/alerter/config/endpoint.tsx index 22f2d29b8c..303a9b4fdb 100644 --- a/ui/src/resources/alerter/config/endpoint.tsx +++ b/ui/src/resources/alerter/config/endpoint.tsx @@ -2,12 +2,16 @@ import { MonacoEditor, ConfigInput, ConfigItem } from "mogh_ui"; import { Select } from "@mantine/core"; import { Types } from "komodo_client"; -const ENDPOINT_TYPES: Types.AlerterEndpoint["type"][] = [ - "Custom", - "Discord", - "Slack", - "Ntfy", - "Pushover", +const ENDPOINT_TYPES: { + value: Types.AlerterEndpoint["type"]; + label: string; +}[] = [ + { value: "Custom", label: "Custom" }, + { value: "Discord", label: "Discord" }, + { value: "Slack", label: "Slack" }, + { value: "Ntfy", label: "Ntfy" }, + { value: "Pushover", label: "Pushover" }, + { value: "Mqtt", label: "MQTT" }, ] as const; export default function AlerterConfigEndpoint({ @@ -29,25 +33,22 @@ export default function AlerterConfigEndpoint({ value={endpoint.type} onChange={(type) => type && - set({ - type: type as Types.AlerterEndpoint["type"], - params: { - url: defaultUrl(type as Types.AlerterEndpoint["type"]), - }, - }) + set(defaultEndpoint(type as Types.AlerterEndpoint["type"])) } disabled={disabled} data={ENDPOINT_TYPES} w={{ base: "85%", lg: 400 }} /> - - set({ ...endpoint, params: { ...endpoint.params, url } }) - } - readOnly={disabled} - /> + {endpoint.type !== "Mqtt" && ( + + set({ ...endpoint, params: { ...endpoint.params, url } }) + } + readOnly={disabled} + /> + )} {endpoint.type === "Ntfy" && ( )} + {endpoint.type === "Mqtt" && ( + <> + + set({ + ...endpoint, + params: { ...endpoint.params, broker_url }, + }) + } + disabled={disabled} + /> + + set({ + ...endpoint, + params: { ...endpoint.params, topic }, + }) + } + disabled={disabled} + /> + + set({ + ...endpoint, + params: { ...endpoint.params, username: username || undefined }, + }) + } + disabled={disabled} + /> + + set({ + ...endpoint, + params: { ...endpoint.params, password: password || undefined }, + }) + } + inputProps={{ type: "password" }} + disabled={disabled} + /> + + set({ + ...endpoint, + params: { + ...endpoint.params, + client_id: client_id || undefined, + }, + }) + } + disabled={disabled} + /> + + + set({ + ...endpoint, + params: { + ...endpoint.params, + retain: retain === "true", + }, + }) + } + disabled={disabled} + w={{ base: "85%", lg: 400 }} + /> + + + )} ); } -function defaultUrl(type: Types.AlerterEndpoint["type"]) { +function defaultEndpoint( + type: Types.AlerterEndpoint["type"], +): Types.AlerterEndpoint { + return type === "Mqtt" + ? { + type, + params: { + broker_url: "mqtt://localhost:1883", + topic: "komodo/events", + username: undefined, + password: undefined, + client_id: undefined, + qos: 0, + retain: false, + }, + } + : { + type, + params: { + url: defaultUrl(type), + }, + }; +} + +function defaultUrl( + type: Exclude, +) { return type === "Custom" ? "http://localhost:7000" : type === "Slack" @@ -78,7 +225,5 @@ function defaultUrl(type: Types.AlerterEndpoint["type"]) { ? "https://discord.com/api/webhooks/XXXXXXXXXXXX/XXXX-XXXXXXXXXX" : type === "Ntfy" ? "https://ntfy.sh/komodo" - : type === "Pushover" - ? "https://api.pushover.net/1/messages.json?token=XXXXXXXXXXXXX&user=XXXXXXXXXXXXX" - : ""; + : "https://api.pushover.net/1/messages.json?token=XXXXXXXXXXXXX&user=XXXXXXXXXXXXX"; }