Bring targets out of alerts

Adds CRUD operations for targets

TODO-
- Storage of targets depending upon RBAC discussion
- Modify alerts to use targets using ID (AlertConfig update)

Author: parmesant

src/alerts/mod.rs

@@ -37,6 +37,7 @@ use ulid::Ulid;
 pub mod alerts_utils;
 pub mod target;
 
+use crate::alerts::target::TARGETS;
 use crate::parseable::{StreamNotFound, PARSEABLE};
 use crate::rbac::map::SessionKey;
 use crate::storage;
@@ -513,23 +514,28 @@ pub struct AlertRequest {
     pub alert_type: AlertType,
     pub aggregates: Aggregates,
     pub eval_config: EvalConfig,
-    pub targets: Vec<Target>,
+    pub targets: Vec<Ulid>,
 }
 
-impl From<AlertRequest> for AlertConfig {
-    fn from(val: AlertRequest) -> AlertConfig {
-        AlertConfig {
+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 config = AlertConfig {
             version: AlertVerison::from(CURRENT_ALERTS_VERSION),
             id: Ulid::new(),
-            severity: val.severity,
-            title: val.title,
-            stream: val.stream,
-            alert_type: val.alert_type,
-            aggregates: val.aggregates,
-            eval_config: val.eval_config,
-            targets: val.targets,
+            severity: self.severity,
+            title: self.title,
+            stream: self.stream,
+            alert_type: self.alert_type,
+            aggregates: self.aggregates,
+            eval_config: self.eval_config,
+            targets,
             state: AlertState::default(),
-        }
+        };
+        Ok(config)
     }
 }
 
@@ -552,14 +558,20 @@ pub struct AlertConfig {
 }
 
 impl AlertConfig {
-    pub fn modify(&mut self, alert: AlertRequest) {
+    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?);
+        }
+
         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 = alert.targets;
+        self.targets = targets;
         self.state = AlertState::default();
