add admin_http post forwarding
This commit is contained in:
Generated
+1
@@ -1677,6 +1677,7 @@ dependencies = [
|
||||
"dotenvy",
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"reqwest",
|
||||
"signal-gateway",
|
||||
"syslog_rfc5424",
|
||||
"tokio",
|
||||
|
||||
@@ -23,6 +23,7 @@ conf-extra = { workspace = true }
|
||||
dotenvy = { workspace = true }
|
||||
hyper = { workspace = true }
|
||||
hyper-util = { workspace = true }
|
||||
reqwest = { workspace = true }
|
||||
syslog_rfc5424 = { workspace = true }
|
||||
tokio = { workspace = true, features = ["rt-multi-thread", "signal"] }
|
||||
tokio-util = { workspace = true }
|
||||
|
||||
@@ -0,0 +1,64 @@
|
||||
//! Admin HTTP client for forwarding messages to an HTTP server
|
||||
//!
|
||||
//! This module handles admin messages not handled by the gateway by making an HTTP POST request
|
||||
//! with the message as the body, and returning the response body as the reply.
|
||||
|
||||
use conf::Conf;
|
||||
use signal_gateway::{AdminMessageResponse, MessageHandler, MessageHandlerResult};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Configuration for the admin HTTP client
|
||||
#[derive(Clone, Conf, Debug)]
|
||||
pub struct AdminHttpConfig {
|
||||
/// URL to POST admin commands to
|
||||
#[conf(long, env)]
|
||||
pub url: String,
|
||||
/// Timeout for the HTTP request
|
||||
#[conf(long, env, default_value = "5s", value_parser = conf_extra::parse_duration)]
|
||||
pub timeout: Duration,
|
||||
}
|
||||
|
||||
impl AdminHttpConfig {
|
||||
/// Create a message handler function from this config.
|
||||
///
|
||||
/// The returned handler makes an HTTP POST request to the configured URL
|
||||
/// with the message as the body, and returns the response body.
|
||||
pub fn into_handler(self) -> MessageHandler {
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(self.timeout)
|
||||
.build()
|
||||
.expect("Failed to build HTTP client");
|
||||
|
||||
Box::new(move |message: String| {
|
||||
let client = client.clone();
|
||||
let url = self.url.clone();
|
||||
Box::pin(async move { handle_message(&client, &url, message).await })
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle a message by POSTing it to the configured HTTP server
|
||||
async fn handle_message(
|
||||
client: &reqwest::Client,
|
||||
url: &str,
|
||||
message: String,
|
||||
) -> MessageHandlerResult {
|
||||
let response = client
|
||||
.post(url)
|
||||
.body(message)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|err| (502u16, format!("HTTP request failed: {err}").into()))?;
|
||||
|
||||
let status = response.status();
|
||||
let body = response
|
||||
.text()
|
||||
.await
|
||||
.map_err(|err| (502u16, format!("Failed to read response body: {err}").into()))?;
|
||||
|
||||
if !status.is_success() {
|
||||
return Err((status.as_u16(), body.into()));
|
||||
}
|
||||
|
||||
Ok(AdminMessageResponse::new(body))
|
||||
}
|
||||
@@ -1,20 +1,44 @@
|
||||
use conf::Conf;
|
||||
use conf::{Conf, Subcommands};
|
||||
use hyper::service::service_fn;
|
||||
use hyper_util::rt::TokioIo;
|
||||
use hyper_util::server::conn::auto;
|
||||
use signal_gateway::{Gateway, GatewayConfig};
|
||||
use signal_gateway::{Gateway, GatewayConfig, MessageHandler};
|
||||
use std::{net::SocketAddr, sync::Arc, time::Duration};
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{error, info, warn};
|
||||
use tracing_subscriber::EnvFilter;
|
||||
|
||||
mod admin_http;
|
||||
use admin_http::AdminHttpConfig;
|
||||
|
||||
mod admin_netcat;
|
||||
use admin_netcat::AdminNetcatConfig;
|
||||
|
||||
mod syslog;
|
||||
use syslog::SyslogUdpConfig;
|
||||
|
||||
/// Admin message handler configuration - select how non-command messages are handled
|
||||
#[derive(Clone, Debug, Subcommands)]
|
||||
enum AdminHandlerCommand {
|
||||
/// Forward (unhandled) admin messages to a TCP endpoint (netcat-style)
|
||||
/// Useful if an http server would be heavy in the target process
|
||||
#[conf(name = "admin-netcat")]
|
||||
Netcat(AdminNetcatConfig),
|
||||
/// Forward (unhandled) admin messages to an HTTP endpoint via POST
|
||||
#[conf(name = "admin-http")]
|
||||
Http(AdminHttpConfig),
|
||||
}
|
||||
|
||||
impl AdminHandlerConfig {
|
||||
fn into_handler(self) -> MessageHandler {
|
||||
match self {
|
||||
AdminHandlerConfig::Netcat(config) => config.into_handler(),
|
||||
AdminHandlerConfig::Http(config) => config.into_handler(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Conf, Debug)]
|
||||
struct Config {
|
||||
/// If true, just validate config and don't start
|
||||
@@ -25,8 +49,9 @@ struct Config {
|
||||
http_listen_addr: SocketAddr,
|
||||
#[conf(flatten, prefix)]
|
||||
syslog_udp: Option<SyslogUdpConfig>,
|
||||
#[conf(flatten, prefix)]
|
||||
admin_netcat: Option<AdminNetcatConfig>,
|
||||
/// Optional admin message handler (netcat or http)
|
||||
#[conf(subcommands)]
|
||||
admin_handler: Option<AdminHandlerCommand>,
|
||||
#[conf(flatten)]
|
||||
gateway: GatewayConfig,
|
||||
}
|
||||
@@ -73,7 +98,7 @@ async fn main() {
|
||||
|
||||
let token = CancellationToken::new();
|
||||
|
||||
let message_handler = config.admin_netcat.map(|c| c.into_handler());
|
||||
let message_handler = config.admin_handler.map(|c| c.into_handler());
|
||||
let gateway = Arc::new(Gateway::new(config.gateway, token.clone(), message_handler).await);
|
||||
|
||||
let listener = TcpListener::bind(config.http_listen_addr).await.unwrap();
|
||||
|
||||
Reference in New Issue
Block a user