fix: label all log lines with level

Author: d-perl

src/config.rs

@@ -227,7 +227,9 @@ pub fn assemble_config(paths: (std::path::PathBuf, Option<std::path::PathBuf>))
             let intervals = MetricIntervalsConfig::from_file(intervals_reference_file.clone());
             config.metrics.intervals.extend(intervals.intervals);
         } else {
-            println!("Metric intervals config file not found at {intervals_reference_file:?}")
+            println!(
+                "ERROR: Metric intervals config file not found at {intervals_reference_file:?}"
+            )
         }
     }
     if let Some(intervals) = metrics_path.map(MetricIntervalsConfig::from_file) {

src/loki_push.rs

@@ -89,13 +89,13 @@ pub async fn consumer_loop(
             .await
         {
             Ok(res) => {
-                println!("Sent {open} logs to loki.");
+                println!("DEBUG: Sent {open} logs to loki.");
                 if !res.status().is_success() {
                     let text = res
                         .text()
                         .await
                         .unwrap_or("[Unable to decode response text!]".into());
-                    println!("Received error response: {text} ");
+                    println!("ERROR: Received error response: {text} ");
                 }
             }
             Err(res) => {
@@ -104,7 +104,7 @@ pub async fn consumer_loop(
         };
         buffer.clear();
     }
-    println!("Producer dropped, consumer exiting");
+    println!("INFO: Producer dropped, consumer exiting");
 }
 
 #[cfg(test)]

src/main.rs

@@ -39,15 +39,15 @@ fn config_paths() -> (std::path::PathBuf, Option<std::path::PathBuf>) {
     let args = Args::parse();
     if !args.config.exists() {
         println!(
-            "Error, config file not found at {:?}",
+            "ERROR: config file not found at {:?}",
             args.config.as_path()
         );
         exit(1)
     }
     if let Some(ref metric_path) = args.metrics_config {
         if !metric_path.exists() {
             println!(
-                "Error, metrics config file not found at {:?}",
+                "ERROR: metrics config file not found at {:?}",
                 metric_path.as_path()
             );
             exit(1)
@@ -57,17 +57,17 @@ fn config_paths() -> (std::path::PathBuf, Option<std::path::PathBuf>) {
 }
 
 async fn run_services(config: &'static IngestorConfig) {
-    println!("Starting log ingestor with config: \n {:?}", &config);
+    println!("INFO: Starting log ingestor with config: \n {:?}", &config);
 
     let metrics = if config.enable_metrics {
-        println!("Starting metrics task...");
+        println!("DEBUG: Starting metrics task...");
         tokio::spawn(metrics_loop(config))
     } else {
         tokio::spawn(futures::future::ready(()))
     };
 
     if config.enable_logging {
-        println!("Starting log ingestor task...");
+        println!("DEBUG: Starting log ingestor task...");
         let (tx, mut rx) = mpsc::unbounded_channel::<LogMsg>();
         let producer = tokio::spawn(producer_loop(tx, config));
         consumer_loop(&mut rx, config).await;
@@ -84,6 +84,6 @@ async fn main() {
     let config = Box::leak(Box::new(assemble_config(config_paths())));
     rustls::crypto::aws_lc_rs::default_provider()
         .install_default()
-        .expect("Failed to install rustls crypto provider");
+        .expect("ERROR: Failed to install rustls crypto provider");
     run_services(config).await
 }

src/metrics.rs

@@ -262,7 +262,7 @@ async fn watchdog_loop(
             break;
         }
         interval.tick().await;
-        println!("Watchdog checking {:?}", futs.lock().await.keys());
+        println!("DEBUG: Watchdog checking {:?}", futs.lock().await.keys());
         let finished: Vec<String> = {
             futs.lock()
                 .await
@@ -288,7 +288,7 @@ async fn watchdog_loop(
                 }
             }
         }
-        println!("Watchdog done.");
+        println!("DEBUG: Watchdog done.");
     }
 }
 
@@ -337,30 +337,30 @@ async fn consumer_loop(rx: &mut mpsc::UnboundedReceiver<TimeSeries>, config: Met
         {
             Ok(res) => {
                 retries = 0;
-                println!("Sent {open} metrics to Mimir.");
+                println!("DEBUG: Sent {open} metrics to Mimir.");
                 if !res.status().is_success() {
                     let text = res
                         .text()
                         .await
                         .unwrap_or("[Unable to decode response text!]".into());
-                    println!("Received error response: {text} ");
+                    println!("ERROR: Received error response: {text} ");
                 }
             }
             Err(res) => {
                 println!("ERROR: {res:?}");
                 if retries == 3 {
-                    println!("Maximum retry attempts exceeded, exiting.");
+                    println!("ERROR: Maximum retry attempts exceeded, exiting.");
                     exit(0x45); // Service unavailable
                 }
-                println!("Retrying in 5s.");
+                println!("DEBUG: Retrying in 5s.");
                 sleep(Duration::from_secs(5)).await;
                 retries += 1;
             }
         };
         buffer.clear();
         proto_encoded_buffer.clear();
     }
-    println!("Producer dropped, consumer exiting");
+    println!("INFO: Producer dropped, consumer exiting");
 }
 
 /// Main routine to start the metrics service

src/redis_logs.rs

@@ -115,11 +115,11 @@ fn check_connection(
 ) -> Result<(), ()> {
     if let Ok(key_exists) = redis_conn.exists::<&str, bool>(&LOGGING_ENDPOINT[0]) {
         if !key_exists {
-            println!("Logging endpoint doesn't exist, exiting.");
+            println!("ERROR: Logging endpoint doesn't exist, exiting.");
             return Err(());
         }
     } else {
-        panic!("Something went wrong checking the logs endpoint")
+        panic!("ERROR: Something went wrong checking the logs endpoint")
     }
     setup_consumer_group(redis_conn, &config);
     Ok(())
@@ -128,7 +128,7 @@ pub async fn producer_loop(tx: mpsc::UnboundedSender<LogMsg>, config: &'static I
     let mut redis_conn =
         create_redis_conn(&config.redis.url.full_url()).expect("Could not connect to Redis!");
     if check_connection(&mut redis_conn, &config).is_err() {
-        println!("");
+        println!("ERROR: could not correctly set up redis connection");
         return;
     }
 
@@ -146,7 +146,7 @@ pub async fn producer_loop(tx: mpsc::UnboundedSender<LogMsg>, config: &'static I
 
                 for record in records {
                     if tx.send(record).is_err() {
-                        println!("Receiver dropped, stopping...");
+                        println!("INFO: Receiver dropped, stopping...");
                         break 'main;
                     }
                 }
@@ -157,10 +157,10 @@ pub async fn producer_loop(tx: mpsc::UnboundedSender<LogMsg>, config: &'static I
                 {
                     let new_conn = create_redis_conn(&config.redis.url.full_url());
                     if new_conn.is_err() {
-                        println!("Error reading from redis, retrying connection in 1s");
+                        println!("ERROR: Error reading from redis, retrying connection in 1s");
                         thread::sleep(Duration::from_millis(1000));
                     } else {
-                        println!("Reconnected");
+                        println!("INFO: Reconnected to redis");
                         let mut conn = new_conn.unwrap();
                         if check_connection(&mut conn, &config).is_err() {
                             return;