feat: add basic loggers

Author: jjeffcaii

capybara-bin/Cargo.toml

@@ -31,7 +31,6 @@ serde_yaml = "0.9.27"
 once_cell = "1.19.0"
 async-trait = "0.1.74"
 ahash = "0.8.11"
-pretty_env_logger = "0.5.0"
 dirs = "5.0.1"
 hashbrown = { version = "0.15.2", features = ["serde"] }
 duration-str = "0.11.2"

capybara-bin/src/bootstrap/config.rs

@@ -1,13 +1,14 @@
 use std::collections::HashMap;
 
+use capybara_core::logger::Config as LoggerConfig;
 use serde::{Deserialize, Serialize};
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BootstrapConf {
     #[serde(default)]
     pub resolvers: HashMap<String, ResolverConf>,
     #[serde(default)]
-    pub loggers: HashMap<String, LoggerConf>,
+    pub loggers: HashMap<String, LoggerConfig>,
     #[serde(default)]
     pub providers: Vec<ProviderConf>,
 }
@@ -18,11 +19,6 @@ pub struct ResolverConf {
     pub props: HashMap<String, serde_yaml::Value>,
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct LoggerConf {
-    pub path: String,
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ProviderConf {
     pub kind: String,

capybara-bin/src/bootstrap/runtime.rs

@@ -1,18 +1,19 @@
-use std::net::{IpAddr, SocketAddr};
+use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 
 use hashbrown::HashMap;
 use tokio::sync::{mpsc, Notify, RwLock};
 
-use capybara_core::proto::{Listener, Signal};
+use capybara_core::logger;
+use capybara_core::proto::{Addr, Listener, Signal};
 use capybara_core::protocol::http::HttpListener;
 use capybara_core::protocol::stream::StreamListener;
 use capybara_core::transport::tcp::TcpStreamPoolBuilder;
 use capybara_core::transport::tls::TlsStreamPoolBuilder;
 use capybara_core::transport::TlsAcceptorBuilder;
-use capybara_core::{CapybaraError, Pool, Pools, RoundRobinPools, WeightedPools};
+use capybara_core::{Pool, Pools, RoundRobinPools, WeightedPools};
 use capybara_etc::{
     BalanceStrategy, Config, ListenerConfig, Properties, TransportKind, UpstreamConfig,
 };
@@ -66,15 +67,15 @@ impl Dispatcher {
 
                     let p = match endpoint.transport.as_ref().unwrap_or(&v.transport) {
                         TransportKind::Tcp => {
+                            let addr = endpoint.addr.parse::<Addr>()?;
+                            let closer = Clone::clone(&self.closer);
                             if endpoint
                                 .tls
                                 .unwrap_or_else(|| endpoint.addr.ends_with(":443"))
                             {
-                                let b = to_tls_stream_pool_builder(&endpoint.addr)?;
-                                Pool::Tls(b.build(Clone::clone(&self.closer)).await?)
+                                Pool::Tls(TlsStreamPoolBuilder::new(addr).build(closer).await?)
                             } else {
-                                let b = to_tcp_stream_pool_builder(&endpoint.addr)?;
-                                Pool::Tcp(b.build(Clone::clone(&self.closer)).await?)
+                                Pool::Tcp(TcpStreamPoolBuilder::new(addr).build(closer).await?)
                             }
                         }
                         TransportKind::Udp => {
@@ -92,17 +93,15 @@ impl Dispatcher {
                 for endpoint in &v.endpoints {
                     let pool = match endpoint.transport.unwrap_or(v.transport) {
                         TransportKind::Tcp => {
+                            let addr = endpoint.addr.parse::<Addr>()?;
+                            let closer = Clone::clone(&self.closer);
                             if endpoint
                                 .tls
                                 .unwrap_or_else(|| endpoint.addr.ends_with(":443"))
                             {
-                                let b = to_tls_stream_pool_builder(&endpoint.addr)?;
-                                let p = b.build(Clone::clone(&self.closer)).await?;
-                                Pool::Tls(p)
+                                Pool::Tls(TlsStreamPoolBuilder::new(addr).build(closer).await?)
                             } else {
-                                let b = to_tcp_stream_pool_builder(&endpoint.addr)?;
-                                let p = b.build(Clone::clone(&self.closer)).await?;
-                                Pool::Tcp(p)
+                                Pool::Tcp(TcpStreamPoolBuilder::new(addr).build(closer).await?)
                             }
                         }
                         TransportKind::Udp => {
@@ -254,48 +253,11 @@ impl Dispatcher {
     }
 }
 
-#[inline]
-fn to_tls_stream_pool_builder(addr: &str) -> anyhow::Result<TlsStreamPoolBuilder> {
-    let mut sp = addr.split(':');
-    if let Some(left) = sp.next() {
-        if let Some(right) = sp.next() {
-            if let Ok(port) = right.parse::<u16>() {
-                if sp.next().is_none() {
-                    return Ok(match left.parse::<IpAddr>() {
-                        Ok(ip) => TlsStreamPoolBuilder::with_addr(SocketAddr::new(ip, port)),
-                        Err(_) => TlsStreamPoolBuilder::with_domain(left, port),
-                    });
-                }
-            }
-        }
-    }
-
-    bail!(CapybaraError::InvalidUpstream(addr.to_string().into()));
-}
-
-#[inline]
-fn to_tcp_stream_pool_builder(addr: &str) -> anyhow::Result<TcpStreamPoolBuilder> {
-    let mut sp = addr.split(':');
-    if let Some(left) = sp.next() {
-        if let Some(right) = sp.next() {
-            if let Ok(port) = right.parse::<u16>() {
-                if sp.next().is_none() {
-                    return Ok(match left.parse::<IpAddr>() {
-                        Ok(ip) => TcpStreamPoolBuilder::with_addr(SocketAddr::new(ip, port)),
-                        Err(_) => TcpStreamPoolBuilder::with_domain(left, port),
-                    });
-                }
-            }
-        }
-    }
-
-    bail!(CapybaraError::InvalidUpstream(addr.to_string().into()));
-}
-
 pub(crate) struct Bootstrap {
     bc: BootstrapConf,
     c: Arc<RwLock<Config>>,
     dispatcher: Arc<RwLock<Dispatcher>>,
+    loggers: Arc<RwLock<HashMap<String, logger::Key>>>,
 }
 
 impl Bootstrap {
@@ -304,9 +266,37 @@ impl Bootstrap {
         w.shutdown().await;
     }
 
+    async fn generate_loggers(&self) -> anyhow::Result<()> {
+        let mut loggers: HashMap<String, logger::Key> = Default::default();
+        let mut has_main = false;
+        for (k, v) in &self.bc.loggers {
+            if k == "main" {
+                has_main = true;
+                logger::init_global(v)?;
+            } else {
+                let lk = logger::register(v)?;
+                loggers.insert(Clone::clone(k), lk);
+            }
+        }
+
+        // use default logger settings if no main logger found
+        if !has_main {
+            logger::init_global(&logger::Config::default())?;
+        }
+
+        // extend loggers
+        {
+            let mut w = self.loggers.write().await;
+            w.extend(loggers);
+        }
+
+        Ok(())
+    }
+
     pub(crate) async fn start(&self) -> anyhow::Result<()> {
-        let (c_tx, mut c_rx) = mpsc::unbounded_channel::<Config>();
+        self.generate_loggers().await?;
 
+        let (c_tx, mut c_rx) = mpsc::unbounded_channel::<Config>();
         {
             let dispatcher = Clone::clone(&self.dispatcher);
             let c = Clone::clone(&self.c);
@@ -428,6 +418,7 @@ impl From<BootstrapConf> for Bootstrap {
             bc: value,
             c: Default::default(),
             dispatcher: Default::default(),
+            loggers: Default::default(),
         }
     }
 }

capybara-bin/src/main.rs

@@ -50,14 +50,8 @@ enum Commands {
     },
 }
 
-fn init() {
-    pretty_env_logger::try_init_timed().ok();
-}
-
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    init();
-
     let cli = Cli::parse();
 
     let shutdown = Arc::new(Notify::new());

capybara-core/Cargo.toml

@@ -73,6 +73,7 @@ liquid = "0.26"
 rustls = "0.23"
 webpki-roots = "0.26"
 tokio-rustls = "0.26"
+dirs = "5.0"
 
 [[example]]
 name = "httpbin"

capybara-core/src/error.rs

@@ -1,6 +1,5 @@
 use std::borrow::Cow;
 use std::io;
-use tokio::time::error::Elapsed;
 
 #[derive(thiserror::Error, Debug)]
 pub enum CapybaraError {
@@ -32,6 +31,9 @@ pub enum CapybaraError {
     #[error("invalid route")]
     InvalidRoute,
 
+    #[error("invalid address '{0}'")]
+    InvalidAddress(/* address */ Cow<'static, str>),
+
     #[error("invalid upstream pool")]
     InvalidUpstreamPool,
 

capybara-core/src/lib.rs

@@ -26,7 +26,7 @@ pub type Result<T> = std::result::Result<T, CapybaraError>;
 
 mod builtin;
 mod error;
-mod logger;
+pub mod logger;
 mod macros;
 pub mod pipeline;
 pub mod proto;

capybara-core/src/logger.rs

@@ -1,5 +1,5 @@
 use std::io::{self, Write};
-use std::path::Path;
+use std::path::PathBuf;
 use std::sync::atomic::{AtomicU32, Ordering};
 
 use bytesize::ByteSize;
@@ -104,7 +104,7 @@ impl Default for Config {
             level: Default::default(),
             stdout: true,
             file: None,
-            line_num: false,
+            line_num: true,
             noh: false,
             chan_size: None,
             overflow: None,
@@ -222,8 +222,16 @@ fn new_logger(c: &Config) -> anyhow::Result<Option<Logger>> {
 
             match &c.file {
                 Some(fc) => {
-                    let path = Path::new(&fc.path);
-
+                    // normalize path
+                    let path = match &fc.path.strip_prefix("~/") {
+                        Some(rel) => {
+                            let pb = PathBuf::from(rel);
+                            dirs::home_dir().unwrap_or_default().join(pb)
+                        }
+                        None => PathBuf::from(&fc.path),
+                    };
+
+                    // ensure logger dir
                     if let Some(dir) = path.parent() {
                         std::fs::create_dir_all(dir)?;
                     }