feat: introduce alert emails in backend (#2825)

Addresses #2601.

No breaking API changes. For destination and template, previous API
payloads work as "http" type destination/template payload. For "email"
destination/template -

**Email destination**
supports multiple emails. Example payload -
```
{
    "name": "email destination",
    "template": "my new template",
    "emails": ["example@hi.com", "hi@mail.com"],
    "type": "email"
}
```
The above payload creates the following Destination instance - 
```
{
    "name": "email destination",
    "url": "", //ignored
    "method": "post", //ignored
    "skip_tls_verify": false, //ignored
    "template": "my new template",
    "emails": [
      "subhradeepc461@gmail.com",
      "chakrabortyabhradeep79@gmail.com"
    ],
    "type": "email"
}
```

**Email template**
Example API payload. Supports all variables in "{}" as it does for the
"http" template -
```
{
    "name": "my new template",
    "body": "<b>Hello</b> {alert_name} <i>World!</i>",
    "isDefault": true,
    "type": "email"
}
```

src/common/meta/alerts/destinations.rs

@@ -25,13 +25,32 @@ use super::templates::Template;
 pub struct Destination {
     #[serde(default)]
     pub name: String,
+    /// Required for `Http` destination_type
+    #[serde(default)]
     pub url: String,
+    /// Required for `Http` destination_type
+    #[serde(default)]
     pub method: HTTPType,
     #[serde(default)]
     pub skip_tls_verify: bool,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub headers: Option<HashMap<String, String>>,
     pub template: String,
+    /// Required when `destination_type` is `Email`
+    #[serde(default)]
+    pub emails: Vec<String>,
+    #[serde(rename = "type")]
+    #[serde(default)]
+    pub destination_type: DestinationType,
+}
+
+#[derive(Serialize, Debug, Default, PartialEq, Eq, Deserialize, Clone, ToSchema)]
+pub enum DestinationType {
+    #[default]
+    #[serde(rename = "http")]
+    Http,
+    #[serde(rename = "email")]
+    Email,
 }
 
 impl Destination {
@@ -43,6 +62,8 @@ impl Destination {
             skip_tls_verify: self.skip_tls_verify,
             headers: self.headers.clone(),
             template,
+            emails: self.emails.clone(),
+            destination_type: self.destination_type.clone(),
         }
     }
 }
@@ -57,6 +78,8 @@ pub struct DestinationWithTemplate {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub headers: Option<HashMap<String, String>>,
     pub template: Template,
+    pub emails: Vec<String>,
+    pub destination_type: DestinationType,
 }
 
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]

src/common/meta/alerts/templates.rs

@@ -16,6 +16,8 @@
 use serde::{Deserialize, Serialize};
 use utoipa::ToSchema;
 
