From 0c489b649a7ea873af7fd4e0222bad5753a659b7 Mon Sep 17 00:00:00 2001 From: Ara Sadoyan Date: Tue, 20 Oct 2026 17:06:55 +0200 Subject: [PATCH] Improved error handling, anti unwrap()-ing --- src/core/bgservice.rs | 11 +++-- src/core/start.rs | 2 +- src/core/webserver.rs | 72 ++++++++++---------------------- src/ingress/consul.rs | 5 ++- src/ingress/kuberconsul.rs | 8 ++-- src/ingress/kubernetes.rs | 11 ++--- src/logging/core.rs | 17 ++++---- src/tls/acme/dns/cloudflare.rs | 5 ++- src/tls/acme/dns/digitalocean.rs | 3 +- src/tls/acme/dns/route53.rs | 19 ++++++--- src/tls/acme/http01.rs | 23 +++------- src/tls/acme/lookup.rs | 3 +- src/tls/acme/order.rs | 7 ++-- src/tls/load.rs | 6 ++- src/utils/deser.rs | 7 ++-- src/utils/grpc.rs | 27 +++++++----- src/utils/healthcheck.rs | 2 +- src/utils/metrics.rs | 49 ++++++++++++++-------- src/utils/parceyaml.rs | 4 +- src/utils/state.rs | 5 ++- src/utils/tools.rs | 25 ++++++----- src/utils/watch.rs | 10 ++--- 22 files changed, 162 insertions(+), 159 deletions(-) diff --git a/src/core/bgservice.rs b/src/core/bgservice.rs index 1bacc7f..60392e8 100644 --- a/src/core/bgservice.rs +++ b/src/core/bgservice.rs @@ -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()); } diff --git a/src/core/start.rs b/src/core/start.rs index b95b9c0..1a00e32 100644 --- a/src/core/start.rs +++ b/src/core/start.rs @@ -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)); diff --git a/src/core/webserver.rs b/src/core/webserver.rs index 03a0dc5..82d1923 100644 --- a/src/core/webserver.rs +++ b/src/core/webserver.rs @@ -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::()).await.unwrap() }); - + let api_server = tokio::spawn(async move { + axum::serve(listener, app.into_make_service_with_connect_info::()) + .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, Query(params): Query>, 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, Query(params): Query { 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, ConnectInfo(addr): ConnectInfo, ConnectInfo(addr): ConnectInfo { 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, Query(params): Query, Query(params): Query { - return Response::builder() - .status(StatusCode::INTERNAL_SERVER_ERROR) - .body(Body::from(format!("Failed to get status: {}", e))) - .unwrap(); + .unwrap_or_else(|_| (StatusCode::INTERNAL_SERVER_ERROR, "Internal Server Error\n").into_response()); } + 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 { diff --git a/src/ingress/consul.rs b/src/ingress/consul.rs index 025221a..f5605e7 100644 --- a/src/ingress/consul.rs +++ b/src/ingress/consul.rs @@ -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); diff --git a/src/ingress/kuberconsul.rs b/src/ingress/kuberconsul.rs index 69a0e8b..06796d6 100644 --- a/src/ingress/kuberconsul.rs +++ b/src/ingress/kuberconsul.rs @@ -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, (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() } diff --git a/src/ingress/kubernetes.rs b/src/ingress/kubernetes.rs index 3359fe5..fed8784 100644 --- a/src/ingress/kubernetes.rs +++ b/src/ingress/kubernetes.rs @@ -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::(100); diff --git a/src/logging/core.rs b/src/logging/core.rs index 2b4413a..ee82c26 100644 --- a/src/logging/core.rs +++ b/src/logging/core.rs @@ -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) { .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) { _ => 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) { .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) { .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); } diff --git a/src/tls/acme/dns/cloudflare.rs b/src/tls/acme/dns/cloudflare.rs index 7236de4..e6dfc59 100644 --- a/src/tls/acme/dns/cloudflare.rs +++ b/src/tls/acme/dns/cloudflare.rs @@ -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(), diff --git a/src/tls/acme/dns/digitalocean.rs b/src/tls/acme/dns/digitalocean.rs index 0f173b0..333b77d 100644 --- a/src/tls/acme/dns/digitalocean.rs +++ b/src/tls/acme/dns/digitalocean.rs @@ -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, diff --git a/src/tls/acme/dns/route53.rs b/src/tls/acme/dns/route53.rs index 134a132..e4b4a2d 100644 --- a/src/tls/acme/dns/route53.rs +++ b/src/tls/acme/dns/route53.rs @@ -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 } } diff --git a/src/tls/acme/http01.rs b/src/tls/acme/http01.rs index 26f482c..df25855 100644 --- a/src/tls/acme/http01.rs +++ b/src/tls/acme/http01.rs @@ -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) -> 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) -> 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, 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, 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); diff --git a/src/tls/acme/order.rs b/src/tls/acme/order.rs index 36718c8..70e7c09 100644 --- a/src/tls/acme/order.rs +++ b/src/tls/acme/order.rs @@ -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)?; diff --git a/src/tls/load.rs b/src/tls/load.rs index 5672204..20d7f86 100644 --- a/src/tls/load.rs +++ b/src/tls/load.rs @@ -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, _grade: &str) -> Option { - 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 = DashMap::new(); for config in configs { diff --git a/src/utils/deser.rs b/src/utils/deser.rs index 31d8cd5..eaccd89 100644 --- a/src/utils/deser.rs +++ b/src/utils/deser.rs @@ -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 = 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) { diff --git a/src/utils/grpc.rs b/src/utils/grpc.rs index e3cb7f9..93a45a2 100644 --- a/src/utils/grpc.rs +++ b/src/utils/grpc.rs @@ -1,23 +1,30 @@ 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 { - Ok(resp) => resp - .headers() - .get("content-type") - .and_then(|v| v.to_str().ok()) - .map(|v| v.starts_with("application/grpc")) - .unwrap_or(false), - Err(_) => false, + + 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") + .and_then(|v| v.to_str().ok()) + .map(|v| v.starts_with("application/grpc")) + .unwrap_or(false), + Err(_) => false, + }, + Err(e) => { + error!("Failed to create request: {}", e); + false + } } } diff --git a/src/utils/healthcheck.rs b/src/utils/healthcheck.rs index ae2e75f..4db090d 100644 --- a/src/utils/healthcheck.rs +++ b/src/utils/healthcheck.rs @@ -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() } diff --git a/src/utils/metrics.rs b/src/utils/metrics.rs index db68e97..987d05b 100644 --- a/src/utils/metrics.rs +++ b/src/utils/metrics.rs @@ -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 = LazyLock::new(|| register_int_gauge!("aralez_open_files", "Number of open file descriptors").unwrap()); -pub static LOGGING_ERRORS: LazyLock = LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").unwrap()); -pub static MEMORY_USAGE: LazyLock = LazyLock::new(|| register_int_gauge!("aralez_memory_bytes", "Total memory allocated in bytes").unwrap()); -pub static ACTIVE_SESSIONS: LazyLock = LazyLock::new(|| register_int_gauge!("aralez_active_sessions", "Current number of active sessions").unwrap()); -pub static REQUEST_COUNT: LazyLock = LazyLock::new(|| register_int_counter!("aralez_requests_total", "Total number of requests handled by Aralez").unwrap()); +pub static OPEN_FILES: LazyLock = + 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 = + LazyLock::new(|| register_int_gauge!("aralez_logging_errors", "Number of log errors").die("Failed to register LOGGING_ERRORS metric")); +pub static MEMORY_USAGE: LazyLock = + 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 = + 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 = + 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 = LazyLock::new(|| register_int_gauge!("aralez_cache_size_bytes", "Current cache size in bytes").unwrap()); -pub static CACHE_ITEMS: LazyLock = LazyLock::new(|| register_int_gauge!("aralez_cache_items", "Current number of cached objects").unwrap()); -pub static CACHE_EVICTED_BYTES: LazyLock = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_bytes_total", "Total bytes evicted from cache").unwrap()); +pub static CACHE_SIZE_BYTES: LazyLock = + 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 = + 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 = + 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 = LazyLock::new(|| register_int_gauge!("aralez_cache_evicted_items_total", "Total cache items evicted").unwrap()); +pub static CACHE_EVICTED_ITEMS: LazyLock = + 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 = - 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 = LazyLock::new(|| { register_histogram!( @@ -36,17 +46,20 @@ pub static RESPONSE_LATENCY: LazyLock = 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 = - 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 = 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 = - LazyLock::new(|| register_int_counter_vec!("aralez_requests_by_upstream", "Number of requests by UPSTREAM server", &["upstream"]).unwrap()); +pub static REQUESTS_BY_UPSTREAM: LazyLock = 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 = - 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 = 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 = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/1.1"])); pub static HTTP_20_COUNT: LazyLock = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/2.0"])); pub static HTTP_30_COUNT: LazyLock = LazyLock::new(|| REQUESTS_BY_VERSION.with_label_values(&["HTTP/3.0"])); diff --git a/src/utils/parceyaml.rs b/src/utils/parceyaml.rs index 73bfdc6..fabb3d5 100644 --- a/src/utils/parceyaml.rs +++ b/src/utils/parceyaml.rs @@ -48,8 +48,8 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option, 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() { diff --git a/src/utils/state.rs b/src/utils/state.rs index da0bf73..c3b6dcc 100644 --- a/src/utils/state.rs +++ b/src/utils/state.rs @@ -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> = 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 } diff --git a/src/utils/tools.rs b/src/utils/tools.rs index 8bca645..ab4ffc9 100644 --- a/src/utils/tools.rs +++ b/src/utils/tools.rs @@ -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 { let mut f = HashMap::new(); let mut certificate_configs: Vec = 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 { fn die(self, message: &str) -> T; } impl Die for Result { + #[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 Die for Result { } impl Die for Option { + #[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); } } } } -// ========================================= // diff --git a/src/utils/watch.rs b/src/utils/watch.rs index b10950a..6df6840 100644 --- a/src/utils/watch.rs +++ b/src/utils/watch.rs @@ -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) { 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::>(1); let _watcher_handle = task::spawn_blocking({ @@ -27,8 +27,8 @@ pub async fn file_watch(fp: String, toreturn: Sender) { }, 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::(1); let _ = rrx.blocking_recv(); } @@ -42,7 +42,7 @@ pub async fn file_watch(fp: String, toreturn: Sender) { 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"); } } _ => (),