Use tracing

Signed-off-by: Xudong Sun <[email protected]>

e2e/Cargo.toml

@@ -33,3 +33,5 @@ thiserror = "1.0.29"
 tokio-stream = { version = "0.1.9", features = ["net"] }
 zookeeper = "0.8"
 tungstenite = "0.20.1"
+tracing = "0.1.36"
+tracing-subscriber = "0.3.17"

e2e/src/common.rs

@@ -14,6 +14,7 @@ use kube::{
 };
 use std::process::Command;
 use thiserror::Error;
+use tracing::*;
 
 #[derive(Debug, Error)]
 pub enum Error {
@@ -72,18 +73,18 @@ pub async fn apply(yaml: String, client: Client, discovery: &Discovery) -> Resul
     let gvk = if let Some(tm) = &obj.types {
         GroupVersionKind::try_from(tm).unwrap()
     } else {
-        println!("cannot apply object without valid TypeMeta {:?}", obj);
+        error!("cannot apply object without valid TypeMeta {:?}", obj);
         return Err(Error::ApplyFailed);
     };
     let name = obj.name_any();
     if let Some((ar, caps)) = discovery.resolve_gvk(&gvk) {
         let api = dynamic_api(ar, caps, client.clone(), namespace, false);
-        println!("Applying {}: \n{}", gvk.kind, serde_yaml::to_string(&obj)?);
+        info!("Applying {}: \n{}", gvk.kind, serde_yaml::to_string(&obj)?);
         let data: serde_json::Value = serde_json::to_value(&obj)?;
         let _r = api.patch(&name, &ssapply, &Patch::Apply(data)).await?;
-        println!("applied {} {}", gvk.kind, name);
+        info!("applied {} {}", gvk.kind, name);
     } else {
-        println!("Cannot apply document for unknown {:?}", gvk);
+        error!("Cannot apply document for unknown {:?}", gvk);
         return Err(Error::ApplyFailed);
     }
 
@@ -124,10 +125,10 @@ pub async fn get_output_and_err(mut attached: AttachedProcess) -> (String, Strin
 }
 
 pub fn run_command(program: &str, args: Vec<&str>, err_msg: &str) -> (String, String) {
-    println!("{} {}", program, args.join(" "));
+    info!("{} {}", program, args.join(" "));
     let cmd = Command::new(program).args(args).output().expect(err_msg);
-    println!("cmd output: {}", String::from_utf8_lossy(&cmd.stdout));
-    println!("cmd error: {}", String::from_utf8_lossy(&cmd.stderr));
+    info!("cmd output: {}", String::from_utf8_lossy(&cmd.stdout));
+    info!("cmd error: {}", String::from_utf8_lossy(&cmd.stderr));
     (
         String::from_utf8_lossy(&cmd.stdout).to_string(),
         String::from_utf8_lossy(&cmd.stderr).to_string(),

e2e/src/fluent_e2e.rs

@@ -1,6 +1,6 @@
 #![allow(unused_imports)]
 #![allow(unused_variables)]
-use k8s_openapi::api::core::v1::{Pod, ServiceAccount, Service};
+use k8s_openapi::api::core::v1::{Pod, Service, ServiceAccount};
 use k8s_openapi::api::rbac::v1::RoleBinding;
 use k8s_openapi::api::{apps::v1::DaemonSet, rbac::v1::Role};
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
@@ -20,6 +20,7 @@ use std::path::PathBuf;
 use std::thread;
 use std::time::{Duration, Instant};
 use tokio::time::sleep;
+use tracing::*;
 
 use crate::common::*;
 
@@ -115,6 +116,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
     loop {
         sleep(Duration::from_secs(5)).await;
         if start.elapsed() > timeout {
+            error!("Time out on desired state test");
             return Err(Error::Timeout);
         }
         // Check daemon set
@@ -127,7 +129,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         let role = role_api.get(&(fb_name.clone() + "-role")).await;
         match role {
             Err(e) => {
-                println!("Get role failed with error {}.", e);
+                info!("Get role failed with error {}.", e);
                 continue;
             }
             _ => {}
@@ -136,7 +138,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         let sa = sa_api.get(&fb_name.clone()).await;
         match sa {
             Err(e) => {
-                println!("Get service account failed with error {}.", e);
+                info!("Get service account failed with error {}.", e);
                 continue;
             }
             _ => {}
@@ -145,7 +147,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         let rb = rb_api.get(&(fb_name.clone() + "-role-binding")).await;
         match rb {
             Err(e) => {
-                println!("Get role binding failed with error {}.", e);
+                info!("Get role binding failed with error {}.", e);
                 continue;
             }
             _ => {}
@@ -154,7 +156,7 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         let svc = svc_api.get(&fb_name).await;
         match svc {
             Err(e) => {
-                println!("Get service failed with error {}.", e);
+                info!("Get service failed with error {}.", e);
                 continue;
             }
             _ => {}
@@ -163,24 +165,24 @@ pub async fn desired_state_test(client: Client, fb_name: String) -> Result<(), E
         let ds = ds_api.get(&fb_name).await;
         match ds {
             Err(e) => {
-                println!("Get daemon set failed with error {}.", e);
+                info!("Get daemon set failed with error {}.", e);
                 continue;
             }
             Ok(ds) => {
                 if ds.status.as_ref().unwrap().number_ready == node_number() {
                     // this number depends on the number of nodes
-                    println!("All daemons are ready now.");
+                    info!("All daemons are ready now.");
                     break;
                 } else {
-                    println!(
+                    info!(
                         "Only {} daemons are ready now.",
                         ds.status.as_ref().unwrap().number_ready
                     );
                 }
             }
         };
     }
-    println!("Desired state test passed.");
+    info!("Desired state test passed.");
     Ok(())
 }
 
@@ -206,14 +208,15 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
     loop {
         sleep(Duration::from_secs(5)).await;
         if start.elapsed() > timeout {
+            error!("Time out on relabel test");
             return Err(Error::Timeout);
         }
 
         // Check daemon set
         let ds = ds_api.get(&fb_name).await;
         match ds {
             Err(e) => {
-                println!("Get daemon set failed with error {}.", e);
+                info!("Get daemon set failed with error {}.", e);
                 continue;
             }
             Ok(ds) => {
@@ -230,7 +233,7 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
                     .unwrap()
                     .contains_key("key")
                 {
-                    println!("Label for pod is not updated yet");
+                    info!("Label for pod is not updated yet");
                     continue;
                 }
 
@@ -241,7 +244,7 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
                     .updated_number_scheduled
                     .is_none()
                 {
-                    println!("No daemon set pod is updated yet.");
+                    info!("No daemon set pod is updated yet.");
                     continue;
                 } else if *ds
                     .status
@@ -252,9 +255,9 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
                     .unwrap()
                     == node_number()
                 {
-                    println!("Relabel is done.");
+                    info!("Relabel is done.");
                 } else {
-                    println!(
+                    info!(
                         "Relabel is in progress. {} pods are updated now.",
                         ds.status
                             .as_ref()
@@ -267,10 +270,10 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
                 }
 
                 if ds.status.as_ref().unwrap().number_ready == node_number() {
-                    println!("All daemon set pods are ready.");
+                    info!("All daemon set pods are ready.");
                     break;
                 } else {
-                    println!(
+                    info!(
                         "Only {} pods are ready now.",
                         ds.status.as_ref().unwrap().number_ready
                     );
@@ -280,7 +283,7 @@ pub async fn relabel_test(client: Client, fb_name: String) -> Result<(), Error>
         };
     }
 
-    println!("Relabel test passed.");
+    info!("Relabel test passed.");
     Ok(())
 }
 
@@ -306,14 +309,15 @@ pub async fn service_selector_test(client: Client, fb_name: String) -> Result<()
     loop {
         sleep(Duration::from_secs(5)).await;
         if start.elapsed() > timeout {
+            error!("Time out on service selector test");
             return Err(Error::Timeout);
         }
 
         // Check daemon set
         let svc = svc_api.get(&fb_name).await;
         match svc {
             Err(e) => {
-                println!("Get service failed with error {}.", e);
+                info!("Get service failed with error {}.", e);
                 continue;
             }
             Ok(svc) => {
@@ -326,16 +330,16 @@ pub async fn service_selector_test(client: Client, fb_name: String) -> Result<()
                     .unwrap()
                     .contains_key("never-match-anything")
                 {
-                    println!("Selector for service is updated yet");
+                    info!("Selector for service is updated yet");
                     break;
                 } else {
-                    println!("Selector for service is not updated yet");
+                    info!("Selector for service is not updated yet");
                 }
             }
         };
     }
 
-    println!("Service selector test passed.");
+    info!("Service selector test passed.");
     Ok(())
 }
 
@@ -346,11 +350,11 @@ pub async fn fluent_e2e_test() -> Result<(), Error> {
     let fb_crd = crd_api.get("fluentbits.anvil.dev").await;
     match fb_crd {
         Err(e) => {
-            println!("No CRD found, create one before run the e2e test.");
+            error!("No CRD found, create one before run the e2e test.");
             return Err(Error::CRDGetFailed(e));
         }
         Ok(crd) => {
-            println!("CRD found, continue to run the e2e test.");
+            info!("CRD found, continue to run the e2e test.");
         }
     }
 
@@ -363,6 +367,6 @@ pub async fn fluent_e2e_test() -> Result<(), Error> {
     relabel_test(client.clone(), fb_name.clone()).await?;
     service_selector_test(client.clone(), fb_name.clone()).await?;
 
-    println!("E2e test passed.");
+    info!("E2e test passed.");
     Ok(())
 }

e2e/src/main.rs

@@ -7,46 +7,49 @@ pub mod zookeeper_e2e;
 
 use common::Error;
 use fluent_e2e::fluent_e2e_test;
-use rabbitmq_e2e::{rabbitmq_e2e_test, rabbitmq_scaling_e2e_test, rabbitmq_ephemeral_e2e_test};
+use rabbitmq_e2e::{rabbitmq_e2e_test, rabbitmq_ephemeral_e2e_test, rabbitmq_scaling_e2e_test};
 use std::str::FromStr;
 use std::{env, sync::Arc};
+use tracing::*;
 use zookeeper_e2e::{zookeeper_e2e_test, zookeeper_ephemeral_e2e_test, zookeeper_scaling_e2e_test};
 
 #[tokio::main]
 async fn main() -> Result<(), Error> {
+    tracing_subscriber::fmt::init();
+
     let args: Vec<String> = env::args().collect();
     let cmd = args[1].clone();
     match cmd.as_str() {
         "zookeeper" => {
-            println!("Running zookeeper end-to-end test");
+            info!("Running zookeeper end-to-end test");
             return zookeeper_e2e_test().await;
         }
         "zookeeper-scaling" => {
-            println!("Running zookeeper end-to-end test for scaling");
+            info!("Running zookeeper end-to-end test for scaling");
             return zookeeper_scaling_e2e_test().await;
         }
         "zookeeper-ephemeral" => {
-            println!("Running zookeeper end-to-end test for ephemeral storage");
+            info!("Running zookeeper end-to-end test for ephemeral storage");
             return zookeeper_ephemeral_e2e_test().await;
         }
         "rabbitmq" => {
-            println!("Running rabbitmq end-to-end test");
+            info!("Running rabbitmq end-to-end test");
             return rabbitmq_e2e_test().await;
         }
         "rabbitmq-scaling" => {
-            println!("Running rabbitmq end-to-end test for scaling");
+            info!("Running rabbitmq end-to-end test for scaling");
             return rabbitmq_scaling_e2e_test().await;
         }
         "rabbitmq-ephemeral" => {
-            println!("Running rabbitmq end-to-end test for ephemeral storage");
+            info!("Running rabbitmq end-to-end test for ephemeral storage");
             return rabbitmq_ephemeral_e2e_test().await;
         }
         "fluent" => {
-            println!("Running fluent end-to-end test");
+            info!("Running fluent end-to-end test");
             return fluent_e2e_test().await;
         }
         _ => {
-            println!("Please specify one controller");
+            error!("Wrong command. Please specify the correct e2e test workload.");
             Ok(())
         }
     }