[pve-devel] [PATCH v2 proxmox 14/42] notify: add notification filter mechanism

[Lukas Wagner <l.wagner@proxmox.com>]

This commit adds a way to filter notifications based on a.) severity and
b.) arbitrary metadata property fields. For better demonstration, an example
configuration file follows:

  sendmail: mail
      recipient root@example.org
      filter only-certain-vms-and-errors

  filter: only-certain-vms-or-errors
      mode or
      min-severity error
      sub-filter only-certain-vms
      sub-filter all-but-one-ct

  filter: only-certain-vms
      mode and
      match-property object_type=vm
      sub-filter vm-ids

  filter: vm-ids
      mode or
      match-property object_id=103
      match-property object_id=104

  filter: all-but-one-ct
      mode and
      invert-match true
      match-property object_type=ct
      match-property object_id=110

In plain English, this translates to: "Send mails for all errors, as
well as all events related to VM with the IDs 103 and 104, and also
all events for any container except the one with ID 110".
The example demonstrates how sub-filters and and/or/not operators can be
used to construct filters with high granularity.

Filters are lazily evaluated, and at most once, in case multiple
endpoints/filters use the same (sub-)filter. Furthermore, there
are checks in place so that recursive sub-filter definitions are
detected.

Signed-off-by: Lukas Wagner <l.wagner@proxmox.com>
---
 proxmox-notify/src/api/gotify.rs         |   3 +
 proxmox-notify/src/api/sendmail.rs       |   4 +
 proxmox-notify/src/config.rs             |   9 +
 proxmox-notify/src/endpoints/gotify.rs   |  12 +
 proxmox-notify/src/endpoints/sendmail.rs |  12 +
 proxmox-notify/src/filter.rs             | 498 +++++++++++++++++++++++
 proxmox-notify/src/lib.rs                | 142 ++++++-
 7 files changed, 674 insertions(+), 6 deletions(-)
 create mode 100644 proxmox-notify/src/filter.rs

diff --git a/proxmox-notify/src/api/gotify.rs b/proxmox-notify/src/api/gotify.rs
index fcc1aabf..fdb9cf53 100644
--- a/proxmox-notify/src/api/gotify.rs
+++ b/proxmox-notify/src/api/gotify.rs
@@ -89,6 +89,7 @@ pub fn update_endpoint(
         for deleteable_property in delete {
             match deleteable_property {
                 DeleteableGotifyProperty::Comment => endpoint.comment = None,
+                DeleteableGotifyProperty::Filter => endpoint.filter = None,
             }
         }
     }
@@ -174,6 +175,7 @@ mod tests {
                 name: "gotify-endpoint".into(),
                 server: "localhost".into(),
                 comment: Some("comment".into()),
+                filter: None,
             },
             &GotifyPrivateConfig {
                 name: "gotify-endpoint".into(),
@@ -233,6 +235,7 @@ mod tests {
             &GotifyConfigUpdater {
                 server: Some("newhost".into()),
                 comment: Some("newcomment".into()),
+                filter: None,
             },
             &GotifyPrivateConfigUpdater {
                 token: Some("changedtoken".into()),
diff --git a/proxmox-notify/src/api/sendmail.rs b/proxmox-notify/src/api/sendmail.rs
index 458893ae..a5379cd3 100644
--- a/proxmox-notify/src/api/sendmail.rs
+++ b/proxmox-notify/src/api/sendmail.rs
@@ -75,6 +75,7 @@ pub fn update_endpoint(
                 DeleteableSendmailProperty::FromAddress => endpoint.from_address = None,
                 DeleteableSendmailProperty::Author => endpoint.author = None,
                 DeleteableSendmailProperty::Comment => endpoint.comment = None,
+                DeleteableSendmailProperty::Filter => endpoint.filter = None,
             }
         }
     }
@@ -136,6 +137,7 @@ pub mod tests {
                 from_address: Some("from@example.com".into()),
                 author: Some("root".into()),
                 comment: Some("Comment".into()),
+                filter: None,
             },
         )?;
 
