Github/aralez
1
0
Fork 0
mirror of https://github.com/sadoyan/aralez.git synced 2026-10-23 06:01:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

Improved error handling, anti unwrap()-ing

Browse Source
This commit is contained in:
Ara Sadoyan
2026-10-20 17:06:55 +02:00
parent f895ba4554
commit 0c489b649a
22 changed files with 162 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/bgservice.rs
+5 -6
View File
@@ -30,16 +30,15 @@ impl BackgroundService for LB {
let config = load_configuration(self.config.upstreams_conf.clone().as_str(), "filepath")
.await
.0
.expect("Failed to load configuration");
.die("Failed to load configuration");
match config.typecfg.as_str() {
"file" => {
info!("Running File discovery, requested type is: {}", config.typecfg);
tx.send(config).await.unwrap();
tx.send(config).await.die("Running File discovery");
let file_load = FromFileProvider {
path: self.config.upstreams_conf.clone(),
};
// let _ = tokio::spawn(async move { file_load.start(tx).await });
drop(tokio::spawn(async move { file_load.start(tx).await }));
}
"kubernetes" => {
@@ -80,7 +79,7 @@ impl BackgroundService for LB {
let ff = self.ump_full.clone();
let im = self.ump_byid.clone();
let (hc_method, hc_interval) = (self.config.hc_method.clone(), self.config.hc_interval);
tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().unwrap())).await });
tokio::spawn(async move { healthcheck::hc2(uu, ff, im, (&*hc_method.to_string(), hc_interval.to_string().parse().die("Unable to spawn HC worker"))).await });
tokio::spawn(async move { calc_cache_metrics().await });
let acme_provider = self.config.acme_dns_provider.clone();
tokio::spawn(async move { refresh_order(certdir, confdir, acme_provider).await });
@@ -93,7 +92,7 @@ impl BackgroundService for LB {
break;
}
val = rx.recv() => {
let start = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
let start = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
if let Some(ss) = val {
clone_dashmap_into(&ss.upstreams, &self.ump_full);
clone_dashmap_into(&ss.upstreams, &self.ump_upst);
@@ -140,7 +139,7 @@ impl BackgroundService for LB {
}
}
}
let end = SystemTime::now().duration_since(UNIX_EPOCH).unwrap();
let end = SystemTime::now().duration_since(UNIX_EPOCH).die(format!("ON {} {}", file!(), line!()).as_str());
let startup = end - start;
info!("Loading upstreams took: {}ms", startup.as_millis());
}
src/core/start.rs
+1 -1
View File
@@ -128,7 +128,7 @@ pub fn run() {
certs_for_watcher.store(Arc::new(new_certs.die("Failed to build the watcher")));
let mut tls_settings =
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).expect("unable to load or parse cert/key");
TlsSettings::intermediate(&certs_for_callback.load().default_cert_path, &certs_for_callback.load().default_key_path).die("unable to load or parse cert/key");
grades::set_tsl_grade(&mut tls_settings, grade.as_str());
tls_settings.set_servername_callback(move |ssl_ref: &mut SslRef, ssl_alert: &mut SslAlert| certs_for_callback.load().server_name_callback(ssl_ref, ssl_alert));
src/core/webserver.rs
+23 -49
View File
@@ -3,7 +3,7 @@ use crate::ingress::APIUpstreamProvider;
use crate::tls::acme::http01::{acme_create, acme_order, http01_challenge};
use crate::utils::lazylock::NODE_STATUS;
use crate::utils::metrics::{MEMORY_USAGE, OPEN_FILES, get_memory_usage, get_open_files};
use crate::utils::tools::{upstreams_liveness_json, upstreams_to_json};
use crate::utils::tools::{Die, upstreams_liveness_json, upstreams_to_json};
use crate::utils::types::{Config, Configuration, UpstreamsDashMap};
use axum::body::Body;
use axum::extract::{Query, State};
@@ -86,14 +86,16 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
let listener = port_is_available("Config API", &config.address).await;
info!("Starting the API server on: {}", config.address);
// let api_server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
let api_server = tokio::spawn(async move { axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap() });
let api_server = tokio::spawn(async move {
axum::serve(listener, app.into_make_service_with_connect_info::<SocketAddr>())
.await
.die("Can't start API server")
});
let (tx, mut rx) = mpsc::channel(1);
std::thread::spawn(move || {
let mut signals = Signals::new([SIGQUIT]).unwrap();
let mut signals = Signals::new([SIGQUIT]).die("Can't initialize OS SIGNALS");
if let Some(sig) = signals.forever().next() {
tx.blocking_send(sig).unwrap();
tx.blocking_send(sig).die("Can;t send signal to main thread");
}
});
rx.recv().await;
@@ -106,7 +108,7 @@ pub async fn run_server(config: &APIUpstreamProvider, mut to_return: mpsc::Sende
async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, String>>, content: String) -> impl IntoResponse {
if !st.config_api_enabled {
return Response::builder().status(StatusCode::FORBIDDEN).body(Body::from("Config API is disabled !\n")).unwrap();
return (StatusCode::FORBIDDEN, "Config API is disabled !\n").into_response();
}
let strcontent = content.as_str();
@@ -118,11 +120,11 @@ async fn conf(State(st): State<AppState>, Query(params): Query<HashMap<String, S
} else {
tokio::spawn(async move { apply_config(content.as_str(), st, false).await });
}
Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap()
(StatusCode::OK, "Accepted! Applying in background\n").into_response()
}
Err(err) => {
error!("Failed to parse upstreams file: {}", err);
Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap()
(StatusCode::BAD_GATEWAY, format!("Failed: {}\n", err)).into_response()
}
}
}
@@ -133,10 +135,7 @@ async fn workerconf(State(st): State<AppState>, ConnectInfo(addr): ConnectInfo<S
let api_key = headers.get("API-KEY").and_then(|v| v.to_str().ok());
if api_key != st.master_key.as_deref() {
warn!("Unauthorised access to workerconf endpoint from {}", addr.ip());
return Response::builder()
.status(StatusCode::UNAUTHORIZED)
.body(Body::from("Unauthorized: Invalid API-KEY\n"))
.unwrap();
return (StatusCode::UNAUTHORIZED, "Unauthorized: Invalid API-KEY\n").into_response();
}
if let Some(member) = NODE_STATUS.get() {
if let Some(mbr) = member.as_ref() {
@@ -148,31 +147,22 @@ async fn workerconf(State(st): State<AppState>, ConnectInfo(addr): ConnectInfo<S
match parsed {
Ok(_) => {
tokio::spawn(async move { apply_config(content.as_str(), st, true).await });
return Response::builder().status(StatusCode::OK).body(Body::from("Accepted! Applying in background\n")).unwrap();
return (StatusCode::OK, "Accepted! Applying in background\n").into_response();
}
Err(err) => {
error!("Failed to parse upstreams file: {}", err);
return Response::builder().status(StatusCode::BAD_GATEWAY).body(Body::from(format!("Failed: {}\n", err))).unwrap();
return (StatusCode::BAD_GATEWAY, format!("Failed: {}\n", err)).into_response();
}
}
}
_ => {
return Response::builder()
.status(StatusCode::BAD_GATEWAY)
.body(Body::from("Node is not running on worker mode !\n"))
.unwrap();
return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
}
}
}
return Response::builder()
.status(StatusCode::BAD_GATEWAY)
.body(Body::from("Node is not running on worker mode !\n"))
.unwrap();
return (StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response();
}
Response::builder()
.status(StatusCode::BAD_GATEWAY)
.body(Body::from("Node is not running on worker mode !\n"))
.unwrap()
(StatusCode::BAD_GATEWAY, "Node is not running on worker mode !\n").into_response()
}
async fn apply_config(content: &str, st: AppState, save: bool) {
@@ -236,17 +226,13 @@ async fn metrics() -> impl IntoResponse {
let encoder = TextEncoder::new();
let mut buffer = Vec::new();
if let Err(e) = encoder.encode(&metric_families, &mut buffer) {
// encoding error fallback
return Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(format!("Failed to encode metrics: {}", e)))
.unwrap();
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to encode metrics: {}", e)).into_response();
}
Response::builder()
.status(StatusCode::OK)
.header("Content-Type", encoder.format_type())
.body(Body::from(buffer))
.unwrap()
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response())
}
#[allow(clippy::needless_return)]
@@ -257,7 +243,7 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(Body::from(format!("{}", r)))
.unwrap();
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
}
if params.contains_key("all") {
let resp = upstreams_to_json(&st.current_upstreams);
@@ -267,29 +253,17 @@ async fn status(State(st): State<AppState>, Query(params): Query<HashMap<String,
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(Body::from(j))
.unwrap();
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
}
Err(e) => {
return Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(format!("Failed to get status: {}", e)))
.unwrap();
Err(e) => return (StatusCode::BAD_GATEWAY, format!("Internal Server Error {}\n", e)).into_response(),
}
}
}
Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from("Parameter mismatch"))
.unwrap()
(StatusCode::INTERNAL_SERVER_ERROR, "Parameter mismatch\n").into_response()
}
pub async fn port_is_available(name: &str, address: &str) -> TcpListener {
let addr = SocketAddr::from_str(address).unwrap_or_else(|e| panic!("{}: Invalid address format: {:?}", name, e));
let t = Duration::from_secs(2);
//if addr.ip() == IpAddr::V4(Ipv4Addr::UNSPECIFIED) {
// addr.set_ip(IpAddr::V4(Ipv4Addr::LOCALHOST));
//}
let p = addr.port();
loop {
match TcpListener::bind(addr).await {
src/ingress/consul.rs
+3 -2
View File
@@ -1,7 +1,8 @@
use crate::ingress::kuberconsul;
use crate::ingress::kuberconsul::{list_to_upstreams, ServiceDiscovery};
use crate::ingress::kuberconsul::{ServiceDiscovery, list_to_upstreams};
use crate::utils::httpclient;
use crate::utils::parceyaml::build_headers;
use crate::utils::tools::Die;
use crate::utils::types::{Configuration, GlobalServiceMapping, UpstreamsDashMap};
use async_trait::async_trait;
use dashmap::DashMap;
@@ -38,7 +39,7 @@ impl ServiceDiscovery for ConsulDiscovery {
});
let end = servers.len().saturating_sub(1);
let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
let consul_data = servers.get(num).unwrap().to_string();
let consul_data = servers.get(num).die("Reading consul data").to_string();
let ss = format!("{}/v1/catalog/service/", consul_data);
let dc = format!("{}/v1/catalog/services?filter=%22aralez.service=yes%22+in+ServiceTags", consul_data);
src/ingress/kuberconsul.rs
+4 -4
View File
@@ -1,9 +1,9 @@
use crate::utils::tools::{clone_dashmap_into, compare_dashmaps, print_upstreams};
use crate::utils::tools::{Die, clone_dashmap_into, compare_dashmaps, print_upstreams};
use crate::utils::types::{Configuration, GlobalServiceMapping, InnerMap, UpstreamsDashMap};
use async_trait::async_trait;
use dashmap::DashMap;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use tokio::fs::File;
use tokio::io::AsyncReadExt;
use tokio::sync::mpsc::Sender;
@@ -31,9 +31,9 @@ pub fn match_path(conf: &GlobalServiceMapping, upstreams: &DashMap<Arc<str>, (Ve
}
pub async fn read_token(path: &str) -> String {
let mut file = File::open(path).await.unwrap();
let mut file = File::open(path).await.die("Reading kubernetes token");
let mut contents = String::new();
file.read_to_string(&mut contents).await.unwrap();
file.read_to_string(&mut contents).await.die("Kubernetes token to string");
contents.trim().to_string()
}
src/ingress/kubernetes.rs
+6 -5
View File
@@ -9,10 +9,11 @@ use std::env;
use std::sync::Arc;
use crate::ingress::kuberconsul;
use crate::ingress::kuberconsul::{clone_compare, list_to_upstreams, ServiceDiscovery};
use crate::ingress::kubewatcher::{start_ingress_watcher_with_config, AralezRouteUpdate};
use rustls::crypto::ring::default_provider;
use crate::ingress::kuberconsul::{ServiceDiscovery, clone_compare, list_to_upstreams};
use crate::ingress::kubewatcher::{AralezRouteUpdate, start_ingress_watcher_with_config};
use crate::utils::tools::Die;
use rustls::crypto::CryptoProvider;
use rustls::crypto::ring::default_provider;
use tokio::sync::mpsc::Sender;
#[derive(Debug, serde::Deserialize)]
@@ -66,12 +67,12 @@ impl ServiceDiscovery for KubernetesDiscovery {
let end = servers.len().saturating_sub(1);
let num = if end > 0 { rand::rng().random_range(0..end) } else { 0 };
let server = servers.get(num).unwrap().to_string();
let server = servers.get(num).die("Getting k8s api server").to_string();
let path = kuber.tokenpath.unwrap_or_else(|| "/var/run/secrets/kubernetes.io/serviceaccount/token".to_string());
let token = kuberconsul::read_token(path.as_str()).await;
if CryptoProvider::get_default().is_none() {
default_provider().install_default().expect("Failed to install rustls crypto provider");
default_provider().install_default().die("Failed to install rustls crypto provider");
}
let (tx, mut rx) = tokio::sync::mpsc::channel::<AralezRouteUpdate>(100);
src/logging/core.rs
+9 -8
View File
@@ -1,13 +1,14 @@
use crate::logging::types::start_logging_backend;
use crate::utils::metrics::LOGGING_ERRORS;
use crate::utils::tools::Die;
use crate::utils::types::AppConfig;
use anyhow::Result;
use log::{error, info, log, Level, LevelFilter, Record};
use log::{Level, LevelFilter, Record, error, info, log};
use log4rs::append::Append;
use log4rs::append::rolling_file::RollingFileAppender;
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
use log4rs::append::rolling_file::policy::compound::roll::fixed_window::FixedWindowRoller;
use log4rs::append::rolling_file::policy::compound::trigger::size::SizeTrigger;
use log4rs::append::rolling_file::policy::compound::CompoundPolicy;
use log4rs::append::rolling_file::RollingFileAppender;
use log4rs::append::Append;
use log4rs::config::Logger;
use log4rs::{
append::console::ConsoleAppender,
@@ -149,7 +150,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
.logger(Logger::builder().build("h2", LevelFilter::Off));
}
let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).unwrap();
let config = config_builder.build(Root::builder().appender("stdout").build(log_level)).die("Logging config builder");
if let Err(e) = log4rs::init_config(config) {
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
@@ -177,7 +178,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
_ => format!("{}.{{}}", path),
};
let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).unwrap();
let roller = FixedWindowRoller::builder().build(pattern_str.as_str(), keep).die("Logging roller builder");
let trigger = SizeTrigger::new(size_mb * 1024 * 1024);
let policy = CompoundPolicy::new(Box::new(trigger), Box::new(roller));
@@ -193,7 +194,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
.appender(Appender::builder().build("file", Box::new(file)))
.logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
.build(Root::builder().appender("file").build(log_level))
.unwrap();
.die("Parsing Lo4rsConfig");
if let Err(e) = log4rs::init_config(config) {
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
}
@@ -204,7 +205,7 @@ pub async fn log_builder(conf: &AppConfig, location: &Option<String>) {
.appender(Appender::builder().build("stdout", Box::new(stdout)))
.logger(Logger::builder().build("pingora_proxy", LevelFilter::Off))
.build(Root::builder().appender("stdout").build(log_level))
.unwrap();
.die("Parsing Lo4rsConfig");
if let Err(e) = log4rs::init_config(config) {
eprintln!("Warning: log4rs logger already initialized: {:?}", e);
}
src/tls/acme/dns/cloudflare.rs
+3 -2
View File
@@ -1,5 +1,6 @@
use crate::tls::acme::lookup::lookup_wait;
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
use crate::utils::tools::Die;
use log::info;
use reqwest::Client;
use serde_json::json;
@@ -12,8 +13,8 @@ pub struct CloudflareProvider {
impl CloudflareProvider {
pub fn new() -> Self {
let api_token = std::env::var("CLOUDFLARE_API_TOKEN").expect("Environment variable CLOUDFLARE_API_TOKEN is required");
let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").expect("Environment variable CLOUDFLARE_ZONE_ID is required");
let api_token = std::env::var("CLOUDFLARE_API_TOKEN").die("Environment variable CLOUDFLARE_API_TOKEN is required");
let zone_id = std::env::var("CLOUDFLARE_ZONE_ID").die("Environment variable CLOUDFLARE_ZONE_ID is required");
Self {
client: Client::new(),
src/tls/acme/dns/digitalocean.rs
+2 -1
View File
@@ -1,5 +1,6 @@
use crate::tls::acme::lookup::lookup_wait;
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
use crate::utils::tools::Die;
use async_trait::async_trait;
use log::info;
use serde_json::json;
@@ -12,7 +13,7 @@ pub struct DigitalOceanProvider {
impl DigitalOceanProvider {
pub fn new() -> Self {
let token = env::var("DIGITALOCEAN_TOKEN").expect("Environment variable DIGITALOCEAN_TOKEN is required");
let token = env::var("DIGITALOCEAN_TOKEN").die("Environment variable DIGITALOCEAN_TOKEN is required");
Self {
token,
src/tls/acme/dns/route53.rs
+13 -6
View File
@@ -4,7 +4,7 @@
use crate::tls::acme::lookup::lookup_wait;
use crate::tls::acme::types::{DnsBackendPlugin, DnsProvider};
use aws_config::SdkConfig;
use crate::utils::tools::Die;
use aws_sdk_route53::Client;
use aws_sdk_route53::types::{Change, ChangeAction, ChangeBatch, ResourceRecord, ResourceRecordSet, RrType};
use log::info;
@@ -17,13 +17,20 @@ pub struct Route53Provider {
impl Route53Provider {
pub fn new() -> Self {
let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").expect("Environment variable AWS_HOSTED_ZONE_ID is required");
// Load AWS SDK config synchronously using Tokio runtime handle to fit factory construction
let config: SdkConfig = if let Ok(handle) = Handle::try_current() {
tokio::task::block_in_place(|| handle.block_on(aws_config::load_from_env()))
let hosted_zone_id = std::env::var("AWS_HOSTED_ZONE_ID").die("Environment variable AWS_HOSTED_ZONE_ID is required");
let config = if let Ok(handle) = Handle::try_current() {
// If we're inside a tokio runtime, we can block on it directly
handle.block_on(aws_config::load_from_env())
} else {
tokio::runtime::Runtime::new().unwrap().block_on(aws_config::load_from_env())
// If we're completely outside tokio, spin up a temporary current-thread runtime
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.die("Failed to create current-thread runtime")
.block_on(aws_config::load_from_env())
};
let client = Client::new(&config);
Self { client, hosted_zone_id }
}
src/tls/acme/http01.rs
+6 -17
View File
@@ -5,7 +5,6 @@ use axum::body::Body;
use axum::extract::State;
use axum::http::{Response, StatusCode};
use axum::response::IntoResponse;
#[allow(clippy::needless_return)]
pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoResponse {
match account::load_or_create(state.cert_creds.as_str()).await {
@@ -14,13 +13,10 @@ pub async fn acme_create(State(state): State<webserver::AppState>) -> impl IntoR
.status(StatusCode::OK)
.header("Content-Type", "text/plain")
.body(Body::from(txt))
.unwrap();
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
}
Err(e) => {
return Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(format!("Failed to create account: {}", e)))
.unwrap();
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Failed to create account {}\n", e)).into_response();
}
};
}
@@ -33,13 +29,10 @@ pub async fn acme_order(State(state): State<webserver::AppState>, axum::extract:
.status(StatusCode::OK)
.header("Content-Type", "text/plain")
.body(Body::from(txt))
.unwrap();
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
}
Err(e) => {
return Response::builder()
.status(StatusCode::INTERNAL_SERVER_ERROR)
.body(Body::from(format!("Failed to order a certificate: {}", e)))
.unwrap();
return (StatusCode::INTERNAL_SERVER_ERROR, format!("Internal Server Error: {}\n", e)).into_response();
}
};
}
@@ -50,12 +43,8 @@ pub async fn http01_challenge(axum::extract::Path(token): axum::extract::Path<St
.status(StatusCode::OK)
.header("Content-Type", "text/plain")
.body(Body::from(key_authorization.clone()))
.unwrap();
.unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response());
}
}
Response::builder()
.status(StatusCode::NOT_FOUND)
.header("Content-Type", "text/plain")
.body(Body::from("Not found"))
.unwrap()
(StatusCode::NOT_FOUND, "Not found\n").into_response()
}
src/tls/acme/lookup.rs
+2 -1
View File
@@ -1,3 +1,4 @@
use crate::utils::tools::Die;
use hickory_resolver::TokioResolver;
use hickory_resolver::config::{NameServerConfig, ResolverConfig, ResolverOpts};
use hickory_resolver::net::runtime::TokioRuntimeProvider;
@@ -17,7 +18,7 @@ pub async fn lookup_wait(record_name: &str, dns_server_ip: Option<&str>, max_wai
let resolver = TokioResolver::builder_with_config(config, TokioRuntimeProvider::default())
.with_options(opts)
.build()
.expect("failed to build DNS resolver");
.die("failed to build DNS resolver");
let start = std::time::Instant::now();
let poll_interval = Duration::from_secs(3);
src/tls/acme/order.rs
+4 -3
View File
@@ -2,6 +2,7 @@ use crate::tls::acme::account::get_account;
use crate::tls::acme::types::get_provider;
use crate::utils::lazylock::CHALLENGES;
use crate::utils::lazylock::DOMAINS;
use crate::utils::tools::Die;
use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use instant_acme::{ChallengeType, Identifier, NewOrder, RetryPolicy};
@@ -63,7 +64,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
let key_auth_str = key_auth.as_str().to_string();
if is_dns {
let prov = provider.as_ref().unwrap();
let prov = provider.as_ref().die("Failed to get DNS provider");
let dns_provider = get_provider(prov.clone());
let dns_value = calculate_dns_value(&key_auth_str);
let clean_domain = domain_name.trim_start_matches("*.").to_string();
@@ -82,7 +83,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
}
} else {
let token = key_auth_str.split('.').next().ok_or("Invalid key authorization")?.to_string();
CHALLENGES.write().unwrap().insert(token, key_auth_str);
CHALLENGES.write().die("Failed to write challenge").insert(token, key_auth_str);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
challenge_handle.set_ready().await?;
}
@@ -98,7 +99,7 @@ pub async fn order(domain: &str, credsfile: &str, certs_dir: String, provider: O
let csr_der = signing_request.der();
order.finalize_csr(csr_der).await?;
let cert_chain_pem = order.poll_certificate(&RetryPolicy::default()).await?;
CHALLENGES.write().unwrap().clear();
CHALLENGES.write().die("Challenges lock poisoned").clear();
let private_key_pem = private_key.serialize_pem();
fs::write(crt, cert_chain_pem)?;
fs::write(key, private_key_pem)?;
src/tls/load.rs
+4 -2
View File
@@ -1,8 +1,9 @@
use crate::tls::grades;
use crate::utils::tools::Die;
use dashmap::DashMap;
use log::error;
use pingora::tls::ssl::{NameType, SniError, SslAlert, SslContext, SslFiletype, SslMethod, SslRef};
use rustls_pemfile::{read_one, Item};
use rustls_pemfile::{Item, read_one};
use serde::Deserialize;
use std::collections::HashSet;
use std::fs::File;
@@ -10,6 +11,7 @@ use std::io::BufReader;
use x509_parser::extensions::GeneralName;
use x509_parser::nom::Err as NomErr;
use x509_parser::prelude::*;
#[derive(Clone, Deserialize, Debug)]
pub struct CertificateConfig {
pub cert_path: String,
@@ -37,7 +39,7 @@ pub struct Certificates {
impl Certificates {
pub fn new(configs: &Vec<CertificateConfig>, _grade: &str) -> Option<Self> {
let default_cert = configs.first().expect("At least one TLS certificate required");
let default_cert = configs.first().die("At least one TLS certificate required");
let mut cert_infos = Vec::new();
let name_map: DashMap<String, SslContext> = DashMap::new();
for config in configs {
src/utils/deser.rs
+4 -3
View File
@@ -1,5 +1,6 @@
use crate::utils::hcclient::{HttpClientRequest, httpclient};
use crate::utils::lazylock::{GLOBAL_PARAMS, MASTER_KEY, WORKERS};
use crate::utils::tools::Die;
use crate::utils::types::{Consul, Extraparams, Headers, Kubernetes, UpstreamsDashMap};
use bytes::Bytes;
use log::{error, info, warn};
@@ -44,7 +45,7 @@ pub struct SendConfigDto {
}
pub fn serialize_config(data: &UpstreamsDashMap) -> String {
let config = GLOBAL_PARAMS.read().unwrap();
let config = GLOBAL_PARAMS.read().die("Failed to read config");
let mut snapshot: HashMap<String, UpstreamConfigDto> = HashMap::new();
for outer_entry in data.iter() {
@@ -268,7 +269,7 @@ pub fn to_yaml(config: &SendConfigDto) -> String {
}
pub async fn send_to_worker(data: String) {
let masterkey = MASTER_KEY.get().expect("Master key not set");
let masterkey = MASTER_KEY.get().die("Master key not set");
if let Some(workers) = WORKERS.get() {
for worker in workers {
let prefix = if worker.tls { "https://" } else { "http://" };
@@ -282,7 +283,7 @@ pub async fn send_to_worker(data: String) {
port: worker.port,
url: worker_url.as_str(),
payload: Bytes::from(data.clone()),
headers: Some(vec![("API-KEY", masterkey.clone().expect("Master key must be set in the cluster mode").deref())]),
headers: Some(vec![("API-KEY", masterkey.clone().die("Master key must be set in the cluster mode").deref())]),
})
.await;
if (200..500).contains(&status.2) {
src/utils/grpc.rs
+10 -3
View File
@@ -1,17 +1,19 @@
use bytes::Bytes;
use http_body_util::Full;
use hyper::Request;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_util::client::legacy::Client;
use hyper_util::client::legacy::connect::HttpConnector;
use hyper_util::rt::TokioExecutor;
use log::error;
pub async fn ping_grpc(addr: &str) -> bool {
let connector = HttpConnector::new();
let client = Client::builder(TokioExecutor::new()).http2_only(true).build(connector);
let uri = format!("{}{}", addr, "/aralez.Probe/Ping");
let body = Full::new(Bytes::from(vec![0, 0, 0, 0, 0]));
let request = Request::post(uri).header("content-type", "application/grpc").header("te", "trailers").body(body).unwrap();
match client.request(request).await {
match Request::post(uri.clone()).header("content-type", "application/grpc").header("te", "trailers").body(body) {
Ok(request) => match client.request(request).await {
Ok(resp) => resp
.headers()
.get("content-type")
@@ -19,5 +21,10 @@ pub async fn ping_grpc(addr: &str) -> bool {
.map(|v| v.starts_with("application/grpc"))
.unwrap_or(false),
Err(_) => false,
},
Err(e) => {
error!("Failed to create request: {}", e);
false
}
}
}
src/utils/healthcheck.rs
+1 -1
View File
@@ -126,7 +126,7 @@ async fn detect_tls(ip: &str, port: &u16, sni: &str) -> bool {
.danger_accept_invalid_certs(true)
.danger_accept_invalid_hostnames(true)
.build()
.unwrap(),
.die("Failed to create TLS connector"),
);
connector.connect(sni, stream).await.is_ok()
}
src/utils/metrics.rs
+31 -18
View File
@@ -1,9 +1,10 @@
use crate::utils::lazylock::EVICTION;
use crate::utils::tools::Die;
use pingora_cache::eviction::EvictionManager;
use pingora_http::Method;
use pingora_http::StatusCode;
use pingora_http::Version;
use prometheus::{register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge, Histogram, IntCounter, IntCounterVec, IntGauge};
use prometheus::{Histogram, IntCounter, IntCounterVec, IntGauge, register_histogram, register_int_counter, register_int_counter_vec, register_int_gauge};
use std::sync::Arc;
use std::sync::LazyLock;
use std::time::Duration;
@@ -16,19 +17,28 @@ pub struct MetricTypes {
pub version: Version,
}
pub static OPEN_FILES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").unwrap());
pub static LOGGING_ERRORS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").unwrap());
pub static MEMORY_USAGE: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").unwrap());
pub static ACTIVE_SESSIONS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap());
pub static REQUEST_COUNT: LazyLock<IntCounter> = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap());
pub static OPEN_FILES: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").die("Failed to register OPEN_FILES metric"));
pub static LOGGING_ERRORS: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").die("Failed to register LOGGING_ERRORS metric"));
pub static MEMORY_USAGE: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").die("Failed to register MEMORY_USAGE metric"));
pub static ACTIVE_SESSIONS: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").die("Failed to register ACTIVE_SESSIONS metric"));
pub static REQUEST_COUNT: LazyLock<IntCounter> =
LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").die("Failed to register REQUEST_COUNT metric"));
pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").unwrap());
pub static CACHE_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").unwrap());
pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").unwrap());
pub static CACHE_SIZE_BYTES: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").die("Failed to register CACHE_SIZE_BYTES metric"));
pub static CACHE_ITEMS: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").die("Failed to register CACHE_ITEMS metric"));
pub static CACHE_EVICTED_BYTES: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").die("Failed to register CACHE_EVICTED_BYTES metric"));
pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").unwrap());
pub static CACHE_EVICTED_ITEMS: LazyLock<IntGauge> =
LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").die("Failed to register CACHE_EVICTED_ITEMS metric"));
pub static RESPONSE_CODES: LazyLock<IntCounterVec> =
LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).unwrap());
LazyLock::new(|| register_int_counter_vec!("aralez_responses_total", "Responses grouped by status code", &["status"]).die("Failed to register RESPONSE_CODES metric"));
pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
register_histogram!(
@@ -36,17 +46,20 @@ pub static RESPONSE_LATENCY: LazyLock<Histogram> = LazyLock::new(|| {
"Response latency in seconds",
vec![0.001, 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0]
)
.unwrap()
.die("Failed to register RESPONSE_LATENCY metric")
});
pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> =
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).unwrap());
pub static REQUESTS_BY_METHOD: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec!("aralez_requests_by_method_total", "Number of requests by HTTP method", &["method"]).die("Failed to register REQUESTS_BY_METHOD metric")
});
pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> =
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).unwrap());
pub static REQUESTS_BY_UPSTREAM: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).die("Failed to register REQUESTS_BY_UPSTREAM metric")
});
pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> =
LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).unwrap());
pub static REQUESTS_BY_VERSION: LazyLock<IntCounterVec> = LazyLock::new(|| {
register_int_counter_vec!("aralez_requests_by_version_total", "Number of requests by HTTP versions", &["version"]).die("Failed to register REQUESTS_BY_VERSION metric")
});
pub static HTTP_11_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/1.1"]));
pub static HTTP_20_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/2.0"]));
pub static HTTP_30_COUNT: LazyLock<IntCounter> = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/3.0"]));
src/utils/parceyaml.rs
+2 -2
View File
@@ -48,8 +48,8 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
return (None, err_msg);
}
let mut confdir = Path::new(d).parent().unwrap().to_path_buf();
let mut autocfg = Path::new(d).parent().unwrap().to_path_buf();
let mut confdir = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
let mut autocfg = Path::new(d).parent().die("Invalid configuration path").to_path_buf();
autocfg.push("autoconfigs");
if fs::metadata(autocfg.clone()).is_err() {
src/utils/state.rs
+3 -2
View File
@@ -1,3 +1,4 @@
use crate::utils::tools::Die;
use std::sync::{LazyLock, RwLock};
#[derive(Debug)]
@@ -8,11 +9,11 @@ pub struct SharedState {
pub static GLOBAL_STATE: LazyLock<RwLock<SharedState>> = LazyLock::new(|| RwLock::new(SharedState { first_run: true }));
pub fn mark_not_first_run() {
let mut state = GLOBAL_STATE.write().unwrap();
let mut state = GLOBAL_STATE.write().die("Writing global state failed");
state.first_run = false;
}
pub fn is_first_run() -> bool {
let state = GLOBAL_STATE.read().unwrap();
let state = GLOBAL_STATE.read().die("Writing global state failed");
state.first_run
}
src/utils/tools.rs
+14 -11
View File
@@ -14,6 +14,7 @@ use std::net::SocketAddr;
use std::net::TcpListener;
use std::os::unix::fs::MetadataExt;
use std::os::unix::fs::OpenOptionsExt;
use std::panic::Location;
use std::process::Command;
use std::str::FromStr;
use std::sync::Arc;
@@ -23,9 +24,9 @@ use std::{env, fs, process, thread, time};
pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams) {
let mut out = String::new();
for host_entry in upstreams.iter() {
writeln!(out, "Hostname: {}", host_entry.key()).unwrap();
writeln!(out, "Hostname: {}", host_entry.key()).unwrap_or_default();
for path_entry in host_entry.value().iter() {
writeln!(out, " Path: {}", path_entry.key()).unwrap();
writeln!(out, " Path: {}", path_entry.key()).unwrap_or_default();
for f in path_entry.value().0.clone() {
writeln!(
out,
@@ -38,7 +39,7 @@ pub fn print_upstreams(upstreams: &UpstreamsDashMap, extraparams: &Extraparams)
if f.rate_limit > 0 { f.rate_limit } else { extraparams.rate_limit },
if f.x4xx_limit > 0 { f.x4xx_limit } else { extraparams.x4xx_limit }
)
.unwrap();
.unwrap_or_default();
}
}
}
@@ -161,9 +162,9 @@ pub fn clone_idmap_into(original: &UpstreamsDashMap, cloned: &UpstreamsIdMap) {
pub fn listdir(dir: String) -> Vec<load::CertificateConfig> {
let mut f = HashMap::new();
let mut certificate_configs: Vec<load::CertificateConfig> = vec![];
let paths = fs::read_dir(dir).unwrap();
let paths = fs::read_dir(dir).die("listing certificates directory failed");
for path in paths {
let path_str = path.unwrap().path().to_str().unwrap().to_owned();
let path_str = path.die("getting certs path").path().to_str().die("certs pat t &str").to_owned();
if path_str.ends_with(".crt") {
let name = path_str.replace(".crt", "");
let mut inner = vec![];
@@ -212,9 +213,9 @@ pub fn drop_priv(user: String, group: String, http_addr: String, tls_addr: Optio
}
pub fn check_priv(addr: &str) {
let port = SocketAddr::from_str(addr).map(|sa| sa.port()).expect("Failed to parse address port ");
let port = SocketAddr::from_str(addr).map(|sa| sa.port()).die("Failed to parse address port ");
if port < 1024 {
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).unwrap();
let meta = std::fs::metadata("/proc/self").map(|m| m.uid()).die("Reading /proc/self");
if meta != 0 {
error!("Running on privileged port requires to start as ROOT");
process::exit(1)
@@ -369,18 +370,19 @@ pub fn get_hostname() -> String {
// log::error!("{message}");
// process::exit(1);
// }
// ========================================= //
pub trait Die<T> {
fn die(self, message: &str) -> T;
}
impl<T, E: Debug> Die<T> for Result<T, E> {
#[track_caller]
fn die(self, message: &str) -> T {
match self {
Ok(value) => value,
Err(e) => {
eprintln!("\x1b[1;31mERROR:\x1b[0m {message}: {e:?}");
let loc = Location::caller();
eprintln!("\x1b[1;31mERROR:\x1b[0m {message}: {e:?}: {}: {}", loc.file(), loc.line());
process::exit(1);
}
}
@@ -388,14 +390,15 @@ impl<T, E: Debug> Die<T> for Result<T, E> {
}
impl<T> Die<T> for Option<T> {
#[track_caller]
fn die(self, message: &str) -> T {
match self {
Some(value) => value,
None => {
eprintln!("\x1b[1;31mERROR:\x1b[0m {message}");
let loc = Location::caller();
eprintln!("\x1b[1;31mERROR:\x1b[0m {message} at {}:{}", loc.file(), loc.line());
process::exit(1);
}
}
}
}
// ========================================= //
src/utils/watch.rs
+5 -5
View File
@@ -1,6 +1,6 @@
use crate::tls::CertificateConfig;
use crate::utils::parceyaml::load_configuration;
use crate::utils::tools::listdir;
use crate::utils::tools::{Die, listdir};
use crate::utils::types::Configuration;
use log::{error, info};
use notify::event::ModifyKind;
@@ -15,7 +15,7 @@ use tokio::task;
pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
sleep(Duration::from_millis(50)).await; // For having nice logs :-)
let file_path = fp.as_str();
let parent_dir = Path::new(file_path).parent().unwrap();
let parent_dir = Path::new(file_path).parent().die("Get parent folder");
let (local_tx, mut local_rx) = tokio::sync::mpsc::channel::<notify::Result<Event>>(1);
let _watcher_handle = task::spawn_blocking({
@@ -27,8 +27,8 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
},
Config::default(),
)
.unwrap();
watcher.watch(&parent_dir, RecursiveMode::Recursive).unwrap();
.die("Initialize file watcher");
watcher.watch(&parent_dir, RecursiveMode::Recursive).die("");
let (_rtx, mut rrx) = tokio::sync::mpsc::channel::<bool>(1);
let _ = rrx.blocking_recv();
}
@@ -42,7 +42,7 @@ pub async fn file_watch(fp: String, toreturn: Sender<Configuration>) {
start = Instant::now();
let snd = load_configuration(file_path, "filepath").await.0;
if let Some(snd) = snd {
toreturn.send(snd).await.unwrap();
toreturn.send(snd).await.die("Send to file watcher");
}
}
_ => (),