[pve-devel] [PATCH v4 proxmox 13/69] notify: add notification filter mechanism
Lukas Wagner
l.wagner at proxmox.com
Thu Jul 20 16:31:40 CEST 2023
This commit adds a way to filter notifications based on severity. The
filter module also has the necessary foundation work for more complex
filters, e.g. matching on properties or for creating arbitarily complex
filter structures using nested sub-filters.
Signed-off-by: Lukas Wagner <l.wagner at proxmox.com>
---
proxmox-notify/src/api/gotify.rs | 3 +
proxmox-notify/src/api/group.rs | 10 +-
proxmox-notify/src/api/sendmail.rs | 4 +
proxmox-notify/src/config.rs | 8 +
proxmox-notify/src/endpoints/gotify.rs | 12 ++
proxmox-notify/src/endpoints/sendmail.rs | 12 ++
proxmox-notify/src/filter.rs | 199 +++++++++++++++++++++++
proxmox-notify/src/group.rs | 8 +
proxmox-notify/src/lib.rs | 148 ++++++++++++++++-
9 files changed, 396 insertions(+), 8 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/group.rs b/proxmox-notify/src/api/group.rs
index cc847364..d62167ab 100644
--- a/proxmox-notify/src/api/group.rs
+++ b/proxmox-notify/src/api/group.rs
@@ -80,6 +80,7 @@ pub fn update_group(
for deleteable_property in delete {
match deleteable_property {
DeleteableGroupProperty::Comment => group.comment = None,
+ DeleteableGroupProperty::Filter => group.filter = None,
}
}
}
@@ -99,6 +100,10 @@ pub fn update_group(
group.comment = Some(comment.into());
}
+ if let Some(filter) = &updater.filter {
+ group.filter = Some(filter.into());
+ }
+
config
.config
.set_data(name, GROUP_TYPENAME, &group)
@@ -156,6 +161,7 @@ mod tests {
name: "group1".into(),
endpoint: vec!["test".to_string()],
comment: None,
+ filter: None,
},
)?;
@@ -171,6 +177,7 @@ mod tests {
name: "group1".into(),
endpoint: vec!["foo".into()],
comment: None,
+ filter: None,
},
)
.is_err());
@@ -228,7 +235,8 @@ mod tests {
"group1",
&GroupConfigUpdater {
endpoint: None,
- comment: Some("newcomment".into())
+ comment: Some("newcomment".into()),
+ filter: None
},
None,
None,
diff --git a/proxmox-notify/src/api/sendmail.rs b/proxmox-notify/src/api/sendmail.rs
index 8eafe359..3917a2e3 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 at example.com".into()),
author: Some("root".into()),
comment: Some("Comment".into()),
+ filter: None,
},
)?;
@@ -178,6 +180,7 @@ pub mod tests {
from_address: Some("root at 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 at 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 53817254..645b7bf6 100644
--- a/proxmox-notify/src/config.rs
+++ b/proxmox-notify/src/config.rs
@@ -2,6 +2,7 @@ use lazy_static::lazy_static;
use proxmox_schema::{ApiType, ObjectSchema};
use proxmox_section_config::{SectionConfig, SectionConfigData, SectionConfigPlugin};
+use crate::filter::{FilterConfig, FILTER_TYPENAME};
use crate::group::{GroupConfig, GROUP_TYPENAME};
use crate::schema::BACKEND_NAME_SCHEMA;
use crate::Error;
@@ -45,6 +46,13 @@ fn config_init() -> SectionConfig {
GROUP_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 37553d4f..349eba4c 100644
--- a/proxmox-notify/src/endpoints/gotify.rs
+++ b/proxmox-notify/src/endpoints/gotify.rs
@@ -37,6 +37,10 @@ pub(crate) const GOTIFY_TYPENAME: &str = "gotify";
optional: true,
schema: COMMENT_SCHEMA,
},
+ filter: {
+ optional: true,
+ schema: ENTITY_NAME_SCHEMA,
+ },
}
)]
#[derive(Serialize, Deserialize, Updater, Default)]
@@ -51,6 +55,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()]
@@ -77,6 +84,7 @@ pub struct GotifyEndpoint {
#[serde(rename_all = "kebab-case")]
pub enum DeleteableGotifyProperty {
Comment,
+ Filter,
}
impl Endpoint for GotifyEndpoint {
@@ -114,4 +122,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 e9458cac..d89ee979 100644
--- a/proxmox-notify/src/endpoints/sendmail.rs
+++ b/proxmox-notify/src/endpoints/sendmail.rs
@@ -22,6 +22,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, Default)]
@@ -42,6 +46,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)]
@@ -50,6 +57,7 @@ pub enum DeleteableSendmailProperty {
FromAddress,
Author,
Comment,
+ Filter,
}
/// A sendmail notification endpoint.
@@ -86,4 +94,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..5967deae
--- /dev/null
+++ b/proxmox-notify/src/filter.rs
@@ -0,0 +1,199 @@
+use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, HashSet};
+
+use proxmox_schema::api_types::COMMENT_SCHEMA;
+use proxmox_schema::{api, Updater};
+
+use crate::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,
+ }
+ }
+}
+
+#[api(
+ properties: {
+ name: {
+ schema: ENTITY_NAME_SCHEMA,
+ },
+ comment: {
+ optional: true,
+ schema: COMMENT_SCHEMA,
+ },
+ })]
+#[derive(Debug, Serialize, Deserialize, Updater, Default)]
+#[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
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub min_severity: Option<Severity>,
+
+ /// Choose between 'and' and 'or' for when multiple properties are specified
+ #[serde(skip_serializing_if = "Option::is_none")]
+ pub mode: Option<FilterModeOperator>,
+
+ /// 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,
+ Mode,
+ 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),
+ );
+
+ Ok(notification_matches != invert_match)
+ }
+
+ 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;
+
+ 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(),
+ }
+ }
+
+ #[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, ¬ification);
+ results.check_filter_match("test")
+ };
+
+ assert!(is_match(Severity::Warning)?);
+ assert!(!is_match(Severity::Notice)?);
+ assert!(is_match(Severity::Error)?);
+
+ Ok(())
+ }
+}
diff --git a/proxmox-notify/src/group.rs b/proxmox-notify/src/group.rs
index bf0b42e5..726b7120 100644
--- a/proxmox-notify/src/group.rs
+++ b/proxmox-notify/src/group.rs
@@ -18,6 +18,10 @@ pub(crate) const GROUP_TYPENAME: &str = "group";
optional: true,
schema: COMMENT_SCHEMA,
},
+ filter: {
+ optional: true,
+ schema: ENTITY_NAME_SCHEMA,
+ },
},
)]
#[derive(Debug, Serialize, Deserialize, Updater, Default)]
@@ -32,10 +36,14 @@ pub struct GroupConfig {
/// 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)]
#[serde(rename_all = "kebab-case")]
pub enum DeleteableGroupProperty {
Comment,
+ Filter,
}
diff --git a/proxmox-notify/src/lib.rs b/proxmox-notify/src/lib.rs
index bb35199f..f90dc0d9 100644
--- a/proxmox-notify/src/lib.rs
+++ b/proxmox-notify/src/lib.rs
@@ -1,6 +1,7 @@
use std::collections::HashMap;
use std::fmt::Display;
+use filter::{FilterConfig, FilterMatcher, FILTER_TYPENAME};
use group::{GroupConfig, GROUP_TYPENAME};
use proxmox_schema::api;
use proxmox_section_config::SectionConfigData;
@@ -13,6 +14,7 @@ use std::error::Error as StdError;
pub mod api;
mod config;
pub mod endpoints;
+mod filter;
pub mod group;
pub mod schema;
@@ -22,7 +24,8 @@ pub enum Error {
ConfigDeserialization(Box<dyn StdError + Send + Sync>),
NotifyFailed(String, Box<dyn StdError + Send + Sync>),
TargetDoesNotExist(String),
- TargetTestFailed(Vec<Box<dyn StdError + Send + Sync + 'static>>),
+ TargetTestFailed(Vec<Box<dyn StdError + Send + Sync>>),
+ FilterFailed(String),
}
impl Display for Error {
@@ -47,6 +50,9 @@ impl Display for Error {
Ok(())
}
+ Error::FilterFailed(message) => {
+ write!(f, "could not apply filter: {message}")
+ }
}
}
}
@@ -59,6 +65,7 @@ impl StdError for Error {
Error::NotifyFailed(_, err) => Some(&**err),
Error::TargetDoesNotExist(_) => None,
Error::TargetTestFailed(errs) => Some(&*errs[0]),
+ Error::FilterFailed(_) => None,
}
}
}
@@ -85,6 +92,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)]
@@ -143,6 +153,7 @@ impl Config {
pub struct Bus {
endpoints: HashMap<String, Box<dyn Endpoint>>,
groups: HashMap<String, GroupConfig>,
+ filters: Vec<FilterConfig>,
}
#[allow(unused_macros)]
@@ -254,7 +265,16 @@ impl Bus {
.map(|group: GroupConfig| (group.name.clone(), group))
.collect();
- Ok(Bus { endpoints, groups })
+ let filters = config
+ .config
+ .convert_to_typed_array(FILTER_TYPENAME)
+ .map_err(|err| Error::ConfigDeserialization(err.into()))?;
+
+ Ok(Bus {
+ endpoints,
+ groups,
+ filters,
+ })
}
#[cfg(test)]
@@ -267,29 +287,63 @@ impl Bus {
self.groups.insert(group.name.clone(), group);
}
+ #[cfg(test)]
+ pub fn add_filter(&mut self, filter: FilterConfig) {
+ self.filters.push(filter)
+ }
+
/// Send a notification to a given target (endpoint or group).
///
/// Any errors will not be returned but only logged.
pub fn send(&self, endpoint_or_group: &str, notification: &Notification) {
+ let mut filter_matcher = FilterMatcher::new(&self.filters, notification);
+
if let Some(group) = self.groups.get(endpoint_or_group) {
+ if !Bus::check_filter(&mut filter_matcher, group.filter.as_deref()) {
+ return;
+ }
+
for endpoint in &group.endpoint {
- self.send_via_single_endpoint(endpoint, notification);
+ self.send_via_single_endpoint(endpoint, notification, &mut filter_matcher);
}
} else {
- self.send_via_single_endpoint(endpoint_or_group, notification);
+ self.send_via_single_endpoint(endpoint_or_group, notification, &mut filter_matcher);
}
}
- fn send_via_single_endpoint(&self, endpoint: &str, notification: &Notification) {
+ fn check_filter(filter_matcher: &mut FilterMatcher, filter: Option<&str>) -> bool {
+ if let Some(filter) = filter {
+ match filter_matcher.check_filter_match(filter) {
+ // If the filter does not match, do nothing
+ Ok(r) => r,
+ Err(err) => {
+ // If there is an error, only log it and still send
+ log::error!("could not apply filter '{filter}': {err}");
+ true
+ }
+ }
+ } else {
+ true
+ }
+ }
+
+ fn send_via_single_endpoint(
+ &self,
+ endpoint: &str,
+ notification: &Notification,
+ filter_matcher: &mut FilterMatcher,
+ ) {
if let Some(endpoint) = self.endpoints.get(endpoint) {
+ if !Bus::check_filter(filter_matcher, endpoint.filter()) {
+ return;
+ }
+
if let Err(e) = endpoint.send(notification) {
// Only log on errors, do not propagate fail to the caller.
log::error!(
"could not notify via target `{name}`: {e}",
name = endpoint.name()
);
- } else {
- log::info!("notified via endpoint `{name}`", name = endpoint.name());
}
} else {
log::error!("could not notify via endpoint '{endpoint}', it does not exist");
@@ -349,6 +403,7 @@ mod tests {
// Needs to be an Rc so that we can clone MockEndpoint before
// passing it to Bus, while still retaining a handle to the Vec
messages: Rc<RefCell<Vec<Notification>>>,
+ filter: Option<String>,
}
impl Endpoint for MockEndpoint {
@@ -361,12 +416,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()
}
}
@@ -410,12 +470,14 @@ mod tests {
name: "group1".to_string(),
endpoint: vec!["mock1".into()],
comment: None,
+ filter: None,
});
bus.add_group(GroupConfig {
name: "group2".to_string(),
endpoint: vec!["mock2".into()],
comment: None,
+ filter: None,
});
bus.add_endpoint(Box::new(endpoint1.clone()));
@@ -443,4 +505,76 @@ 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_group(GroupConfig {
+ name: "channel1".to_string(),
+ endpoint: vec!["mock1".into(), "mock2".into()],
+ comment: None,
+ filter: None,
+ });
+
+ bus.add_filter(FilterConfig {
+ name: "filter1".into(),
+ min_severity: Some(Severity::Warning),
+ mode: None,
+ invert_match: None,
+ comment: None,
+ });
+
+ bus.add_filter(FilterConfig {
+ name: "filter2".into(),
+ min_severity: Some(Severity::Error),
+ mode: 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(),
+ },
+ );
+ };
+
+ 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.39.2
More information about the pve-devel
mailing list