Integration testing for all commands

.github/workflows/test.yml

@@ -61,14 +61,13 @@ jobs:
             target
           key: ${{ runner.os }}-cargo-${{ hashFiles('**/Cargo.lock') }}
       - name: Install system dependencies
-        run: dnf install -y gcc git
+        run: dnf install -y gcc git make procps
       - name: Install toolchain
         uses: actions-rs/toolchain@v1
         with:
           toolchain: "1.84"
           override: true
       - name: Run tests
-        uses: actions-rs/cargo@v1
-        with:
-          command: test
-          args: --all-features
+        run: exec make tests
+        env:
+          RUST_LOG: debug

Makefile

@@ -0,0 +1,3 @@
+.PHONY: tests
+tests:
+	cargo test test_cargo_stop -- --nocapture --test-threads=1

src/commands/config/apply.rs

@@ -1,11 +1,12 @@
 use anyhow::{Context, Result};
 use derive_builder::Builder;
+use log::info;
 use serde::Deserialize;
 use serde_yaml::Value;
 use std::{
     collections::HashMap,
     fs,
-    io::Write,
+    io::{BufRead, BufReader, Read, Write},
     path::{Path, PathBuf},
     process::{Command, Stdio},
 };
@@ -28,14 +29,17 @@ fn apply_service_config(
     }
 
     for query in queries {
+        log::info!("picodata admin: {query}");
+
         let mut picodata_admin = Command::new("picodata")
             .arg("admin")
             .arg(
                 admin_socket
                     .to_str()
                     .context("path to picodata admin socket contains invalid characters")?,
             )
-            .stdout(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
             .stdin(Stdio::piped())
             .spawn()
             .context("failed to run picodata admin")?;
@@ -53,6 +57,18 @@ fn apply_service_config(
         picodata_admin
             .wait()
             .context("failed to wait for picodata admin")?;
+
+        let outputs: [Box<dyn Read + Send>; 2] = [
+            Box::new(picodata_admin.stdout.unwrap()),
+            Box::new(picodata_admin.stderr.unwrap()),
+        ];
+        for output in outputs {
+            let reader = BufReader::new(output);
+            for line in reader.lines() {
+                let line = line.expect("failed to read picodata admin output");
+                log::info!("picodata admin: {line}");
+            }
+        }
     }
 
     Ok(())
@@ -78,6 +94,8 @@ pub struct Params {
 }
 
 pub fn cmd(params: &Params) -> Result<()> {
+    info!("Applying plugin config...");
+
     let admin_socket = params
         .data_dir
         .join("cluster")
@@ -108,5 +126,7 @@ pub fn cmd(params: &Params) -> Result<()> {
         ))?;
     }
 
+    info!("Plugin config successfully applied.");
+
     Ok(())
 }

src/commands/plugin/pack.rs

