feat: component debug feature

Cargo.toml

@@ -31,6 +31,7 @@ hyper-rustls = { version = "0.27.2", features = ["http2", "ring"] }
 hyper-util = { version = "0.1.5", features = ["full"] }
 ipnetwork = "0.20.0"
 json_comments = "0.2.2"
+json_pretty = "0.1.2"
 libflate = "2.1.0"
 log = "0.4"
 openssl = "0.10"

src/config/config.rs

@@ -2,6 +2,7 @@ use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::sync::OnceLock;
 
+use crate::Options;
 use serde::Deserialize;
 use tracing::level_filters::LevelFilter;
 
@@ -91,13 +92,15 @@ pub struct LogConfiguration {
     #[serde_as(as = "serde_with::DisplayFromStr")]
     pub level: LevelFilter,
     pub span: Option<String>,
+    pub debug_component: Option<String>,
 }
 
 impl Default for LogConfiguration {
     fn default() -> Self {
         Self {
             level: LevelFilter::INFO,
             span: None,
+            debug_component: None,
         }
     }
 }
@@ -258,12 +261,15 @@ fn read_config(path: Option<&Path>) -> Result<StaticConfiguration, String> {
     }
 }
 
-// TODO: Add more configuration validations
-// TODO: Improve error messages for configuration errors
-pub fn init(path: Option<&Path>) {
-    let config = read_config(path).expect("should read config file");
+pub fn init(options: &Options) {
+    let path = options.config_path.as_deref();
+    let mut config = read_config(path).expect("should read config file");
     config.validate().unwrap();
 
+    if let Some(component) = options.debug_component.as_deref() {
+        config.log.debug_component = Some(component.to_string());
+    }
+
     CONFIG.set(config).expect("Should initialize config");
 }
 

src/logger.rs

@@ -13,13 +13,14 @@ pub enum LogFormat {
 /// Initializing logging facilities
 ///
 /// Log filtering is configured with env var and config in this priority order:
+/// - `log_filter` parameter
 /// - `EDGEE_LOG` env variable, formatted for [tracing_subscriber::EnvFilter]
 /// - `RUST_LOG` "standard" env variable, also formatted for [tracing_subscriber::EnvFilter]
 /// - Config file, specifying level and optionally span
 ///
 /// In the case something goes wrong with parsing of these directives, logging is done
 /// using the log level defined in config
-pub fn init(log_format: LogFormat) {
+pub fn init(log_format: LogFormat, log_filter: Option<String>) {
     use std::env;
 
     use tracing_subscriber::prelude::*;
@@ -29,7 +30,9 @@ pub fn init(log_format: LogFormat) {
 
     let config = &config::get().log;
 
-    let fmt_layer = fmt::layer();
+    let with_target = log_filter.is_none();
+    let fmt_layer = fmt::layer().with_target(with_target);
+
     let fmt_layer = match log_format {
         LogFormat::Basic | LogFormat::Pretty => fmt_layer.boxed(),
         LogFormat::Json => fmt_layer.json().boxed(),
@@ -38,10 +41,11 @@ pub fn init(log_format: LogFormat) {
     let filter_layer = {
         let builder = EnvFilter::builder().with_default_directive(config.level.into());
 
-        // Get logging directives from EDGEE_LOG or standard RUST_LOG env variables
-        let directives = env::var("EDGEE_LOG")
-            .or_else(|_| env::var("RUST_LOG"))
-            .unwrap_or_else(|_| {
+        // Get logging directives from log_filter or EDGEE_LOG or standard RUST_LOG env variables
+        let directives = log_filter
+            .or_else(|| env::var("EDGEE_LOG").ok())
+            .or_else(|| env::var("RUST_LOG").ok())
+            .unwrap_or_else(|| {
                 if let Some(ref span) = config.span {
                     format!("edgee[{span}]={}", config.level)
                 } else {

src/main.rs

@@ -18,14 +18,29 @@ struct Options {
 
     #[arg(short = 'f', long = "config", env = "EDGEE_CONFIG_PATH")]
     config_path: Option<PathBuf>,
+
+    #[arg(
+        short = 'c',
+        long = "debug-component",
+        help = "Launch Edgee and log only the specified component requests and responses to debug.",
+        id = "COMPONENT_NAME"
+    )]
+    debug_component: Option<String>,
 }
 
 #[tokio::main]
 async fn main() {
     let options = Options::parse();
 
-    config::config::init(options.config_path.as_deref());
-    logger::init(options.log_format);
+    config::config::init(&options);
+    // if debug_component is set, we only want to log the specified component. We change the options.log_format to do it.
+    let mut log_filter = None;
+    if options.debug_component.is_some() {
+        // We disable all logs because component will print things to stdout directly
+        log_filter = Some("[none]".to_string());
+    }
+
+    logger::init(options.log_format, log_filter);
     proxy::compute::data_collection::components::init();
 
     tokio::select! {

src/proxy/compute/data_collection/components/mod.rs

@@ -2,7 +2,8 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use http::{HeaderMap, HeaderName, HeaderValue};
-use tracing::{error, info, Instrument};
+use json_pretty::PrettyFormatter;
+use tracing::{error, info, span, Instrument, Level};
 
 use crate::config::config;
 use crate::proxy::compute::data_collection::payload::Event;
@@ -50,6 +51,14 @@ pub async fn send_data_collection(events: &Vec<Event>) -> anyhow::Result<()> {
         let user_agent = HeaderValue::from_str(&provider_event.context.client.user_agent)?;
 
         for cfg in config.components.data_collection.iter() {
+            let span = span!(
+                Level::INFO,
+                "component",
+                name = cfg.name.as_str(),
+                event = event_str
+            );
+            let _enter = span.enter();
+
             if !event.is_component_enabled(&cfg.name) {
                 continue;
             }
@@ -80,17 +89,15 @@ pub async fn send_data_collection(events: &Vec<Event>) -> anyhow::Result<()> {
                 Ok(Ok(request)) => request,
                 Ok(Err(err)) => {
                     error!(
-                        provider = cfg.name,
-                        event = event_str,
+                        step = "request",
                         err = err.to_string(),
                         "failed to handle data collection payload"
                     );
                     continue;
                 }
                 Err(err) => {
                     error!(
-                        provider = cfg.name,
-                        event = event_str,
+                        step = "request",
                         err = err.to_string(),
                         "failed to handle data collection payload"
                     );
@@ -110,24 +117,26 @@ pub async fn send_data_collection(events: &Vec<Event>) -> anyhow::Result<()> {
 
             let client = client.clone();
 
+            let method_str = match request.method {
+                provider::HttpMethod::Get => "GET",
+                provider::HttpMethod::Put => "PUT",
+                provider::HttpMethod::Post => "POST",
+                provider::HttpMethod::Delete => "DELETE",
+            };
+
+            info!(
+                step = "request",
+                method = method_str,
+                url = request.url,
+                body = request.body
+            );
+            debug_request(&request, cfg.name.as_str());
+
+
             // spawn a separated async thread
             tokio::spawn(
                 async move {
-                    let method_str = match request.method {
-                        provider::HttpMethod::Get => "GET",
-                        provider::HttpMethod::Put => "PUT",
-                        provider::HttpMethod::Post => "POST",
-                        provider::HttpMethod::Delete => "DELETE",
-                    };
-                    info!(
-                        step = "request",
-                        provider = cfg.name,
-                        event = event_str,
-                        method = method_str,
-                        url = request.url,
-                        body = request.body
-                    );
-
+                    let timer_start = std::time::Instant::now();
                     let res = match request.method {
                         provider::HttpMethod::Get => {
                             client.get(request.url).headers(headers).send().await
@@ -155,44 +164,89 @@ pub async fn send_data_collection(events: &Vec<Event>) -> anyhow::Result<()> {
 
                     match res {
                         Ok(res) => {
-                            if res.status().is_success() {
-                                let status_str = format!("{:?}", res.status());
-                                let body_res_str = res.text().await.unwrap_or_default();
-                                info!(
-                                    step = "response",
-                                    provider = cfg.name,
-                                    event = event_str,
-                                    method = method_str,
-                                    status = status_str,
-                                    body = body_res_str
-                                );
+                            let is_success = res.status().is_success();
+                            let status_str = format!("{:?}", res.status());
+                            let body_res_str = res.text().await.unwrap_or_default();
+
+                            if is_success {
+                                info!(step = "response", status = status_str, body = body_res_str);
                             } else {
-                                let status_str = format!("{:?}", res.status());
-                                let body_res_str = res.text().await.unwrap_or_default();
-                                error!(
-                                    step = "response",
-                                    provider = cfg.name,
-                                    event = event_str,
-                                    method = method_str,
-                                    status = status_str,
-                                    body = body_res_str
-                                );
+                                error!(step = "response", status = status_str, body = body_res_str);
                             }
+                            debug_response(cfg.name.as_str(), &status_str, timer_start, body_res_str, "".to_string());
                         }
                         Err(err) => {
-                            error!(
-                                step = "response",
-                                provider = cfg.name,
-                                event = event_str,
-                                method = method_str,
-                                err = err.to_string()
-                            );
+                            error!(step = "response", status = "500", err = err.to_string());
+                            debug_response(cfg.name.as_str(), "500", timer_start, "".to_string(), err.to_string());
                         }
                     }
                 }
-                .in_current_span(),
+                    .in_current_span(),
             );
         }
     }
     Ok(())
 }
+
+fn debug_request(request: &provider::EdgeeRequest, component_name: &str) {
+    let config = config::get();
+
+    let method_str = match request.method {
+        provider::HttpMethod::Get => "GET",
+        provider::HttpMethod::Put => "PUT",
+        provider::HttpMethod::Post => "POST",
+        provider::HttpMethod::Delete => "DELETE",
+    };
+
+    if config.log.debug_component.is_some() && config.log.debug_component.as_ref().unwrap() == component_name {
+        println!("-----------");
+        println!("  REQUEST  ");
+        println!("-----------\n");
+        println!("Method:   {}", method_str);
+        println!("Url:      {}", request.url);
+        if !request.headers.is_empty() {
+            print!("Headers:  ");
+            for (i, (key, value)) in request.headers.iter().enumerate() {
+                if i == 0 {
+                    println!("{}: {}", key, value);
+                } else {
+                    println!("          {}: {}", key, value);
+                }
+            }
+        } else {
+            println!("Headers:  None");
+        }
+
+        if !request.body.is_empty() {
+            println!("Body:");
+            let formatter = PrettyFormatter::from_str(request.body.as_str());
+            let result = formatter.pretty();
+            println!("{}", result);
+        } else {
+            println!("Body:     None");
+        }
+        println!();
+    }
+}
+
+fn debug_response(component_name: &str, status: &str, timer_start: std::time::Instant, body: String, error: String) {
+    let config = config::get();
+
+    if config.log.debug_component.is_some() && config.log.debug_component.as_ref().unwrap() == component_name {
+        println!("------------");
+        println!("  RESPONSE  ");
+        println!("------------\n");
+        println!("Status:   {}", status);
+        println!("Duration: {}ms", timer_start.elapsed().as_millis());
+        if !body.is_empty() {
+            println!("Body:");
+            let formatter = PrettyFormatter::from_str(body.as_str());
+            let result = formatter.pretty();
+            println!("{}", result);
+        }
+        if !error.is_empty() {
+            println!("Error:    {}", error);
+        }
+        println!();
+    }
+}