diff --git a/Cargo.lock b/Cargo.lock index 6dea536..ac408dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1724,9 +1724,6 @@ dependencies = [ "conf-extra", "displaydoc", "futures-util", - "http", - "http-body", - "http-body-util", "humantime", "jsonrpsee", "prometheus-http-client", @@ -1797,6 +1794,9 @@ dependencies = [ "conf", "conf-extra", "dotenvy", + "http", + "http-body", + "http-body-util", "hyper", "hyper-util", "reqwest", diff --git a/signal-gateway-bin/Cargo.toml b/signal-gateway-bin/Cargo.toml index 9fbcde4..c54de95 100644 --- a/signal-gateway-bin/Cargo.toml +++ b/signal-gateway-bin/Cargo.toml @@ -25,6 +25,9 @@ async-trait = { workspace = true } conf = { workspace = true } conf-extra = { workspace = true } dotenvy = { workspace = true } +http = { workspace = true } +http-body = { workspace = true } +http-body-util = { workspace = true } hyper = { workspace = true } hyper-util = { workspace = true } reqwest = { workspace = true } diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 3ee8dbb..6efc9c4 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -3,14 +3,19 @@ #![deny(missing_docs)] use conf::Conf; +use http::{Method, Request, Response, StatusCode}; +use http_body::Body; +use http_body_util::BodyExt; use hyper::service::service_fn; use hyper_util::{rt::TokioIo, server::conn::auto}; -use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling}; +use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling, alertmanager::AlertPost}; use signal_gateway_app_code::AppCodeTools; use signal_gateway_assistant_claude::{ClaudeAssistant, ClaudeConfig}; -use std::{env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; +use std::{ + convert::Infallible, env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration, +}; use tokio::net::TcpListener; -use tokio_util::sync::CancellationToken; +use tokio_util::{bytes::Buf, sync::CancellationToken}; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; @@ -222,10 +227,7 @@ fn start_http_task(listener: TcpListener, gateway: Arc) -> tokio::task: if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new()) .serve_connection( io, - service_fn(|req| { - let thread_gateway = thread_gateway.clone(); - async move { thread_gateway.handle_http_request(req).await } - }), + service_fn(|req| handle_http_request(thread_gateway.clone(), req)), ) .await { @@ -236,6 +238,96 @@ fn start_http_task(listener: TcpListener, gateway: Arc) -> tokio::task: }) } +async fn handle_http_request( + gateway: Arc, + req: Request, +) -> Result, Infallible> { + match handle_http_request_impl(gateway, req).await { + Ok(resp) => Ok(resp), + Err(resp) => Ok(resp), + } +} + +async fn handle_http_request_impl( + gateway: Arc, + req: Request, +) -> Result, Response> +where + B: Body + Send, + B::Data: Buf + Send, + B::Error: std::fmt::Display, +{ + info!( + "Received http request: {} {} (version: {:?})", + req.method(), + req.uri().path(), + req.version() + ); + + fn ok_resp() -> Response { + Response::new("OK".into()) + } + fn err_resp(code: StatusCode, text: impl Into) -> Response { + let mut resp = Response::new(text.into()); + *resp.status_mut() = code; + resp + } + + match req.uri().path() { + "/" | "/health" | "/ready" => { + if !matches!(req.method(), &Method::GET | &Method::HEAD) { + Ok(err_resp( + StatusCode::NOT_IMPLEMENTED, + "Use GET or HEAD with this route", + )) + } else { + Ok(ok_resp()) + } + } + "/alert" => { + if !matches!(req.method(), &Method::POST) { + return Ok(err_resp( + StatusCode::NOT_IMPLEMENTED, + "Use POST with this route", + )); + } + let body_bytes = req + .into_body() + .collect() + .await + .map_err(|err| { + err_resp( + StatusCode::BAD_REQUEST, + format!("When reading body bytes: {err}"), + ) + })? + .to_bytes() + .to_vec(); + + let body_text = str::from_utf8(&body_bytes).map_err(|err| { + warn!("When reading body bytes: {err}"); + err_resp(StatusCode::BAD_REQUEST, "Request body was not utf-8") + })?; + + let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| { + error!("Could not parse json: {err}:\n{body_text}"); + err_resp(StatusCode::BAD_REQUEST, "Invalid Json") + })?; + + if let Err(msg) = gateway.handle_alertmanager_post(alert_msg).await { + error!("gateway (handle_alertmanager_post): {msg}"); + Ok(err_resp(StatusCode::INTERNAL_SERVER_ERROR, msg)) + } else { + Ok(ok_resp()) + } + } + _ => Ok(err_resp( + StatusCode::NOT_FOUND, + format!("Not found '{} {}'", req.method(), req.uri().path()), + )), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index 9785ab1..d3bc919 100644 --- a/signal-gateway/Cargo.toml +++ b/signal-gateway/Cargo.toml @@ -21,9 +21,6 @@ conf = { workspace = true } conf-extra = { workspace = true } displaydoc = { workspace = true } futures-util = { workspace = true } -http = { workspace = true } -http-body = { workspace = true } -http-body-util = { workspace = true } humantime = { workspace = true } jsonrpsee = { workspace = true } regex = { workspace = true } diff --git a/signal-gateway/src/assistant/mod.rs b/signal-gateway/src/assistant/mod.rs index 52b7128..6a37d93 100644 --- a/signal-gateway/src/assistant/mod.rs +++ b/signal-gateway/src/assistant/mod.rs @@ -56,7 +56,7 @@ impl AssistantAgent { /// Spawns a background worker task that processes requests serially. pub fn new(assistant: Box, cancellation_token: CancellationToken) -> Self { let (input_tx, input_rx) = mpsc::channel(REQUEST_QUEUE_SIZE); - let (stop_tx, stop_rx) = mpsc::channel(REQUEST_QUEUE_SIZE); + let (stop_tx, stop_rx) = mpsc::channel(3); let worker = AssistantWorker::new(assistant, input_rx, stop_rx, cancellation_token.clone()); diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 36f6351..07e1784 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/signal-gateway/src/gateway/mod.rs @@ -16,9 +16,6 @@ use async_trait::async_trait; use chrono::Utc; use conf::{Conf, Subcommands}; use futures_util::FutureExt; -use http::{Method, Request, Response, StatusCode}; -use http_body::Body; -use http_body_util::BodyExt; use prometheus_http_client::{AlertStatus, ExtractLabels}; use std::{ fmt::Write, @@ -31,7 +28,7 @@ use tokio::{ join, sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel}, }; -use tokio_util::{bytes::Buf, sync::CancellationToken}; +use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; mod signal_trust_set; @@ -655,68 +652,6 @@ impl Gateway { } } - /// Handle an incoming HTTP request (e.g., webhooks from Alertmanager). - pub async fn handle_http_request(&self, req: Request) -> Result, String> - where - B: Body + Send, - B::Data: Buf + Send, - B::Error: std::fmt::Display, - { - info!( - "Received http request: {} {} (version: {:?})", - req.method(), - req.uri().path(), - req.version() - ); - - fn ok_resp() -> Response { - Response::new("OK".into()) - } - fn err_resp(code: StatusCode, text: impl Into) -> Response { - let mut resp = Response::new(text.into()); - *resp.status_mut() = code; - resp - } - - match req.uri().path() { - "/" | "/health" | "/ready" => { - if !matches!(req.method(), &Method::GET | &Method::HEAD) { - Ok(err_resp( - StatusCode::NOT_IMPLEMENTED, - "Use GET or HEAD with this route", - )) - } else { - Ok(ok_resp()) - } - } - "/alert" => { - if !matches!(req.method(), &Method::POST) { - return Ok(err_resp( - StatusCode::NOT_IMPLEMENTED, - "Use POST with this route", - )); - } - let v = req - .into_body() - .collect() - .await - .map_err(|err| format!("When reading body bytes: {err}"))? - .to_bytes() - .to_vec(); - - if let Err((code, msg)) = self.handle_post_alert(&v).await { - Ok(err_resp(code, msg)) - } else { - Ok(ok_resp()) - } - } - _ => Ok(err_resp( - StatusCode::NOT_FOUND, - format!("Not found '{} {}'", req.method(), req.uri().path()), - )), - } - } - async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult { match cmd { GatewayCommand::Log { filter } => { @@ -887,17 +822,8 @@ impl Gateway { } } - async fn handle_post_alert(&self, body_bytes: &[u8]) -> Result<(), (StatusCode, &'static str)> { - let body_text = str::from_utf8(body_bytes).map_err(|err| { - warn!("When reading body bytes: {err}"); - (StatusCode::BAD_REQUEST, "Request body was not utf-8") - })?; - - let alert_msg: AlertPost = serde_json::from_str(body_text).map_err(|err| { - error!("Could not parse json: {err}:\n{body_text}"); - (StatusCode::BAD_REQUEST, "Invalid Json") - })?; - + /// Handle a POST body from alertmanager + pub async fn handle_alertmanager_post(&self, alert_msg: AlertPost) -> Result<(), &'static str> { let text = self .format_alert_text(&alert_msg) .unwrap_or_else(|err| format!("error formatting alert text: {err}:\n{alert_msg:#?}")); @@ -921,20 +847,16 @@ impl Gateway { } // Build summary: status sigils followed by alert names - let summary = alert_msg - .alerts - .iter() - .map(|alert| { - let symbol = alert.status.symbol(); - let name = alert - .labels - .get("alertname") - .map(|s| s.as_str()) - .unwrap_or("?"); - format!("{symbol}{name}") - }) - .collect::>() - .join(" "); + let mut summary = String::with_capacity(32); + alert_msg.alerts.iter().for_each(|alert| { + let symbol = alert.status.symbol(); + let name = alert + .labels + .get("alertname") + .map(|s| s.as_str()) + .unwrap_or("?"); + write!(&mut summary, "{symbol}{name} ").unwrap(); + }); self.signal_alert_mq_tx .send(SignalAlertMessage { @@ -944,13 +866,7 @@ impl Gateway { summary: Summary::Owned(summary.into()), destination_override: None, }) - .map_err(|_err| { - error!("Could not send alert message, queue is closed"); - ( - StatusCode::INTERNAL_SERVER_ERROR, - "Can't send signal msg right now, queue is closed", - ) - }) + .map_err(|_err| "Can't send signal msg right now, queue is closed") } fn format_alert_text(&self, msg: &AlertPost) -> Result {