@@ -68,8 +68,10 @@ pub fn cmd(pack_debug: bool, target_dir: &PathBuf) -> Result<()> {
     let normalized_package_name = cargo_manifest.package.name.replace('-', "_");
 
     let compressed_file = File::create(format!(
-        "target/{}-{}.tar.gz",
-        &normalized_package_name, cargo_manifest.package.version
+        "{}/{}-{}.tar.gz",
+        target_dir.display(),
+        &normalized_package_name,
+        cargo_manifest.package.version
     ))
     .context("failed to pack the plugin")?;
 

src/commands/run.rs

@@ -1,8 +1,9 @@
 use anyhow::{bail, Context, Result};
 use colored::Colorize;
+use core::panic;
 use derive_builder::Builder;
 use lib::cargo_build;
-use log::{error, info};
+use log::{error, info, warn};
 use rand::Rng;
 use serde::Deserialize;
 use std::collections::BTreeMap;
@@ -99,7 +100,7 @@ fn enable_plugins(topology: &Topology, data_dir: &Path, picodata_path: &PathBuf)
         // add migration context
         for migration_env in &plugin.migration_context {
             queries.push(format!(
-                "ALTER PLUGIN {plugin_name} {plugin_version} SET migration_context.{}='{}';",
+                "ALTER PLUGIN \"{plugin_name}\" {plugin_version} SET migration_context.{}='{}';",
                 migration_env.name, migration_env.value
             ));
         }
@@ -353,6 +354,8 @@ impl Drop for PicodataInstance {
             return;
         }
 
+        warn!("I AM DESTRUCTOR");
+
         self.child
             .wait()
             .expect("Failed to wait for picodata instance");
@@ -474,21 +477,20 @@ pub fn cmd(params: &Params) -> Result<()> {
 
     // Run in the loop until the child processes are killed
     // with cargo stop or Ctrl+C signal is recieved
-    let picodata_processes = pico_instances.clone();
-    ctrlc::set_handler(move || {
-        info!("received Ctrl+C. Shutting down ...");
-
-        let mut childs = picodata_processes.lock().unwrap();
-        for process in childs.iter_mut() {
-            process.kill().unwrap_or_else(|e| {
-                error!("failed to kill picodata instances: {:#}", e);
-            });
-        }
-
-        exit(0);
-    })
-    .context("failed to set Ctrl+c handler")?;
-
+    // let picodata_processes = pico_instances.clone();
+    // ctrlc::set_handler(move || {
+    //     info!("received Ctrl+C. Shutting down ...");
+
+    //     let mut childs = picodata_processes.lock().unwrap();
+    //     for process in childs.iter_mut() {
+    //         process.kill().unwrap_or_else(|e| {
+    //             error!("failed to kill picodata instances: {:#}", e);
+    //         });
+    //     }
+
+    //     exit(0);
+    // })
+    // .context("failed to set Ctrl+c handler")?;
     for instance in pico_instances.lock().unwrap().iter_mut() {
         instance.join();
     }

src/commands/stop.rs

@@ -1,6 +1,8 @@
 use anyhow::{bail, Context, Result};
 use derive_builder::Builder;
 use log::info;
+use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus};
+use nix::unistd::Pid;
 use std::fs::{self};
 use std::io::{self, BufRead};
 use std::path::{Path, PathBuf};
@@ -85,6 +87,18 @@ fn kill_process_by_pid(pid: u32) -> Result<()> {
     let output = Command::new("kill")
         .args(["-9", &pid.to_string()])
         .output()?;
+    let wait_status = waitpid(Pid::from_raw(pid as i32), Some(WaitPidFlag::WNOHANG));
+
+    match wait_status {
+        Ok(WaitStatus::Exited(_, status)) => {
+            info!("Process reaped successfully, exit code: {}", status)
+        }
+        Ok(WaitStatus::Signaled(_, signal, _)) => {
+            info!("Process killed by signal: {:?}", signal)
+        }
+        Ok(_) => bail!("Unexpected waitpid result"),
+        Err(e) => bail!("Failed to reap process: {}", e),
+    }
 
     if !output.status.success() {
         bail!("failed to kill picodata instance (pid: {pid}): {output:?}");

src/main.rs

@@ -1,6 +1,10 @@
 use anyhow::{Context, Result};
 use clap::{Parser, Subcommand};
-use nix::unistd::{fork, ForkResult};
+use log::info;
+use nix::{
+    sys::wait::{waitpid, WaitPidFlag, WaitStatus},
+    unistd::{fork, ForkResult, Pid},
+};
 use std::{env, path::PathBuf, process, thread, time::Duration};
 
 mod commands;
@@ -158,6 +162,21 @@ fn run_child_killer() {
         let ret = unsafe { libc::kill(master_pid, 0) };
         if ret != 0 {
             unsafe { libc::killpg(master_pid, libc::SIGKILL) };
+
+            while let Ok(status) = waitpid(Pid::from_raw(-master_pid), Some(WaitPidFlag::WNOHANG)) {
+                match status {
+                    WaitStatus::Exited(pid, code) => {
+                        info!("Reaped PID {} with exit code {}", pid, code)
+                    }
+                    WaitStatus::Signaled(pid, sig, _) => {
+                        info!("Reaped PID {} killed by {:?}", pid, sig)
+                    }
+                    WaitStatus::StillAlive
+                    | WaitStatus::Stopped(_, _)
+                    | WaitStatus::Continued(_) => break,
+                    _ => {}
+                }
+            }
             break;
         }
 

tests/assets/topology.toml

@@ -0,0 +1,11 @@
+[tier.default]
+replicasets = 2
+replication_factor = 2
+
+[plugin.test-plugin]
+migration_context = [
+    { name = "example_name", value = "example_value" },
+]
+
+[plugin.test-plugin.service.main]
+tiers = ["default"]

tests/build.rs

@@ -0,0 +1,65 @@
+mod helpers;
+
+use helpers::{
+    build_plugin, check_plugin_version_artefacts, run_pike, wait_for_proc, PLUGIN_DIR, TESTS_DIR,
+};
+use std::{
+    fs::{self},
+    path::Path,
+    time::Duration,
+    vec,
+};
+
+#[test]
+fn test_cargo_build() {
+    // Cleaning up metadata from past run
+    if Path::new(PLUGIN_DIR).exists() {
+        fs::remove_dir_all(PLUGIN_DIR).unwrap();
+    }
+
+    let mut plugin_creation_proc =
+        run_pike(vec!["plugin", "new", "test-plugin"], TESTS_DIR, &vec![]).unwrap();
+
+    wait_for_proc(&mut plugin_creation_proc, Duration::from_secs(10));
+
+    build_plugin(&helpers::BuildType::Debug, "0.1.0");
+    build_plugin(&helpers::BuildType::Debug, "0.1.1");
+    build_plugin(&helpers::BuildType::Release, "0.1.0");
+    build_plugin(&helpers::BuildType::Release, "0.1.1");
+
+    assert!(check_plugin_version_artefacts(
+        &Path::new(PLUGIN_DIR)
+            .join("target")
+            .join("debug")
+            .join("test-plugin")
+            .join("0.1.0"),
+        false
+    ));
+
+    assert!(check_plugin_version_artefacts(
+        &Path::new(PLUGIN_DIR)
+            .join("target")
+            .join("debug")
+            .join("test-plugin")
+            .join("0.1.1"),
+        true
+    ));
+
+    assert!(check_plugin_version_artefacts(
+        &Path::new(PLUGIN_DIR)
+            .join("target")
+            .join("release")
+            .join("test-plugin")
+            .join("0.1.0"),
+        false
+    ));
+
+    assert!(check_plugin_version_artefacts(
+        &Path::new(PLUGIN_DIR)
+            .join("target")
+            .join("release")
+            .join("test-plugin")
+            .join("0.1.1"),
+        true
+    ));
+}

tests/config.rs

@@ -0,0 +1,34 @@
+mod helpers;
+
+use helpers::{get_picodata_table, run_cluster, run_pike, wait_for_proc, CmdArguments, PLUGIN_DIR};
+use std::{
+    path::Path, thread, time::{Duration, Instant}, vec
+};
+
+const TOTAL_INSTANCES: i32 = 4;
+
+#[test]
+fn test_config_apply() {
+    let _cluster_handle = run_cluster(
+        Duration::from_secs(120),
+        TOTAL_INSTANCES,
+        CmdArguments::default(),
+    )
+    .unwrap();
+
+    thread::sleep(Duration::from_secs(30));
+
+    let mut plugin_creation_proc = run_pike(vec!["config", "apply"], PLUGIN_DIR, &vec![]).unwrap();
+
+    wait_for_proc(&mut plugin_creation_proc, Duration::from_secs(10));
+
+    let start = Instant::now();
+    while Instant::now().duration_since(start) < Duration::from_secs(60) {
+        let pico_plugin_config = get_picodata_table(Path::new("tmp"), "_pico_plugin_config");
+        if pico_plugin_config.contains("value") && pico_plugin_config.contains("changed") {
+            return;
+        }
+    }
+
+    panic!("Timeouted while trying to apply cluster config, value hasn't changed");
+}