Skip to content
81 changes: 81 additions & 0 deletions src/alerts/alert_enums.rs
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,26 @@ impl Display for AlertType {
}
}

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, Copy, PartialEq, Eq, Default)]
#[serde(rename_all = "camelCase")]
pub enum AlertQueryType {
#[default]
Builder,
#[serde(alias = "sql")]
Code,
Promql,
}

impl Display for AlertQueryType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
AlertQueryType::Builder => write!(f, "builder"),
AlertQueryType::Code => write!(f, "code"),
AlertQueryType::Promql => write!(f, "promql"),
}
}
}

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub enum AlertOperator {
Expand Down Expand Up @@ -211,6 +231,67 @@ impl Display for WhereConfigOperator {
}
}

#[cfg(test)]
mod tests {
use super::AlertQueryType;

#[test]
fn alert_query_type_deserializes_supported_modes() {
assert_eq!(
serde_json::from_str::<AlertQueryType>("\"builder\"").unwrap(),
AlertQueryType::Builder
);
assert_eq!(
serde_json::from_str::<AlertQueryType>("\"code\"").unwrap(),
AlertQueryType::Code
);
assert_eq!(
serde_json::from_str::<AlertQueryType>("\"promql\"").unwrap(),
AlertQueryType::Promql
);
}

#[test]
fn alert_query_type_accepts_legacy_sql_as_code() {
assert_eq!(
serde_json::from_str::<AlertQueryType>("\"sql\"").unwrap(),
AlertQueryType::Code
);
}

#[test]
fn alert_query_type_rejects_unknown_mode() {
assert!(serde_json::from_str::<AlertQueryType>("\"rawSql\"").is_err());
}

#[test]
fn alert_request_deserializes_promql_query_type() {
let request: crate::alerts::alert_structs::AlertRequest =
serde_json::from_value(serde_json::json!({
"severity": "high",
"title": "Test alert",
"alertType": "threshold",
"queryType": "promql",
"query": "sum({\"k8s.pod.cpu.usage\"}) by (\"k8s.namespace.name\")",
"thresholdConfig": {"operator": ">", "value": -1.0},
"evalConfig": {
"rollingWindow": {
"evalStart": "10 minutes",
"evalEnd": "now",
"evalFrequency": 10
}
},
"targets": [],
"notificationConfig": {"interval": 1},
"datasets": ["azure-prod-cluster-metrics"]
}))
.unwrap();

assert_eq!(request.query_type, AlertQueryType::Promql);
assert_eq!(request.datasets, vec!["azure-prod-cluster-metrics"]);
}
}

