Update to Pingora 0.9, improved logging, Fix cache key bug.

src/utils/parceyaml.rs

@@ -91,7 +91,7 @@ pub async fn load_configuration(d: &str, kind: &str) -> (Option<Configuration>,
             }
 
             info!("Reading upstreams from {}", d);
-            data // [2606:4700:2ff9::1]:443
+            data
         }
         "content" => {
             info!("Reading upstreams from API post body");

src/web/logging.rs

@@ -1,10 +1,11 @@
 use crate::utils::metrics::LOGGING_ERRORS;
 use crate::utils::structs::AppConfig;
-use log::{info, LevelFilter};
+use log::{error, info, warn, LevelFilter};
 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::config::Logger;
 use log4rs::{
     append::console::ConsoleAppender,
     config::{Appender, Config as Log4rsConfig, Root},
@@ -15,7 +16,7 @@ use pingora_http::Version;
 use pingora_proxy::Session;
 use std::net::{IpAddr, Ipv4Addr};
 use std::sync::OnceLock;
-use tokio::sync::mpsc;
+use tokio::sync::mpsc; // Add Logger import
 
 #[derive(Debug)]
 pub struct LogMessage {
@@ -74,6 +75,7 @@ pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
 
         let config = Log4rsConfig::builder()
             .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();
         log4rs::init_config(config).unwrap();
@@ -83,6 +85,7 @@ pub fn log_builder(conf: &AppConfig, location: &Option<String>) {
 
         let config = Log4rsConfig::builder()
             .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();
         log4rs::init_config(config).unwrap();
@@ -112,13 +115,30 @@ impl LogLevel {
     }
 }
 
+#[derive(Debug)]
+pub enum MatchStatus {
+    Ok2xx,
+    Er4xx,
+    Er5xx,
+}
+impl MatchStatus {
+    #[inline]
+    pub fn from_code(code: u16) -> Self {
+        match code {
+            100..=399 => Self::Ok2xx,
+            400..=499 => Self::Er4xx,
+            _ => Self::Er5xx,
+        }
+    }
+}
 pub fn access_log(response_code: u16, summary: &str, session: &Session) {
     let level = ACCESS_LOG.get().unwrap_or(&LogLevel::None);
-
+    let status = MatchStatus::from_code(response_code);
     let should_log = match level {
         LogLevel::Access => true,
         LogLevel::None => false,
-        LogLevel::Error => !(100..=399).contains(&response_code),
+        LogLevel::Error => matches!(status, MatchStatus::Er5xx),
+        // LogLevel::Error => matches!(status, MatchStatus::Er4xx | MatchStatus::Er5xx),
     };
 
     if !should_log {
@@ -132,12 +152,6 @@ pub fn access_log(response_code: u16, summary: &str, session: &Session) {
         .unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST));
 
     let user_agent = session.req_header().headers.get("user-agent").and_then(|v| v.to_str().ok()).unwrap_or("-");
-    // let cache_status = match session.cache.phase() {
-    //     CachePhase::Hit => "Hit",
-    //     CachePhase::Miss => "Miss",
-    //     CachePhase::Expired => "Expired",
-    //     _ => "Disabled",
-    // };
     let log = LogMessage {
         response_code,
         summary: summary.to_owned(),
@@ -167,14 +181,34 @@ pub fn init_logging(enabled: Option<String>) {
 
 pub fn log_receiver(mut receiver: mpsc::Receiver<LogMessage>) {
     while let Some(msg) = receiver.blocking_recv() {
-        info!(
-            "{}, {}, {}, client: {}, version: {:?}, useragent: {}",
-            msg.response_code,
-            msg.cache_status.as_str(),
-            msg.summary,
-            msg.client_ip,
-            msg.version,
-            msg.user_agent,
-        );
+        match MatchStatus::from_code(msg.response_code) {
+            MatchStatus::Ok2xx => info!(
+                "{}, {}, {}, client: {}, version: {:?}, useragent: {}",
+                msg.response_code,
+                msg.cache_status.as_str(),
+                msg.summary,
+                msg.client_ip,
+                msg.version,
+                msg.user_agent,
+            ),
+            MatchStatus::Er4xx => warn!(
+                "{}, {}, {}, client: {}, version: {:?}, useragent: {}",
+                msg.response_code,
+                msg.cache_status.as_str(),
+                msg.summary,
+                msg.client_ip,
+                msg.version,
+                msg.user_agent,
+            ),
+            MatchStatus::Er5xx => error!(
+                "{}, {}, {}, client: {}, version: {:?}, useragent: {}",
+                msg.response_code,
+                msg.cache_status.as_str(),
+                msg.summary,
+                msg.client_ip,
+                msg.version,
+                msg.user_agent,
+            ),
+        }
     }
 }

src/web/proxyhttp.rs

@@ -46,6 +46,9 @@ pub struct Context {
     x4xx_limit: Option<u32>,
 }
 
+thread_local! {
+    static KEY_BUF: RefCell<String> = RefCell::new(String::with_capacity(256));
+}
 #[async_trait]
 impl ProxyHttp for LB {
     type CTX = Context;
@@ -61,9 +64,22 @@ impl ProxyHttp for LB {
         }
     }
 
-    fn cache_key_callback(&self, session: &Session, _ctx: &mut Self::CTX) -> Result<CacheKey> {
-        Ok(CacheKey::new(_ctx.hostname.clone().unwrap_or_default().as_ref(), session.req_header().uri.to_string(), ""))
+    fn cache_key_callback(&self, session: &Session, ctx: &mut Self::CTX) -> Result<CacheKey> {
+        let host = ctx.hostname.as_deref().unwrap_or("default");
+        let path = session.req_header().uri.path();
+
+        let primary_key = KEY_BUF.with(|buf| {
+            let mut b = buf.borrow_mut();
+            b.clear();
+            b.push_str(host);
+            b.push_str(path);
+            b.clone()
+        });
+
+        let secondary_key = session.req_header().uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
+        Ok(CacheKey::new(primary_key, secondary_key))
     }
+
     fn request_cache_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<()> {
         if self.cache_enabled && session.req_header().method == Method::GET {
             if let Some(eviction) = EVICTION.get() {
@@ -339,13 +355,6 @@ impl ProxyHttp for LB {
                 }
             }
         }
-        // match session.cache.phase() {
-        //     CachePhase::Hit => info!("Cache Status: HIT (Served from Memory)"),
-        //     CachePhase::Miss => info!("Cache Status: MISS (Fetched from Upstream)"),
-        //     CachePhase::Expired => info!("Cache Status: EXPIRED (Revalidating)"),
-        //     _ => {}
-        // }
-
         access_log(response_code, &self.request_summary(session, ctx), session);
     }
 }