+        Ok(())
     }
 
     pub fn get_base_query(&self) -> String {
@@ -801,6 +813,8 @@ pub enum AlertError {
     InvalidAlertModifyRequest,
     #[error("{0}")]
     FromStrError(#[from] FromStrError),
+    #[error("Invalid Target ID- {0}")]
+    InvalidTargetID(String),
 }
 
 impl actix_web::ResponseError for AlertError {
@@ -818,6 +832,7 @@ impl actix_web::ResponseError for AlertError {
             Self::Anyhow(_) => StatusCode::INTERNAL_SERVER_ERROR,
             Self::InvalidAlertModifyRequest => StatusCode::BAD_REQUEST,
             Self::FromStrError(_) => StatusCode::BAD_REQUEST,
+            Self::InvalidTargetID(_) => StatusCode::BAD_REQUEST,
         }
     }
 

src/alerts/target.rs

@@ -27,14 +27,65 @@ use base64::Engine;
 use chrono::Utc;
 use http::{header::AUTHORIZATION, HeaderMap, HeaderValue};
 use humantime_serde::re::humantime;
+use itertools::Itertools;
+use once_cell::sync::Lazy;
 use reqwest::ClientBuilder;
+use tokio::sync::RwLock;
 use tracing::{error, trace, warn};
+use ulid::Ulid;
 use url::Url;
 
+use crate::alerts::AlertError;
+
 use super::ALERTS;
 
 use super::{AlertState, CallableTarget, Context};
 
+pub static TARGETS: Lazy<TargetConfigs> = Lazy::new(|| TargetConfigs {
+    target_configs: RwLock::new(HashMap::new()),
+});
+
+#[derive(Debug)]
+pub struct TargetConfigs {
+    pub target_configs: RwLock<HashMap<Ulid, Target>>,
+}
+
+impl TargetConfigs {
+    pub async fn update(&self, target: Target) -> Result<(), AlertError> {
+        let id = target.id;
+        self.target_configs.write().await.insert(id, target);
+        Ok(())
+    }
+
+    pub async fn list(&self) -> Result<Vec<Target>, AlertError> {
+        let targets = self
+            .target_configs
+            .read()
+            .await
+            .iter()
+            .map(|(_, v)| v.clone())
+            .collect_vec();
+        Ok(targets)
+    }
+
+    pub async fn get_target_by_id(&self, target_id: Ulid) -> Result<Target, AlertError> {
+        self.target_configs
+            .read()
+            .await
+            .get(&target_id)
+            .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
+            .cloned()
+    }
+
+    pub async fn delete(&self, target_id: Ulid) -> Result<Target, AlertError> {
+        self.target_configs
+            .write()
+            .await
+            .remove(&target_id)
+            .ok_or(AlertError::InvalidTargetID(target_id.to_string()))
+    }
+}
+
 #[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 #[serde(untagged)]
@@ -57,9 +108,16 @@ pub struct Target {
     pub target: TargetType,
     #[serde(default, rename = "repeat")]
     pub timeout: Timeout,
+    #[serde(default = "Ulid::new")]
+    pub id: Ulid,
 }
 
 impl Target {
+    pub async fn validate(&self) {
+        // just check for liveness
+        // what if the target is not live yet but is added by the user?
+    }
+
     pub fn call(&self, context: Context) {
         trace!("target.call context- {context:?}");
         let timeout = &self.timeout;
@@ -193,6 +251,8 @@ pub struct TargetVerifier {
     pub target: TargetType,
     #[serde(default)]
     pub repeat: Option<RepeatVerifier>,
+    #[serde(default = "Ulid::new")]
+    pub id: Ulid,
 }
 
 impl TryFrom<TargetVerifier> for Target {
@@ -225,6 +285,7 @@ impl TryFrom<TargetVerifier> for Target {
         Ok(Target {
             target: value.target,
             timeout,
+            id: value.id,
         })
     }
 }

src/handlers/http/alerts.rs

@@ -48,7 +48,7 @@ pub async fn post(
     req: HttpRequest,
     Json(alert): Json<AlertRequest>,
 ) -> Result<impl Responder, AlertError> {
-    let alert: AlertConfig = alert.into();
+    let alert: AlertConfig = alert.into().await?;
     alert.validate().await?;
 
     // validate the incoming alert query

src/handlers/http/mod.rs

@@ -47,6 +47,7 @@ pub mod prism_logstream;
 pub mod query;
 pub mod rbac;
 pub mod role;
+pub mod targets;
 pub mod users;
 pub const MAX_EVENT_PAYLOAD_SIZE: usize = 10485760;
 pub const API_BASE_PATH: &str = "api";

src/handlers/http/modal/server.rs

@@ -26,6 +26,7 @@ use crate::handlers::http::base_path;
 use crate::handlers::http::health_check;
 use crate::handlers::http::prism_base_path;
 use crate::handlers::http::query;
+use crate::handlers::http::targets;
 use crate::handlers::http::users::dashboards;
 use crate::handlers::http::users::filters;
 use crate::hottier::HotTierManager;
@@ -87,6 +88,7 @@ impl ParseableServer for Server {
                     .service(Self::get_roles_webscope())
                     .service(Self::get_counts_webscope())
                     .service(Self::get_alerts_webscope())
+                    .service(Self::get_targets_webscope())
                     .service(Self::get_metrics_webscope()),
             )
             .service(
@@ -253,6 +255,25 @@ impl Server {
             )
     }
 
+    pub fn get_targets_webscope() -> Scope {
+        web::scope("/targets")
+            .service(
+                web::resource("")
+                    .route(web::get().to(targets::list).authorize(Action::GetAlert))
+                    .route(web::post().to(targets::post).authorize(Action::PutAlert)),
+            )
+            .service(
+                web::resource("/{target_id}")
+                    .route(web::get().to(targets::get).authorize(Action::GetAlert))
+                    .route(web::put().to(targets::update).authorize(Action::PutAlert))
+                    .route(
+                        web::delete()
+                            .to(targets::delete)
+                            .authorize(Action::DeleteAlert),
+                    ),
+            )
+    }
+
     // get the dashboards web scope
     pub fn get_dashboards_webscope() -> Scope {
         web::scope("/dashboards")

src/handlers/http/targets.rs

@@ -0,0 +1,75 @@
+use actix_web::{
+    web::{self, Json, Path},
+    HttpRequest, Responder,
+};
+use ulid::Ulid;
+
+use crate::alerts::{
+    target::{Target, TARGETS},
+    AlertError,
+};
+
+// POST /targets
+pub async fn post(
+    _req: HttpRequest,
+    Json(target): Json<Target>,
+) -> Result<impl Responder, AlertError> {
+    // should check for duplicacy and liveness (??)
+    target.validate().await;
+
+    // add to the map
+    TARGETS.update(target.clone()).await?;
+
+    Ok(web::Json(target))
+}
+
+// GET /targets
+pub async fn list(_req: HttpRequest) -> Result<impl Responder, AlertError> {
+    // add to the map
+    let list = TARGETS.list().await?;
+
+    Ok(web::Json(list))
+}
+
+// GET /targets/{target_id}
+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?;
+
+    Ok(web::Json(target))
+}
+
+// POST /targets/{target_id}
+pub async fn update(
+    _req: HttpRequest,
+    target_id: Path<Ulid>,
+    Json(mut target): Json<Target>,
+) -> Result<impl Responder, AlertError> {
+    let target_id = target_id.into_inner();
+
+    // if target_id does not exist, error
+    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;
+
+    // add to the map
+    TARGETS.update(target.clone()).await?;
+
+    Ok(web::Json(target))
+}
+
+// DELETE /targets/{target_id}
+pub async fn delete(
+    _req: HttpRequest,
+    target_id: Path<Ulid>,
+) -> Result<impl Responder, AlertError> {
+    let target_id = target_id.into_inner();
+
+    let target = TARGETS.delete(target_id).await?;
+
+    Ok(web::Json(target))
+}