#[derive(Debug, serde::Serialize, serde::Deserialize, Clone)]
#[serde(rename_all = "camelCase")]
pub enum AggregateFunction {
Expand Down
35 changes: 31 additions & 4 deletions src/alerts/alert_structs.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,14 @@ use crate::{
alerts::{
AlertError, CURRENT_ALERTS_VERSION,
alert_enums::{
AlertOperator, AlertState, AlertTask, AlertType, AlertVersion, EvalConfig,
LogicalOperator, NotificationState, Severity, WhereConfigOperator,
AlertOperator, AlertQueryType, AlertState, AlertTask, AlertType, AlertVersion,
EvalConfig, LogicalOperator, NotificationState, Severity, WhereConfigOperator,
},
alert_traits::AlertTrait,
target::{NotificationConfig, TARGETS},
},
metastore::metastore_traits::MetastoreObject,
parseable::PARSEABLE,
query::resolve_stream_names,
storage::object_storage::{alert_json_path, alert_state_json_path, mttr_json_path},
};
Expand All @@ -45,6 +46,10 @@ const RESERVED_FIELDS: &[&str] = &[
"severity",
"title",
"query",
"queryType",
"query_type",
"creationType",
"creation_type",
"datasets",
"alertType",
"alert_type",
Expand Down Expand Up @@ -282,6 +287,10 @@ pub struct AlertRequest {
pub severity: Severity,
pub title: String,
pub query: String,
#[serde(default)]
pub query_type: AlertQueryType,
#[serde(default)]
pub datasets: Vec<String>,
pub alert_type: String,
pub anomaly_config: Option<AnomalyConfig>,
pub forecast_config: Option<ForecastConfig>,
Expand Down Expand Up @@ -326,11 +335,24 @@ impl AlertRequest {
for id in &self.targets {
TARGETS.get_target_by_id(id, &tenant_id).await?;
}
let datasets = resolve_stream_names(&self.query)?;
let datasets = match self.query_type {
AlertQueryType::Builder | AlertQueryType::Code => resolve_stream_names(&self.query)?,
AlertQueryType::Promql => self.datasets,
};

if datasets.len() != 1 {
return Err(AlertError::ValidationFailure(format!(
"Query should include only one dataset. Found: {datasets:?}"
"Alert should include only one dataset. Found: {datasets:?}"
)));
}
if self.query_type == AlertQueryType::Promql
&& !PARSEABLE
.check_or_load_stream(&datasets[0], &tenant_id)
.await
{
return Err(AlertError::ValidationFailure(format!(
"Invalid PromQL metrics stream: {}",
datasets[0]
)));
}

Expand All @@ -342,6 +364,7 @@ impl AlertRequest {
severity: self.severity,
title: self.title,
query: self.query,
query_type: self.query_type,
datasets,
alert_type: {
match self.alert_type.as_str() {
Expand Down Expand Up @@ -393,6 +416,8 @@ pub struct AlertConfig {
pub severity: Severity,
pub title: String,
pub query: String,
#[serde(default)]
pub query_type: AlertQueryType,
pub datasets: Vec<String>,
pub alert_type: AlertType,
pub threshold_config: ThresholdConfig,
Expand Down Expand Up @@ -420,6 +445,7 @@ pub struct AlertConfigResponse {
pub severity: Severity,
pub title: String,
pub query: String,
pub query_type: AlertQueryType,
pub datasets: Vec<String>,
pub alert_type: &'static str,
pub anomaly_config: Option<AnomalyConfig>,
Expand Down Expand Up @@ -466,6 +492,7 @@ impl AlertConfig {
severity: self.severity,
title: self.title,
query: self.query,
query_type: self.query_type,
datasets: self.datasets,
alert_type: {
match self.alert_type {
Expand Down
3 changes: 2 additions & 1 deletion src/alerts/alert_traits.rs
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@

use crate::{
alerts::{
AlertConfig, AlertError, AlertState, AlertType, EvalConfig, Severity,
AlertConfig, AlertError, AlertQueryType, AlertState, AlertType, EvalConfig, Severity,
alert_enums::NotificationState,
alert_structs::{Context, ThresholdConfig},
},
Expand Down Expand Up @@ -64,6 +64,7 @@ pub trait AlertTrait: Debug + Send + Sync + MetastoreObject {
fn get_severity(&self) -> &Severity;
fn get_title(&self) -> &str;
fn get_query(&self) -> &str;
fn get_query_type(&self) -> AlertQueryType;
fn get_alert_type(&self) -> &AlertType;
fn get_threshold_config(&self) -> &ThresholdConfig;
fn get_eval_config(&self) -> &EvalConfig;
Expand Down
20 changes: 18 additions & 2 deletions src/alerts/alert_types.rs
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,8 @@ use ulid::Ulid;

use crate::{
alerts::{
AlertConfig, AlertError, AlertState, AlertType, AlertVersion, EvalConfig, Severity,
ThresholdConfig,
AlertConfig, AlertError, AlertQueryType, AlertState, AlertType, AlertVersion, EvalConfig,
Severity, ThresholdConfig,
alert_enums::NotificationState,
alert_structs::{AlertStateEntry, GroupResult},
alert_traits::{AlertTrait, MessageCreation},
Expand Down Expand Up @@ -58,6 +58,8 @@ pub struct ThresholdAlert {
pub severity: Severity,
pub title: String,
pub query: String,
#[serde(default)]
pub query_type: AlertQueryType,
pub alert_type: AlertType,
pub threshold_config: ThresholdConfig,
pub eval_config: EvalConfig,
Expand Down Expand Up @@ -89,6 +91,10 @@ impl MetastoreObject for ThresholdAlert {
#[async_trait]
impl AlertTrait for ThresholdAlert {
async fn eval_alert(&self) -> Result<Option<String>, AlertError> {
if self.query_type == AlertQueryType::Promql {
return Err(AlertError::NotPresentInOSS("promql alerts"));
}

Comment thread
nikhilsinhaparseable marked this conversation as resolved.
Outdated
let time_range = extract_time_range(&self.eval_config)?;

let tenant = self.tenant_id.as_deref().unwrap_or(DEFAULT_TENANT);
Expand Down Expand Up @@ -188,6 +194,10 @@ impl AlertTrait for ThresholdAlert {
return Err(AlertError::InvalidAlertQuery("Empty query".into()));
}

if self.query_type == AlertQueryType::Promql {
return Err(AlertError::NotPresentInOSS("promql alerts"));
}

let tables = resolve_stream_names(&self.query)?;
if tables.is_empty() {
return Err(AlertError::InvalidAlertQuery(
Expand Down Expand Up @@ -315,6 +325,10 @@ impl AlertTrait for ThresholdAlert {
&self.query
}

fn get_query_type(&self) -> AlertQueryType {
self.query_type
}

fn get_severity(&self) -> &Severity {
&self.severity
}
Expand Down Expand Up @@ -436,6 +450,7 @@ impl From<AlertConfig> for ThresholdAlert {
severity: value.severity,
title: value.title,
query: value.query,
query_type: value.query_type,
alert_type: value.alert_type,
threshold_config: value.threshold_config,
eval_config: value.eval_config,
Expand All @@ -461,6 +476,7 @@ impl From<ThresholdAlert> for AlertConfig {
severity: val.severity,
title: val.title,
query: val.query,
query_type: val.query_type,
alert_type: val.alert_type,
threshold_config: val.threshold_config,
eval_config: val.eval_config,
Expand Down
39 changes: 34 additions & 5 deletions src/alerts/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ pub mod alerts_utils;
pub mod target;

pub use crate::alerts::alert_enums::{
AggregateFunction, AlertOperator, AlertState, AlertTask, AlertType, AlertVersion, EvalConfig,
LogicalOperator, NotificationState, Severity, WhereConfigOperator,
AggregateFunction, AlertOperator, AlertQueryType, AlertState, AlertTask, AlertType,
AlertVersion, EvalConfig, LogicalOperator, NotificationState, Severity, WhereConfigOperator,
};
pub use crate::alerts::alert_structs::{
AlertConfig, AlertInfo, AlertRequest, AlertStateEntry, Alerts, AlertsInfo, AlertsInfoByState,
Expand All @@ -61,6 +61,7 @@ use crate::metastore::MetastoreError;
use crate::parseable::{DEFAULT_TENANT, PARSEABLE, StreamNotFound};
use crate::query::{QUERY_SESSION, resolve_stream_names};
use crate::rbac::map::{SessionKey, sessions};
use crate::rbac::{Response, Users, role::Action};
use crate::sse::{SSE_HANDLER, SSEAlertInfo, SSEEvent};
use crate::storage;
use crate::storage::ObjectStorageError;
Expand Down Expand Up @@ -102,6 +103,33 @@ pub fn create_default_alerts_manager() -> Alerts {
alerts
}

pub async fn user_auth_for_alert_config(
session: &SessionKey,
alert: &AlertConfig,
) -> Result<(), actix_web::Error> {
match alert.query_type {
AlertQueryType::Builder | AlertQueryType::Code => {
user_auth_for_query(session, &alert.query).await
}
AlertQueryType::Promql => {
let [dataset] = alert.datasets.as_slice() else {
return Err(actix_web::error::ErrorUnauthorized(
"User does not have access to PromQL alert stream",
));
};

if Users.authorize(session.clone(), Action::Query, Some(dataset), None)
!= Response::Authorized
{
return Err(actix_web::error::ErrorUnauthorized(format!(
"User does not have access to stream- {dataset}"
)));
}
Ok(())
}
}
}

impl AlertConfig {
/// Migration function to convert v1 alerts to v2 structure
pub async fn migrate_from_v1(
Expand All @@ -125,6 +153,7 @@ impl AlertConfig {
severity: basic_fields.severity,
title: basic_fields.title,
query,
query_type: AlertQueryType::Builder,
datasets,
alert_type: AlertType::Threshold,
threshold_config,
Expand Down Expand Up @@ -627,7 +656,7 @@ impl AlertConfig {
let active_session = sessions().get_active_sessions();
let mut broadcast_to = vec![];
for (session, _, _) in active_session {
if user_auth_for_query(&session, &self.query).await.is_ok()
if user_auth_for_alert_config(&session, self).await.is_ok()
&& let SessionKey::SessionId(id) = &session
{
broadcast_to.push(*id);
Expand Down Expand Up @@ -1193,7 +1222,7 @@ impl AlertManagerTrait for Alerts {
let futures: Vec<_> = all_alerts
.into_iter()
.map(|alert| async {
if user_auth_for_query(&session.clone(), &alert.query)
if user_auth_for_alert_config(&session.clone(), &alert)
.await
.is_ok()
{
Expand All @@ -1214,7 +1243,7 @@ impl AlertManagerTrait for Alerts {
let futures: Vec<_> = all_alerts
.into_iter()
.map(|alert| async {
if user_auth_for_query(&session, &alert.query).await.is_ok() {
if user_auth_for_alert_config(&session, &alert).await.is_ok() {
Some(alert)
} else {
None
Expand Down
Loading
Loading