From 9a449a7e0aacee98fa94d40f119320c18db40fd0 Mon Sep 17 00:00:00 2001 From: Chris Beck Date: Mon, 15 Dec 2025 15:46:24 -0700 Subject: [PATCH] move http stuff to submodule in signal-gateway-bin --- signal-gateway-bin/src/listen_http.rs | 136 +++++++++++++++++++++ signal-gateway-bin/src/main.rs | 141 ++-------------------- signal-gateway/src/gateway/log_handler.rs | 2 +- 3 files changed, 144 insertions(+), 135 deletions(-) create mode 100644 signal-gateway-bin/src/listen_http.rs diff --git a/signal-gateway-bin/src/listen_http.rs b/signal-gateway-bin/src/listen_http.rs new file mode 100644 index 0000000..0872f9a --- /dev/null +++ b/signal-gateway-bin/src/listen_http.rs @@ -0,0 +1,136 @@ +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::{Gateway, alertmanager::AlertPost}; +use std::{convert::Infallible, sync::Arc, time::Duration}; +use tokio::net::TcpListener; +use tokio_util::bytes::Buf; +use tracing::{error, info, warn}; + +/// Start http listening task +pub fn start_http_task(listener: TcpListener, gateway: Arc) -> tokio::task::JoinHandle<()> { + // Loop waiting for http incoming connections, and pass them to gateway + tokio::task::spawn(async move { + loop { + let Ok((stream, remote_addr)) = listener + .accept() + .await + .inspect_err(|err| error!("Error accepting connection: {err}")) + else { + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + }; + info!("New connection from: {}", remote_addr); + + // Spawn a new task to handle each connection + let thread_gateway = gateway.clone(); + tokio::spawn(async move { + let io = TokioIo::new(stream); + + // Serve the connection using auto protocol detection (HTTP/1 or HTTP/2) + if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new()) + .serve_connection( + io, + service_fn(|req| handle_http_request(thread_gateway.clone(), req)), + ) + .await + { + error!("Error serving connection: {err}"); + } + }); + } + }) +} + +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()), + )), + } +} + diff --git a/signal-gateway-bin/src/main.rs b/signal-gateway-bin/src/main.rs index 6efc9c4..8088b62 100644 --- a/signal-gateway-bin/src/main.rs +++ b/signal-gateway-bin/src/main.rs @@ -3,19 +3,12 @@ #![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, alertmanager::AlertPost}; +use signal_gateway::{CommandRouter, Gateway, GatewayConfig, Handling}; use signal_gateway_app_code::AppCodeTools; use signal_gateway_assistant_claude::{ClaudeAssistant, ClaudeConfig}; -use std::{ - convert::Infallible, env, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration, -}; +use std::{env, fs, net::SocketAddr, path::PathBuf, sync::Arc}; use tokio::net::TcpListener; -use tokio_util::{bytes::Buf, sync::CancellationToken}; +use tokio_util::sync::CancellationToken; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; @@ -25,6 +18,9 @@ use admin_http::AdminHttpConfig; mod app_code; use app_code::AppCodeConfigExt; +mod listen_http; +use listen_http::start_http_task; + use signal_gateway_log_ingest::{JsonConfig, SyslogConfig}; /// Top-level configuration for signal-gateway. @@ -204,134 +200,11 @@ async fn main() -> Result<(), Box> { Ok(()) } -fn start_http_task(listener: TcpListener, gateway: Arc) -> tokio::task::JoinHandle<()> { - // Loop waiting for http incoming connections, and pass them to gateway - tokio::task::spawn(async move { - loop { - let Ok((stream, remote_addr)) = listener - .accept() - .await - .inspect_err(|err| error!("Error accepting connection: {err}")) - else { - tokio::time::sleep(Duration::from_secs(1)).await; - continue; - }; - info!("New connection from: {}", remote_addr); - - // Spawn a new task to handle each connection - let thread_gateway = gateway.clone(); - tokio::spawn(async move { - let io = TokioIo::new(stream); - - // Serve the connection using auto protocol detection (HTTP/1 or HTTP/2) - if let Err(err) = auto::Builder::new(hyper_util::rt::TokioExecutor::new()) - .serve_connection( - io, - service_fn(|req| handle_http_request(thread_gateway.clone(), req)), - ) - .await - { - error!("Error serving connection: {err}"); - } - }); - } - }) -} - -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::*; use conf::Conf; + use std::time::Duration; #[test] fn test_toml_config() { diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index b12f0b3..d15bed4 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/signal-gateway/src/gateway/log_handler.rs @@ -59,7 +59,7 @@ pub struct LogHandlerConfig { #[conf(long, env, default_value = "64")] pub log_buffer_size: usize, /// Max age of an origin, after this we remove it to reclaim memory - #[conf(long, env, value_parser = conf_extra::parse_duration, default_value = "3d")] + #[conf(long, env, value_parser = conf_extra::parse_duration, default_value = "72h")] pub max_origin_age: Duration, /// Debug logging level for suppressed messages. /// 0 = no logging, 1 = log only overall limiter, 2 = log routes + overall, 3 = log all.