rustfs-cli 0.1.31

A Rust S3 CLI client for S3-compatible object storage
Documentation
#![cfg(not(windows))]

mod admin_support;

use std::net::{SocketAddr, TcpListener};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::thread;
use std::time::{Duration, Instant};

use admin_support::{rc_binary, rc_host_alias};
use serde_json::Value;

const ACCESS_KEY: &str = "ACCESS_KEY";
const SECRET_KEY: &str = "SECRET_KEY";
const ALIAS: &str = "realilm";
const BUCKET: &str = "rc-manual-transition-real-backend";
const RUSTFS_BINARY_ENV: &str = "RC_REAL_RUSTFS_BINARY";

struct RustfsProcess {
    child: Child,
}

impl Drop for RustfsProcess {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

#[test]
#[ignore = "requires RC_REAL_RUSTFS_BINARY pointing at a rustfs binary"]
fn ilm_transition_status_wait_cancel_hits_real_rustfs_backend() {
    let Some(rustfs_binary) = std::env::var_os(RUSTFS_BINARY_ENV).map(PathBuf::from) else {
        eprintln!("skipping real backend ILM transition CLI e2e; set {RUSTFS_BINARY_ENV}");
        return;
    };
    assert!(
        rustfs_binary.is_file(),
        "{RUSTFS_BINARY_ENV} must point at a rustfs binary: {}",
        rustfs_binary.display()
    );

    let config_dir = tempfile::tempdir().expect("create rc config dir");
    let data_dir = tempfile::tempdir().expect("create rustfs data dir");
    let address = free_loopback_addr();
    let endpoint = format!("http://{address}");
    let mut rustfs = start_rustfs(&rustfs_binary, &address, data_dir.path());
    wait_for_rc_ready(config_dir.path(), &endpoint);

    run_rc_success(
        config_dir.path(),
        &endpoint,
        &["mb", &format!("{ALIAS}/{BUCKET}")],
        "create test bucket",
    );
    let run = run_rc_success(
        config_dir.path(),
        &endpoint,
        &[
            "--json",
            "admin",
            "ilm",
            "transition",
            "run",
            ALIAS,
            BUCKET,
            "--async",
            "--max-objects",
            "1",
        ],
        "start async manual transition job",
    );
    let run_json: Value = serde_json::from_slice(&run.stdout).expect("run stdout should be JSON");
    assert_eq!(run_json["type"], "manual_transition_run");
    assert_eq!(run_json["data"]["state"], "accepted");
    assert_eq!(run_json["data"]["mode"], "durable_job");
    let job_id = run_json["data"]["job_id"]
        .as_str()
        .expect("async run must return job_id");
    assert!(!job_id.is_empty(), "async run must return non-empty job_id");
    assert_eq!(
        run_json["data"]["status_endpoint"],
        run_json["data"]["cancel_endpoint"]
    );

    let status = run_rc_success(
        config_dir.path(),
        &endpoint,
        &[
            "--json",
            "admin",
            "ilm",
            "transition",
            "status",
            ALIAS,
            job_id,
        ],
        "read manual transition job status",
    );
    let status_json: Value =
        serde_json::from_slice(&status.stdout).expect("status stdout should be JSON");
    assert_eq!(status_json["type"], "manual_transition_job_status");
    assert_eq!(status_json["data"]["job_id"], job_id);

    let wait = run_rc_success(
        config_dir.path(),
        &endpoint,
        &[
            "--json",
            "admin",
            "ilm",
            "transition",
            "wait",
            ALIAS,
            job_id,
            "--poll-interval-seconds",
            "1",
            "--timeout-seconds",
            "30",
        ],
        "wait for manual transition job",
    );
    let wait_json: Value =
        serde_json::from_slice(&wait.stdout).expect("wait stdout should be JSON");
    assert_eq!(wait_json["type"], "manual_transition_job_wait");
    assert_eq!(wait_json["data"]["job_id"], job_id);
    assert!(
        matches!(
            wait_json["data"]["status"].as_str(),
            Some("completed" | "partial")
        ),
        "wait should return a successful terminal state: {wait_json}"
    );

    let cancel = run_rc_success(
        config_dir.path(),
        &endpoint,
        &[
            "--json",
            "admin",
            "ilm",
            "transition",
            "cancel",
            ALIAS,
            job_id,
        ],
        "cancel/query terminal manual transition job",
    );
    let cancel_json: Value =
        serde_json::from_slice(&cancel.stdout).expect("cancel stdout should be JSON");
    assert_eq!(cancel_json["type"], "manual_transition_job_cancel");
    assert_eq!(cancel_json["data"]["job_id"], job_id);
    assert!(
        matches!(
            cancel_json["data"]["status"].as_str(),
            Some("completed" | "partial" | "failed" | "unknown")
        ),
        "terminal cancel must not rewrite a finished job into cancelled: {cancel_json}"
    );

    let _ = rustfs.child.kill();
}

fn start_rustfs(binary: &Path, address: &SocketAddr, data_dir: &Path) -> RustfsProcess {
    let child = Command::new(binary)
        .args([
            "--address",
            &address.to_string(),
            "--access-key",
            ACCESS_KEY,
            "--secret-key",
            SECRET_KEY,
            data_dir
                .to_str()
                .expect("temp data dir path should be UTF-8"),
        ])
        .env("RUSTFS_CONSOLE_ENABLE", "false")
        .env("RUSTFS_SCANNER_ENABLED", "false")
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .expect("start rustfs process");
    RustfsProcess { child }
}

fn wait_for_rc_ready(config_dir: &Path, endpoint: &str) {
    let deadline = Instant::now() + Duration::from_secs(60);
    loop {
        let output = run_rc(config_dir, endpoint, &["ls", ALIAS]);
        if output.status.success() {
            return;
        }
        assert!(
            Instant::now() < deadline,
            "rustfs backend did not become ready: stdout={}, stderr={}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
        thread::sleep(Duration::from_millis(250));
    }
}

fn run_rc_success(config_dir: &Path, endpoint: &str, args: &[&str], context: &str) -> Output {
    let output = run_rc(config_dir, endpoint, args);
    assert!(
        output.status.success(),
        "{context} failed: status={:?}, stdout={}, stderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    output
}

fn run_rc(config_dir: &Path, endpoint: &str, args: &[&str]) -> Output {
    Command::new(rc_binary())
        .args(args)
        .env("RC_CONFIG_DIR", config_dir)
        .env(format!("RC_HOST_{ALIAS}"), rc_host_alias(endpoint))
        .output()
        .expect("run rc command")
}

fn free_loopback_addr() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").expect("bind free loopback port");
    listener.local_addr().expect("read loopback address")
}