@@ -178,6 +180,7 @@ pub mod tests {
                 from_address: Some("root@example.com".into()),
                 author: Some("newauthor".into()),
                 comment: Some("new comment".into()),
+                filter: None,
             },
             None,
             Some(&[0; 32]),
@@ -202,6 +205,7 @@ pub mod tests {
                 from_address: Some("root@example.com".into()),
                 author: Some("newauthor".into()),
                 comment: Some("new comment".into()),
+                filter: None,
             },
             None,
             Some(&digest),
diff --git a/proxmox-notify/src/config.rs b/proxmox-notify/src/config.rs
index a73b7849..6f96446a 100644
--- a/proxmox-notify/src/config.rs
+++ b/proxmox-notify/src/config.rs
@@ -3,6 +3,7 @@ use proxmox_schema::{ApiType, ObjectSchema};
 use proxmox_section_config::{SectionConfig, SectionConfigData, SectionConfigPlugin};
 
 use crate::channel::{ChannelConfig, CHANNEL_TYPENAME};
+use crate::filter::{FilterConfig, FILTER_TYPENAME};
 use crate::schema::BACKEND_NAME_SCHEMA;
 use crate::Error;
 
@@ -44,6 +45,14 @@ fn config_init() -> SectionConfig {
         CHANNEL_SCHEMA,
     ));
 
+    const FILTER_SCHEMA: &ObjectSchema = FilterConfig::API_SCHEMA.unwrap_object_schema();
+
+    config.register_plugin(SectionConfigPlugin::new(
+        FILTER_TYPENAME.to_string(),
+        Some(String::from("name")),
+        FILTER_SCHEMA,
+    ));
+
     config
 }
 
diff --git a/proxmox-notify/src/endpoints/gotify.rs b/proxmox-notify/src/endpoints/gotify.rs
index 74dd4868..0d306964 100644
--- a/proxmox-notify/src/endpoints/gotify.rs
+++ b/proxmox-notify/src/endpoints/gotify.rs
@@ -36,6 +36,10 @@ pub(crate) const GOTIFY_TYPENAME: &str = "gotify";
             optional: true,
             schema: COMMENT_SCHEMA,
         },
+        filter: {
+            optional: true,
+            schema: ENTITY_NAME_SCHEMA,
+        },
     }
 )]
 #[derive(Serialize, Deserialize, Updater)]
@@ -50,6 +54,9 @@ pub struct GotifyConfig {
     /// Comment
     #[serde(skip_serializing_if = "Option::is_none")]
     pub comment: Option<String>,
+    /// Filter to apply
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub filter: Option<String>,
 }
 
 #[api()]
@@ -76,6 +83,7 @@ pub struct GotifyEndpoint {
 #[serde(rename_all = "kebab-case")]
 pub enum DeleteableGotifyProperty {
     Comment,
+    Filter,
 }
 
 impl Endpoint for GotifyEndpoint {
@@ -113,4 +121,8 @@ impl Endpoint for GotifyEndpoint {
     fn name(&self) -> &str {
         &self.config.name
     }
+
+    fn filter(&self) -> Option<&str> {
+        self.config.filter.as_deref()
+    }
 }
diff --git a/proxmox-notify/src/endpoints/sendmail.rs b/proxmox-notify/src/endpoints/sendmail.rs
index f9b3df83..ee96c10a 100644
--- a/proxmox-notify/src/endpoints/sendmail.rs
+++ b/proxmox-notify/src/endpoints/sendmail.rs
@@ -21,6 +21,10 @@ pub(crate) const SENDMAIL_TYPENAME: &str = "sendmail";
             optional: true,
             schema: COMMENT_SCHEMA,
         },
+        filter: {
+            optional: true,
+            schema: ENTITY_NAME_SCHEMA,
+        },
     },
 )]
 #[derive(Debug, Serialize, Deserialize, Updater)]
