Integrated Targets with Alerts

Alerts creation will require the ULID of the target instead of the target body

Author: parmesant

src/alerts/mod.rs

@@ -45,8 +45,6 @@ use crate::storage::ObjectStorageError;
 use crate::sync::alert_runtime;
 use crate::utils::user_auth_for_query;
 
-use self::target::Target;
-
 // these types describe the scheduled task for an alert
 pub type ScheduledTaskHandlers = (JoinHandle<()>, Receiver<()>, Sender<()>);
 
@@ -519,10 +517,10 @@ pub struct AlertRequest {
 
 impl AlertRequest {
     pub async fn into(self) -> Result<AlertConfig, AlertError> {
-        let mut targets = Vec::new();
-        for id in self.targets {
-            targets.push(TARGETS.get_target_by_id(id).await?);
-        }
+        // let mut targets = Vec::new();
+        // for id in self.targets {
+        //     targets.push(TARGETS.get_target_by_id(id).await?);
+        // }
         let config = AlertConfig {
             version: AlertVerison::from(CURRENT_ALERTS_VERSION),
             id: Ulid::new(),
@@ -532,7 +530,7 @@ impl AlertRequest {
             alert_type: self.alert_type,
             aggregates: self.aggregates,
             eval_config: self.eval_config,
-            targets,
+            targets: self.targets,
             state: AlertState::default(),
         };
         Ok(config)
@@ -551,25 +549,24 @@ pub struct AlertConfig {
     pub alert_type: AlertType,
     pub aggregates: Aggregates,
     pub eval_config: EvalConfig,
-    pub targets: Vec<Target>,
+    pub targets: Vec<Ulid>,
     // for new alerts, state should be resolved
     #[serde(default)]
     pub state: AlertState,
 }
 
 impl AlertConfig {
     pub async fn modify(&mut self, alert: AlertRequest) -> Result<(), AlertError> {
-        let mut targets = Vec::new();
-        for id in alert.targets {
-            targets.push(TARGETS.get_target_by_id(id).await?);
-        }
-
+        // let mut targets = Vec::new();
+        // for id in alert.targets {
+        //     targets.push(id);
+        // }
         self.title = alert.title;
         self.stream = alert.stream;
         self.alert_type = alert.alert_type;
         self.aggregates = alert.aggregates;
         self.eval_config = alert.eval_config;
-        self.targets = targets;
+        self.targets = alert.targets;
         self.state = AlertState::default();
         Ok(())
     }
@@ -593,7 +590,8 @@ impl AlertConfig {
         };
 
         // validate that target repeat notifs !> eval_frequency
-        for target in &self.targets {
+        for target_id in &self.targets {
+            let target = TARGETS.get_target_by_id(target_id).await?;
             match &target.timeout.times {
                 target::Retry::Infinite => {}
                 target::Retry::Finite(repeat) => {
@@ -779,7 +777,8 @@ impl AlertConfig {
     pub async fn trigger_notifications(&self, message: String) -> Result<(), AlertError> {
         let mut context = self.get_context();
         context.message = message;
-        for target in &self.targets {
+        for target_id in &self.targets {
+            let target = TARGETS.get_target_by_id(target_id).await?;
             trace!("Target (trigger_notifications)-\n{target:?}");
             target.call(context.clone());
         }

src/alerts/target.rs

@@ -35,7 +35,7 @@ use tracing::{error, trace, warn};
 use ulid::Ulid;
 use url::Url;
 
-use crate::alerts::AlertError;
+use crate::{alerts::AlertError, parseable::PARSEABLE};
 
 use super::ALERTS;
 
@@ -51,6 +51,18 @@ pub struct TargetConfigs {
 }
 
 impl TargetConfigs {
+    /// Loads alerts from disk, blocks
+    pub async fn load(&self) -> anyhow::Result<()> {
+        let mut map = self.target_configs.write().await;
+        let store = PARSEABLE.storage.get_object_store();
+
+        for alert in store.get_targets().await.unwrap_or_default() {
+            map.insert(alert.id, alert);
+        }
+
+        Ok(())
+    }
+
     pub async fn update(&self, target: Target) -> Result<(), AlertError> {
         let id = target.id;
         self.target_configs.write().await.insert(id, target);
@@ -68,20 +80,20 @@ impl TargetConfigs {
         Ok(targets)
     }
 
-    pub async fn get_target_by_id(&self, target_id: Ulid) -> Result<Target, AlertError> {
+    pub async fn get_target_by_id(&self, target_id: &Ulid) -> Result<Target, AlertError> {
         self.target_configs
             .read()
             .await
-            .get(&target_id)
+            .get(target_id)
             .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
             .cloned()
     }
 
-    pub async fn delete(&self, target_id: Ulid) -> Result<Target, AlertError> {
+    pub async fn delete(&self, target_id: &Ulid) -> Result<Target, AlertError> {
         self.target_configs
             .write()
             .await
-            .remove(&target_id)
+            .remove(target_id)
             .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
     }
 }

src/handlers/http/modal/query_server.rs

@@ -69,6 +69,7 @@ impl ParseableServer for QueryServer {
                     .service(Server::get_counts_webscope())
                     .service(Server::get_metrics_webscope())
                     .service(Server::get_alerts_webscope())
+                    .service(Server::get_targets_webscope())
                     .service(Self::get_cluster_web_scope()),
             )
             .service(

src/handlers/http/targets.rs

@@ -2,11 +2,16 @@ use actix_web::{
     web::{self, Json, Path},
     HttpRequest, Responder,
 };
+use bytes::Bytes;
 use ulid::Ulid;
 
-use crate::alerts::{
-    target::{Target, TARGETS},
-    AlertError,
+use crate::{
+    alerts::{
+        target::{Target, TARGETS},
+        AlertError,
+    },
+    parseable::PARSEABLE,
+    storage::object_storage::target_json_path,
 };
 
 // POST /targets
@@ -17,6 +22,12 @@ pub async fn post(
     // should check for duplicacy and liveness (??)
     target.validate().await;
 
+    let path = target_json_path(target.id);
+
+    let store = PARSEABLE.storage.get_object_store();
+    let target_bytes = serde_json::to_vec(&target)?;
+    store.put_object(&path, Bytes::from(target_bytes)).await?;
+
     // add to the map
     TARGETS.update(target.clone()).await?;
 
@@ -35,12 +46,12 @@ pub async fn list(_req: HttpRequest) -> Result<impl Responder, AlertError> {
 pub async fn get(_req: HttpRequest, target_id: Path<Ulid>) -> Result<impl Responder, AlertError> {
     let target_id = target_id.into_inner();
 
-    let target = TARGETS.get_target_by_id(target_id).await?;
+    let target = TARGETS.get_target_by_id(&target_id).await?;
 
     Ok(web::Json(target))
 }
 
-// POST /targets/{target_id}
+// PUT /targets/{target_id}
 pub async fn update(
     _req: HttpRequest,
     target_id: Path<Ulid>,
@@ -49,13 +60,19 @@ pub async fn update(
     let target_id = target_id.into_inner();
 
     // if target_id does not exist, error
-    TARGETS.get_target_by_id(target_id).await?;
+    TARGETS.get_target_by_id(&target_id).await?;
 
     // esnure that the supplied target id is assigned to the target config
     target.id = target_id;
     // should check for duplicacy and liveness (??)
     target.validate().await;
 
+    let path = target_json_path(target.id);
+
+    let store = PARSEABLE.storage.get_object_store();
+    let target_bytes = serde_json::to_vec(&target)?;
+    store.put_object(&path, Bytes::from(target_bytes)).await?;
+
     // add to the map
     TARGETS.update(target.clone()).await?;
 
@@ -69,7 +86,7 @@ pub async fn delete(
 ) -> Result<impl Responder, AlertError> {
     let target_id = target_id.into_inner();
 
-    let target = TARGETS.delete(target_id).await?;
+    let target = TARGETS.delete(&target_id).await?;
 
     Ok(web::Json(target))
 }

src/storage/mod.rs

@@ -59,6 +59,7 @@ pub const STREAM_ROOT_DIRECTORY: &str = ".stream";
 pub const PARSEABLE_ROOT_DIRECTORY: &str = ".parseable";
 pub const SCHEMA_FILE_NAME: &str = ".schema";
 pub const ALERTS_ROOT_DIRECTORY: &str = ".alerts";
+pub const TARGETS_ROOT_DIRECTORY: &str = ".targets";
 pub const MANIFEST_FILE: &str = "manifest.json";
 
 // max concurrent request allowed for datafusion object store

src/storage/object_storage.rs

@@ -43,6 +43,7 @@ use tracing::info;
 use tracing::{error, warn};
 use ulid::Ulid;
 
+use crate::alerts::target::Target;
 use crate::alerts::AlertConfig;
 use crate::catalog::{self, manifest::Manifest, snapshot::Snapshot};
 use crate::correlation::{CorrelationConfig, CorrelationError};
@@ -58,6 +59,7 @@ use crate::option::Mode;
 use crate::parseable::LogStream;
 use crate::parseable::PARSEABLE;
 use crate::stats::FullStats;
+use crate::storage::TARGETS_ROOT_DIRECTORY;
 
 use super::{
     retention::Retention, ObjectStorageError, ObjectStoreFormat, StorageMetadata,
@@ -422,6 +424,25 @@ pub trait ObjectStorage: Debug + Send + Sync + 'static {
         Ok(alerts)
     }
 
+    async fn get_targets(&self) -> Result<Vec<Target>, ObjectStorageError> {
+        let targets_path = RelativePathBuf::from(TARGETS_ROOT_DIRECTORY);
+        let targets = self
+            .get_objects(
+                Some(&targets_path),
+                Box::new(|file_name| file_name.ends_with(".json")),
+            )
+            .await?
+            .iter()
+            .filter_map(|bytes| {
+                serde_json::from_slice(bytes)
+                    .inspect_err(|err| warn!("Expected compatible json, error = {err}"))
+                    .ok()
+            })
+            .collect();
+
+        Ok(targets)
+    }
+
     async fn upsert_stream_metadata(
         &self,
         stream_name: &str,
@@ -974,6 +995,12 @@ pub fn alert_json_path(alert_id: Ulid) -> RelativePathBuf {
     RelativePathBuf::from_iter([ALERTS_ROOT_DIRECTORY, &format!("{alert_id}.json")])
 }
 
+/// TODO: Needs to be updated for distributed mode
+#[inline(always)]
+pub fn target_json_path(target_id: Ulid) -> RelativePathBuf {
+    RelativePathBuf::from_iter([TARGETS_ROOT_DIRECTORY, &format!("{target_id}.json")])
+}
+
 #[inline(always)]
 pub fn manifest_path(prefix: &str) -> RelativePathBuf {
     match &PARSEABLE.options.mode {