feat: UI improvement for integration test runner (#645)

* improve dir resolving and start up ordering

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* fix orphan process

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* Update tests/runner/src/util.rs, fix typo

Co-authored-by: Dongxu Wang <dongxu@apache.org>

* simplify logic via tokio timeout

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
Co-authored-by: Dongxu Wang <dongxu@apache.org>

tests/runner/src/env.rs

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 use std::fmt::Display;
+use std::fs::OpenOptions;
+use std::process::Stdio;
 use std::time::Duration;
 
 use async_trait::async_trait;
@@ -23,10 +25,12 @@ use client::{Client, Database as DB, Error as ClientError, ObjectResult, Select}
 use comfy_table::{Cell, Table};
 use sqlness::{Database, Environment};
 use tokio::process::{Child, Command};
-use tokio::time;
 
 use crate::util;
 
+const SERVER_ADDR: &str = "127.0.0.1:4001";
+const SERVER_LOG_FILE: &str = "/tmp/greptime-sqlness.log";
+
 pub struct Env {}
 
 #[async_trait]
@@ -48,16 +52,48 @@ impl Environment for Env {
 }
 
 impl Env {
+    #[allow(clippy::print_stdout)]
     pub async fn start_standalone() -> GreptimeDB {
-        let server_process = Command::new("cargo")
-            .current_dir("../")
-            .args(["run", "--", "standalone", "start", "-m"])
+        // Build the DB with `cargo build --bin greptime`
+        println!("Going to build the DB...");
+        let cargo_build_result = Command::new("cargo")
+            .current_dir(util::get_workspace_root())
+            .args(["build", "--bin", "greptime"])
+            .stdout(Stdio::null())
+            .output()
+            .await
+            .expect("Failed to start GreptimeDB")
+            .status;
+        if !cargo_build_result.success() {
+            panic!("Failed to build GreptimeDB (`cargo build` fails)");
+        }
+        println!("Build finished, starting...");
+
+        // Open log file (build logs will be truncated).
+        let log_file = OpenOptions::new()
+            .create(true)
+            .write(true)
+            .truncate(true)
+            .open(SERVER_LOG_FILE)
+            .unwrap_or_else(|_| panic!("Cannot open log file at {}", SERVER_LOG_FILE));
+        // Start the DB
+        let server_process = Command::new("./greptime")
+            .current_dir(util::get_binary_dir("debug"))
+            .args(["standalone", "start", "-m"])
+            .stdout(log_file)
             .spawn()
-            .unwrap_or_else(|_| panic!("Failed to start GreptimeDB"));
+            .expect("Failed to start the DB");
 
-        time::sleep(Duration::from_secs(3)).await;
+        let is_up = util::check_port(SERVER_ADDR.parse().unwrap(), Duration::from_secs(10)).await;
+        if !is_up {
+            panic!("Server doesn't up in 10 seconds, quit.")
+        }
+        println!(
+            "Started, going to test. Log will be write to {}",
+            SERVER_LOG_FILE
+        );
 
-        let client = Client::with_urls(vec!["127.0.0.1:4001"]);
+        let client = Client::with_urls(vec![SERVER_ADDR]);
         let db = DB::new("greptime", client.clone());
 
         GreptimeDB {

tests/runner/src/main.rs

@@ -21,7 +21,7 @@ mod util;
 #[tokio::main]
 async fn main() {
     let config = ConfigBuilder::default()
-        .case_dir("../cases".to_string())
+        .case_dir(util::get_case_dir())
         .build()
         .unwrap();
     let runner = Runner::new_with_config(config, Env {}).await.unwrap();

tests/runner/src/util.rs

@@ -12,8 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::time::Duration;
+
 use client::api::v1::column::Values;
 use client::api::v1::ColumnDataType;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpSocket;
+use tokio::time;
+
+/// Check port every 0.1 second.
+const PORT_CHECK_INTERVAL: Duration = Duration::from_millis(100);
 
 pub fn values_to_string(data_type: ColumnDataType, values: Values) -> Vec<String> {
     match data_type {
@@ -95,3 +105,58 @@ pub fn values_to_string(data_type: ColumnDataType, values: Values) -> Vec<String
             .collect(),
     }
 }
+
+/// Get the dir of test cases. This function only works when the runner is run
+/// under the project's dir because it depends on some envs set by cargo.
+pub fn get_case_dir() -> String {
+    // retrieve the manifest runner (./tests/runner)
+    let mut runner_crate_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+
+    // change directory to cases' dir from runner's (should be runner/../cases)
+    runner_crate_path.pop();
+    runner_crate_path.push("cases");
+
+    runner_crate_path.into_os_string().into_string().unwrap()
+}
+
+/// Get the dir that contains workspace manifest (the top-level Cargo.toml).
+pub fn get_workspace_root() -> String {
+    // retrieve the manifest runner (./tests/runner)
+    let mut runner_crate_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+
+    // change directory to workspace's root (runner/../..)
+    runner_crate_path.pop();
+    runner_crate_path.pop();
+
+    runner_crate_path.into_os_string().into_string().unwrap()
+}
+
+pub fn get_binary_dir(mode: &str) -> String {
+    // first go to the workspace root.
+    let mut workspace_root = PathBuf::from(get_workspace_root());
+
+    // change directory to target dir (workspace/target/<build mode>/)
+    workspace_root.push("target");
+    workspace_root.push(mode);
+
+    workspace_root.into_os_string().into_string().unwrap()
+}
+
+/// Spin-waiting a socket address is available, or timeout.
+/// Returns whether the addr is up.
+pub async fn check_port(ip_addr: SocketAddr, timeout: Duration) -> bool {
+    let check_task = async {
+        loop {
+            let socket = TcpSocket::new_v4().expect("Cannot create v4 socket");
+            match socket.connect(ip_addr).await {
+                Ok(mut stream) => {
+                    let _ = stream.shutdown().await;
+                    break;
+                }
+                Err(_) => time::sleep(PORT_CHECK_INTERVAL).await,
+            }
+        }
+    };
+
+    tokio::time::timeout(timeout, check_task).await.is_ok()
+}