@@ -41,6 +45,9 @@ pub struct SendmailConfig {
     /// Comment
     #[serde(skip_serializing_if = "Option::is_none")]
     pub comment: Option<String>,
+    /// Filter to apply
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub filter: Option<String>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -49,6 +56,7 @@ pub enum DeleteableSendmailProperty {
     FromAddress,
     Author,
     Comment,
+    Filter,
 }
 
 /// A sendmail notification endpoint.
@@ -85,4 +93,8 @@ impl Endpoint for SendmailEndpoint {
     fn name(&self) -> &str {
         &self.config.name
     }
+
+    fn filter(&self) -> Option<&str> {
+        self.config.filter.as_deref()
+    }
 }
diff --git a/proxmox-notify/src/filter.rs b/proxmox-notify/src/filter.rs
new file mode 100644
index 00000000..00614ff1
--- /dev/null
+++ b/proxmox-notify/src/filter.rs
@@ -0,0 +1,498 @@
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::collections::{HashMap, HashSet};
+
+use proxmox_schema::{api, property_string::PropertyIterator, Updater};
+
+use crate::schema::{COMMENT_SCHEMA, ENTITY_NAME_SCHEMA};
+use crate::{Error, Notification, Severity};
+
+pub const FILTER_TYPENAME: &str = "filter";
+
+#[api]
+#[derive(Debug, Serialize, Deserialize, Default, Clone, Copy)]
+#[serde(rename_all = "kebab-case")]
+pub enum FilterModeOperator {
+    /// All filter properties have to match (AND)
+    #[default]
+    And,
+    /// At least one filter property has to match (OR)
+    Or,
+}
+
+impl FilterModeOperator {
+    /// Apply the mode operator to two bools, lhs and rhs
+    fn apply(&self, lhs: bool, rhs: bool) -> bool {
+        match self {
+            FilterModeOperator::And => lhs && rhs,
+            FilterModeOperator::Or => lhs || rhs,
+        }
+    }
+
+    fn neutral_element(&self) -> bool {
+        match self {
+            FilterModeOperator::And => true,
+            FilterModeOperator::Or => false,
+        }
+    }
+
+    /// Check if we need to evaluate any other properties, or if we can return early, since
+    /// false AND (...) = false
+    /// true OR (...) = true
+    fn short_circuit_return_possible(&self, value: bool) -> bool {
+        matches!(
+            (self, value),
+            (FilterModeOperator::And, false) | (FilterModeOperator::Or, true)
+        )
+    }
+}
+
+#[api(
+    properties: {
+        name: {
+            schema: ENTITY_NAME_SCHEMA,
+        },
+        "sub-filter": {
+            optional: true,
+            type: Array,
+            items: {
+                schema: ENTITY_NAME_SCHEMA,
+            },
+        },
+        "match-property": {
+            optional: true,
+            type: Array,
+            items: {
+                description: "Notification properties to match",
+                type: String,
+            },
+        },
+        comment: {
+            optional: true,
+            schema: COMMENT_SCHEMA,
+        },
+    })]
+#[derive(Debug, Serialize, Deserialize, Updater)]
+#[serde(rename_all = "kebab-case")]
+/// Config for Sendmail notification endpoints
+pub struct FilterConfig {
+    /// Name of the filter
+    #[updater(skip)]
+    pub name: String,
+
+    /// Minimum severity to match
+    pub min_severity: Option<Severity>,
+
+    /// Sub-filter, allows arbitrary nesting (no recursion allowed)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub sub_filter: Option<Vec<String>>,
+
+    /// Choose between 'and' and 'or' for when multiple properties are specified
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub mode: Option<FilterModeOperator>,
+
+    /// Notification properties to match.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub match_property: Option<Vec<String>>,
+
+    /// Invert match of the whole filter
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub invert_match: Option<bool>,
+
+    /// Comment
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub comment: Option<String>,
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum DeleteableFilterProperty {
+    MinSeverity,
+    SubFilter,
+    Mode,
+    MatchProperty,
+    InvertMatch,
+    Comment,
+}
+
+/// A caching, lazily-evaluating notification filter. Parameterized with the notification itself,
+/// since there are usually multiple filters to check for a single notification that is to be sent.
+pub(crate) struct FilterMatcher<'a> {
+    filters: HashMap<&'a str, &'a FilterConfig>,
+    cached_results: HashMap<&'a str, bool>,
+    notification: &'a Notification,
+}
+
+impl<'a> FilterMatcher<'a> {
+    pub(crate) fn new(filters: &'a [FilterConfig], notification: &'a Notification) -> Self {
+        let filters = filters.iter().map(|f| (f.name.as_str(), f)).collect();
+
+        Self {
+            filters,
+            cached_results: Default::default(),
+            notification,
+        }
+    }
+
+    /// Check if the notification that was used to instantiate Self matches a given filter
+    pub(crate) fn check_filter_match(&mut self, filter_name: &str) -> Result<bool, Error> {
+        let mut visited = HashSet::new();
+
+        self.do_check_filter(filter_name, &mut visited)
+    }
+
+    fn do_check_filter(
+        &mut self,
+        filter_name: &str,
+        visited: &mut HashSet<String>,
+    ) -> Result<bool, Error> {
+        if visited.contains(filter_name) {
+            return Err(Error::FilterFailed(format!(
+                "recursive filter definition: {filter_name}"
+            )));
+        }
+
+        if let Some(is_match) = self.cached_results.get(filter_name) {
+            return Ok(*is_match);
+        }
+
+        visited.insert(filter_name.into());
+
+        let filter_config =
+            self.filters.get(filter_name).copied().ok_or_else(|| {
+                Error::FilterFailed(format!("filter '{filter_name}' does not exist"))
+            })?;
+
+        let invert_match = filter_config.invert_match.unwrap_or_default();
+
+        let mode_operator = filter_config.mode.unwrap_or_default();
+
+        let mut notification_matches = mode_operator.neutral_element();
+
+        notification_matches = mode_operator.apply(
+            notification_matches,
+            self.check_severity_match(filter_config, mode_operator),
+        );
+
+        if mode_operator.short_circuit_return_possible(notification_matches) {
+            return Ok(notification_matches != invert_match);
+        }
+
+        notification_matches = mode_operator.apply(
+            notification_matches,
+            self.check_property_match(filter_config, mode_operator)?,
+        );
+
+        if mode_operator.short_circuit_return_possible(notification_matches) {
+            return Ok(notification_matches != invert_match);
+        }
+
+        if let Some(sub_filters) = &filter_config.sub_filter {
+            for filter in sub_filters {
+                let is_match = self.do_check_filter(filter, visited)?;
+
+                self.cached_results.insert(filter.as_str(), is_match);
+
+                notification_matches = mode_operator.apply(notification_matches, is_match);
+
+                if mode_operator.short_circuit_return_possible(notification_matches) {
+                    return Ok(notification_matches != invert_match);
+                }
+            }
+        }
+
+        Ok(notification_matches != invert_match)
+    }
+
+    fn check_property_match(
+        &self,
+        filter_config: &FilterConfig,
+        mode_operator: FilterModeOperator,
+    ) -> Result<bool, Error> {
+        let mut notification_matches = mode_operator.neutral_element();
+
+        if let Some(match_property_operators) = filter_config.match_property.as_ref() {
+            for op in match_property_operators {
+                for prop in PropertyIterator::new(op) {
+                    let prop = prop.map_err(|err| {
+                        Error::FilterFailed(format!(
+                            "invalid match-property statement '{op}': {err}"
+                        ))
+                    })?;
+
+                    if let (Some(key), expected_value) = prop {
+                        let value = self
+                            .notification
+                            .properties
+                            .as_ref()
+                            .and_then(|v| v.as_object())
+                            .and_then(|m| m.get(key))
+                            .unwrap_or(&Value::Null);
+
+                        let actual_value = match value {
+                            Value::String(s) => Ok(s.clone()),
+                            Value::Array(_) => Err(Error::FilterFailed(
+                                "match-property cannot match arrays".into(),
+                            )),
+                            Value::Object(_) => Err(Error::FilterFailed(
+                                "match-property cannot match objects".into(),
+                            )),
+                            v => Ok(v.to_string()),
+                        }?;
+
+                        notification_matches = mode_operator.apply(
+                            notification_matches,
+                            actual_value.as_str() == expected_value,
+                        );
+
+                        if mode_operator.short_circuit_return_possible(notification_matches) {
+                            return Ok(notification_matches);
+                        }
+                    }
+                }
+            }
+        }
+
+        Ok(notification_matches)
+    }
+
+    fn check_severity_match(
+        &self,
+        filter_config: &FilterConfig,
+        mode_operator: FilterModeOperator,
+    ) -> bool {
+        if let Some(min_severity) = filter_config.min_severity {
+            self.notification.severity >= min_severity
+        } else {
+            mode_operator.neutral_element()
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config;
+    use serde_json::json;
+
+    #[test]
+    fn test_filter_config_parses_correctly() -> Result<(), Error> {
+        let (c, _) = config::config(
+            r"
+filter: foo
+    min-severity info
+    match-property object_type=vm
+    match-property object_id=103
+    invert-match true
+    mode and
+
+filter: bar
+    min-severity warning
+    match-property object_type=ct,object_id=104
+    sub-filter foo
+    mode or
+",
+        )?;
+
+        let filters: Vec<FilterConfig> = c.convert_to_typed_array("filter").unwrap();
+
+        assert_eq!(filters.len(), 2);
+
+        Ok(())
+    }
+
+    fn parse_filters(config: &str) -> Result<Vec<FilterConfig>, Error> {
+        let (config, _) = config::config(config)?;
+        Ok(config.convert_to_typed_array("filter").unwrap())
+    }
+
+    fn empty_notification_with_severity(severity: Severity) -> Notification {
+        Notification {
+            title: String::new(),
+            body: String::new(),
+            severity,
+            properties: Default::default(),
+        }
+    }
+
+    fn empty_notification_with_metadata(metadata: Value) -> Notification {
+        Notification {
+            title: String::new(),
+            body: String::new(),
+            severity: Severity::Error,
+            properties: Some(metadata),
+        }
+    }
+
+    #[test]
+    fn test_trivial_severity_filters() -> Result<(), Error> {
+        let config = "
+filter: test
+    min-severity warning
+";
+
+        let filters = parse_filters(config)?;
+
+        let is_match = |severity| {
+            let notification = empty_notification_with_severity(severity);
+            let mut results = FilterMatcher::new(&filters, &notification);
+            results.check_filter_match("test")
+        };
+
+        assert!(is_match(Severity::Warning)?);
+        assert!(!is_match(Severity::Notice)?);
+        assert!(is_match(Severity::Error)?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_recursive_filter_loop() -> Result<(), Error> {
+        let config = "
+filter: direct-a
+    sub-filter direct-b
+
+filter: direct-b
+    sub-filter direct-a
+
+filter: indirect-c
+    sub-filter indirect-d
+
+filter: indirect-d
+    sub-filter indirect-e
+
+filter: indirect-e
+    sub-filter indirect-c
+";
+
+        let filters = parse_filters(config)?;
+
+        let notification = empty_notification_with_severity(Severity::Info);
+        let mut results = FilterMatcher::new(&filters, &notification);
+        assert!(results.check_filter_match("direct-a").is_err());
+        assert!(results.check_filter_match("indirect-c").is_err());
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_property_matches() -> Result<(), Error> {
+        let config = "
+filter: test
+    match-property object_type=vm
+
+filter: multiple-and
+    mode and
+    match-property a=foo,b=bar
+    match-property c=lorem,d=ipsum
+
+filter: multiple-or
+    mode or
+    match-property a=foo,b=bar
+    match-property c=lorem,d=ipsum
+    ";
+        let filters = parse_filters(config)?;
+
+        let is_match = |filter, metadata| -> Result<bool, Error> {
+            let notification = empty_notification_with_metadata(metadata);
+            let mut results = FilterMatcher::new(&filters, &notification);
+            results.check_filter_match(filter)
+        };
+
+        assert!(is_match(
+            "test",
+            json!({
+                "object_type": "vm"
+            })
+        )?);
+        assert!(!is_match("test", json!({"object_type": "ct"}))?);
+        assert!(is_match(
+            "multiple-and",
+            json!({"a": "foo", "b": "bar", "c": "lorem", "d": "ipsum"}),
+        )?);
+        assert!(!is_match(
+            "multiple-and",
+            json!({
+                "a": "invalid",
+                "b": "bar",
+                "c": "lorem",
+                "d": "ipsum"
+            }),
+        )?);
+        assert!(!is_match("multiple-and", json!({"a": "foo", "b": "bar"}))?);
+        assert!(is_match("multiple-or", json!({"a": "foo"}))?);
+        assert!(is_match("multiple-or", json!({"a": "foo"}))?);
+        assert!(is_match("multiple-or", json!({"b": "bar"}))?);
+        assert!(is_match("multiple-or", json!({"d": "ipsum"}))?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_invert_match() -> Result<(), Error> {
+        let config = "
+filter: test
+    match-property object_type=vm
+    invert-match true
+    ";
+        let filters = parse_filters(config)?;
+
+        let notification = empty_notification_with_metadata(json!({"object_type": "vm"}));
+        let mut results = FilterMatcher::new(&filters, &notification);
+        assert!(!results.check_filter_match("test")?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_invert_match_early_return() -> Result<(), Error> {
+        let config = "
+filter: sub1
+    match-property object_type=vm
+
+filter: sub2
+    match-property object_type=vm
+
+filter: test
+    mode or
+    sub-filter sub1
+    sub-filter sub2
+    invert-match true
+    ";
+        let filters = parse_filters(config)?;
+
+        let notification = empty_notification_with_metadata(json!({"object_type": "vm"}));
+        let mut results = FilterMatcher::new(&filters, &notification);
+        assert!(!results.check_filter_match("test")?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_sub_filter_matches() -> Result<(), Error> {
+        let config = "
+filter: test
+    match-property object_type=vm
+    sub-filter vm-ids
+
+filter: vm-ids
+    mode or
+    match-property object_id=100
+    match-property object_id=101
+    ";
+        let filters = parse_filters(config)?;
+
+        let is_match = |metadata| -> Result<bool, Error> {
+            let notification = empty_notification_with_metadata(metadata);
+            let mut results = FilterMatcher::new(&filters, &notification);
+            results.check_filter_match("test")
+        };
+
+        assert!(is_match(json!({"object_type": "vm", "object_id": "100"}))?);
+        assert!(is_match(json!({"object_type": "vm", "object_id": "101"}))?);
+        assert!(!is_match(json!({"object_type": "ct", "object_id": "101"}))?);
+        assert!(!is_match(json!({"object_type": "vm", "object_id": "111"}))?);
+
+        Ok(())
+    }
+}
diff --git a/proxmox-notify/src/lib.rs b/proxmox-notify/src/lib.rs
index 147dbc3c..33d3dbc7 100644
--- a/proxmox-notify/src/lib.rs
+++ b/proxmox-notify/src/lib.rs
@@ -1,6 +1,7 @@
 use std::fmt::Display;
 
 use channel::{ChannelConfig, CHANNEL_TYPENAME};
+use filter::{FilterConfig, FilterMatcher, FILTER_TYPENAME};
 use proxmox_schema::api;
 use proxmox_section_config::SectionConfigData;
 use serde::{Deserialize, Serialize};
@@ -13,6 +14,7 @@ pub mod api;
 pub mod channel;
 mod config;
 pub mod endpoints;
+mod filter;
 pub mod schema;
 
 #[derive(Debug)]
@@ -22,6 +24,7 @@ pub enum Error {
     NotifyFailed(String, Box<dyn StdError + Send + Sync + 'static>),
     EndpointDoesNotExist(String),
     ChannelDoesNotExist(String),
+    FilterFailed(String),
 }
 
 impl Display for Error {
@@ -42,6 +45,9 @@ impl Display for Error {
             Error::ChannelDoesNotExist(channel) => {
                 write!(f, "channel '{channel}' does not exist")
             }
+            Error::FilterFailed(message) => {
+                write!(f, "could not apply filter: {message}")
+            }
         }
     }
 }
@@ -54,6 +60,7 @@ impl StdError for Error {
             Error::NotifyFailed(_, err) => Some(&**err),
             Error::EndpointDoesNotExist(_) => None,
             Error::ChannelDoesNotExist(_) => None,
+            Error::FilterFailed(_) => None,
         }
     }
 }
@@ -80,6 +87,9 @@ pub trait Endpoint {
 
     /// The name/identifier for this endpoint
     fn name(&self) -> &str;
+
+    /// The name of the filter to use
+    fn filter(&self) -> Option<&str>;
 }
 
 #[derive(Debug, Clone)]
@@ -150,6 +160,7 @@ impl Config {
 pub struct Bus {
     endpoints: Vec<Box<dyn Endpoint>>,
     channels: Vec<ChannelConfig>,
+    filters: Vec<FilterConfig>,
 }
 
 #[allow(unused_macros)]
@@ -249,9 +260,15 @@ impl Bus {
             .convert_to_typed_array(CHANNEL_TYPENAME)
             .map_err(|err| Error::ConfigDeserialization(err.into()))?;
 
+        let filters = config
+            .config
+            .convert_to_typed_array(FILTER_TYPENAME)
+            .map_err(|err| Error::ConfigDeserialization(err.into()))?;
+
         Ok(Bus {
             endpoints,
             channels,
+            filters,
         })
     }
 
@@ -265,6 +282,11 @@ impl Bus {
         self.channels.push(channel);
     }
 
+    #[cfg(test)]
+    pub fn add_filter(&mut self, filter: FilterConfig) {
+        self.filters.push(filter)
+    }
+
     pub fn send(&self, channel: &str, notification: &Notification) -> Result<(), Error> {
         log::debug!(
             "sending notification with title `{title}`",
@@ -279,6 +301,8 @@ impl Bus {
             .find(|c| c.name == channel)
             .ok_or(Error::ChannelDoesNotExist(channel.into()))?;
 
+        let mut notification_filter = FilterMatcher::new(&self.filters, notification);
+
         for endpoint in &self.endpoints {
             if !channel.should_notify_via_endpoint(endpoint.name()) {
                 log::debug!(
@@ -289,13 +313,37 @@ impl Bus {
                 continue;
             }
 
-            if let Err(e) = endpoint.send(notification) {
-                log::error!(
-                    "could not notfiy via endpoint `{name}`: {e}",
-                    name = endpoint.name()
-                );
+            let should_notify = if let Some(filter) = endpoint.filter() {
+                notification_filter
+                    .check_filter_match(filter)
+                    .unwrap_or_else(|e| {
+                        log::error!(
+                            "could not apply filter `{filter}` for endpoint `{name}: {e}`",
+                            name = endpoint.name()
+                        );
+                        // If the filter is somehow erroneous, we send a notification by default,
+                        // so no events are missed
+                        true
+                    })
+            } else {
+                true
+            };
+
+            if should_notify {
+                if let Err(e) = endpoint.send(notification) {
+                    log::error!(
+                        "could not notify via endpoint `{name}`: {e}",
+                        name = endpoint.name()
+                    );
+                } else {
+                    log::info!("notified via endpoint `{name}`", name = endpoint.name());
+                }
             } else {
-                log::info!("notified via endpoint `{name}`", name = endpoint.name());
+                log::debug!(
+                    "skipped endpoint `{name}`, filter `{filter}` did not match",
+                    name = endpoint.name(),
+                    filter = endpoint.filter().unwrap_or_default()
+                );
             }
         }
 
@@ -330,6 +378,7 @@ mod tests {
     struct MockEndpoint {
         name: &'static str,
         messages: Rc<RefCell<Vec<Notification>>>,
+        filter: Option<String>,
     }
 
     impl Endpoint for MockEndpoint {
@@ -342,12 +391,17 @@ mod tests {
         fn name(&self) -> &str {
             self.name
         }
+
+        fn filter(&self) -> Option<&str> {
+            self.filter.as_deref()
+        }
     }
 
     impl MockEndpoint {
         fn new(name: &'static str, filter: Option<String>) -> Self {
             Self {
                 name,
+                filter,
                 ..Default::default()
             }
         }
@@ -430,4 +484,80 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_severity_ordering() {
+        // Not intended to be exhaustive, just a quick
+        // sanity check ;)
+
+        assert!(Severity::Info < Severity::Notice);
+        assert!(Severity::Info < Severity::Warning);
+        assert!(Severity::Info < Severity::Error);
+        assert!(Severity::Error > Severity::Warning);
+        assert!(Severity::Warning > Severity::Notice);
+    }
+
+    #[test]
+    fn test_multiple_endpoints_with_different_filters() -> Result<(), Error> {
+        let endpoint1 = MockEndpoint::new("mock1", Some("filter1".into()));
+        let endpoint2 = MockEndpoint::new("mock2", Some("filter2".into()));
+
+        let mut bus = Bus::default();
+
+        bus.add_endpoint(Box::new(endpoint1.clone()));
+        bus.add_endpoint(Box::new(endpoint2.clone()));
+
+        bus.add_channel(ChannelConfig {
+            name: "channel1".to_string(),
+            endpoint: Some(vec!["mock1".into(), "mock2".into()]),
+            comment: None,
+        });
+
+        bus.add_filter(FilterConfig {
+            name: "filter1".into(),
+            min_severity: Some(Severity::Warning),
+            sub_filter: None,
+            mode: None,
+            match_property: None,
+            invert_match: None,
+            comment: None,
+        });
+
+        bus.add_filter(FilterConfig {
+            name: "filter2".into(),
+            min_severity: Some(Severity::Error),
+            sub_filter: None,
+            mode: None,
+            match_property: None,
+            invert_match: None,
+            comment: None,
+        });
+
+        let send_with_severity = |severity| {
+            bus.send(
+                "channel1",
+                &Notification {
+                    title: "Title".into(),
+                    body: "Body".into(),
+                    severity,
+                    properties: Default::default(),
+                },
+            )
+            .unwrap();
+        };
+
+        send_with_severity(Severity::Info);
+        assert_eq!(endpoint1.messages().len(), 0);
+        assert_eq!(endpoint2.messages().len(), 0);
+
+        send_with_severity(Severity::Warning);
+        assert_eq!(endpoint1.messages().len(), 1);
+        assert_eq!(endpoint2.messages().len(), 0);
+
+        send_with_severity(Severity::Error);
+        assert_eq!(endpoint1.messages().len(), 2);
+        assert_eq!(endpoint2.messages().len(), 1);
+
+        Ok(())
+    }
 }
-- 
2.30.2