+use super::destinations::DestinationType;
+
 #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize, Deserialize, ToSchema)]
 pub struct Template {
     #[serde(default)]
@@ -25,4 +27,9 @@ pub struct Template {
     #[serde(rename = "isDefault")]
     #[serde(default)]
     pub is_default: Option<bool>,
+    /// Indicates whether the body is
+    /// http or email body
+    #[serde(rename = "type")]
+    #[serde(default)]
+    pub template_type: DestinationType,
 }

src/handler/http/request/alerts/destinations.rs

@@ -48,7 +48,10 @@ pub async fn save_destination(
     let dest = dest.into_inner();
     match destinations::save(&org_id, "", dest, true).await {
         Ok(_) => Ok(MetaHttpResponse::ok("Alert destination saved")),
-        Err(e) => Ok(MetaHttpResponse::bad_request(e)),
+        Err(e) => match e {
+            (http::StatusCode::BAD_REQUEST, e) => Ok(MetaHttpResponse::bad_request(e)),
+            (_, e) => Ok(MetaHttpResponse::internal_error(e)),
+        },
     }
 }
 
@@ -80,7 +83,10 @@ pub async fn update_destination(
     let name = name.trim();
     match destinations::save(&org_id, name, dest, false).await {
         Ok(_) => Ok(MetaHttpResponse::ok("Alert destination saved")),
-        Err(e) => Ok(MetaHttpResponse::bad_request(e)),
+        Err(e) => match e {
+            (http::StatusCode::BAD_REQUEST, e) => Ok(MetaHttpResponse::bad_request(e)),
+            (_, e) => Ok(MetaHttpResponse::internal_error(e)),
+        },
     }
 }
 

src/service/alerts/destinations.rs

@@ -14,12 +14,13 @@
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 use actix_web::http;
+use config::CONFIG;
 
 use crate::{
     common::{
         infra::config::STREAM_ALERTS,
         meta::{
-            alerts::destinations::{Destination, DestinationWithTemplate},
+            alerts::destinations::{Destination, DestinationType, DestinationWithTemplate},
             authz::Authz,
         },
         utils::auth::{remove_ownership, set_ownership},
@@ -32,35 +33,74 @@ pub async fn save(
     name: &str,
     mut destination: Destination,
     create: bool,
-) -> Result<(), anyhow::Error> {
-    if db::alerts::templates::get(org_id, &destination.template)
-        .await
-        .is_err()
-    {
-        return Err(anyhow::anyhow!(
-            "Alert template {} not found",
-            destination.template
-        ));
+) -> Result<(), (http::StatusCode, anyhow::Error)> {
+    // First validate the `destination` according to its `destination_type`
+    match destination.destination_type {
+        DestinationType::Http => {
+            if destination.url.is_empty() {
+                return Err((
+                    http::StatusCode::BAD_REQUEST,
+                    anyhow::anyhow!("Alert destination URL needs to be specified"),
+                ));
+            }
+        }
+        DestinationType::Email => {
+            if destination.emails.is_empty() {
+                return Err((
+                    http::StatusCode::BAD_REQUEST,
+                    anyhow::anyhow!("Atleast one alert destination email is required"),
+                ));
+            }
+            if !CONFIG.smtp.smtp_enabled {
+                return Err((
+                    http::StatusCode::INTERNAL_SERVER_ERROR,
+                    anyhow::anyhow!("SMTP not configured"),
+                ));
+            }
+        }
     }
+
     if !name.is_empty() {
         destination.name = name.to_string();
     }
     if destination.name.is_empty() {
-        return Err(anyhow::anyhow!("Alert destination name is required"));
+        return Err((
+            http::StatusCode::BAD_REQUEST,
+            anyhow::anyhow!("Alert destination name is required"),
+        ));
     }
     if destination.name.contains('/') {
-        return Err(anyhow::anyhow!("Alert destination name cannot contain '/'"));
+        return Err((
+            http::StatusCode::BAD_REQUEST,
+            anyhow::anyhow!("Alert destination name cannot contain '/'"),
+        ));
+    }
+
+    if db::alerts::templates::get(org_id, &destination.template)
+        .await
+        .is_err()
+    {
+        return Err((
+            http::StatusCode::BAD_REQUEST,
+            anyhow::anyhow!("Alert template {} not found", destination.template),
+        ));
     }
 
     match db::alerts::destinations::get(org_id, &destination.name).await {
         Ok(_) => {
             if create {
-                return Err(anyhow::anyhow!("Alert destination already exists"));
+                return Err((
+                    http::StatusCode::BAD_REQUEST,
+                    anyhow::anyhow!("Alert destination already exists"),
+                ));
             }
         }
         Err(_) => {
             if !create {
-                return Err(anyhow::anyhow!("Alert destination not found"));
+                return Err((
+                    http::StatusCode::BAD_REQUEST,
+                    anyhow::anyhow!("Alert destination not found"),
+                ));
             }
         }
     }
@@ -72,7 +112,7 @@ pub async fn save(
             }
             Ok(())
         }
-        Err(e) => Err(e),
+        Err(e) => Err((http::StatusCode::BAD_REQUEST, e)),
     }
 }
 

src/service/alerts/mod.rs

@@ -25,15 +25,16 @@ use config::{
         base64,
         json::{Map, Value},
     },
-    CONFIG,
+    CONFIG, SMTP_CLIENT,
 };
+use lettre::{message::SinglePart, AsyncTransport, Message};
 
 use super::promql;
 use crate::{
     common::{
         meta::{
             alerts::{
-                destinations::{DestinationWithTemplate, HTTPType},
+                destinations::{DestinationType, DestinationWithTemplate, HTTPType},
                 AggFunction, Alert, Condition, Operator, QueryCondition, QueryType,
             },
             authz::Authz,
@@ -790,7 +791,18 @@ pub async fn send_notification(
     } else {
         process_row_template(&alert.row_template, alert, rows)
     };
-    let msg = process_dest_template(&dest.template.body, alert, rows, &rows_tpl_val).await;
+    let msg: String = process_dest_template(&dest.template.body, alert, rows, &rows_tpl_val).await;
+
+    match dest.destination_type {
+        DestinationType::Http => send_http_notification(dest, msg.clone()).await,
+        DestinationType::Email => send_email_notification(&alert.name, dest, msg).await,
+    }
+}
+
+pub async fn send_http_notification(
+    dest: &DestinationWithTemplate,
+    msg: String,
+) -> Result<(), anyhow::Error> {
     let client = if dest.skip_tls_verify {
         reqwest::Client::builder()
             .danger_accept_invalid_certs(true)
@@ -831,6 +843,37 @@ pub async fn send_notification(
     Ok(())
 }
 
+pub async fn send_email_notification(
+    alert_name: &str,
+    dest: &DestinationWithTemplate,
+    msg: String,
+) -> Result<(), anyhow::Error> {
+    if !CONFIG.smtp.smtp_enabled {
+        return Err(anyhow::anyhow!("SMTP configuration not enabled"));
+    }
+
+    let mut recepients = vec![];
+    for recepient in &dest.emails {
+        recepients.push(recepient);
+    }
+
+    let mut email = Message::builder()
+        .from(CONFIG.smtp.smtp_from_email.parse()?)
+        .subject(format!("Openobserve Alert - {}", alert_name));
+
+    for recepient in recepients {
+        email = email.to(recepient.parse()?);
+    }
+
+    let email = email.singlepart(SinglePart::html(msg)).unwrap();
+
+    // Send the email
+    match SMTP_CLIENT.as_ref().unwrap().send(email).await {
+        Ok(_) => Ok(()),
+        Err(e) => Err(anyhow::anyhow!("Error sending email: {e}")),
+    }
+}
+
 fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]) -> Vec<String> {
     let alert_type = if alert.is_real_time {
         "realtime"