espipe 0.2.0

A minimalist command-line utility to pipe documents from a file or I/O stream into an Elasticsearch cluster.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

use elasticsearch::indices::{IndicesDeleteParts, IndicesRefreshParts};
use elasticsearch::{
    http::transport::{SingleNodeConnectionPool, TransportBuilder},
    CountParts, Elasticsearch,
};
use eyre::Result;
use serde_json::Value;
use std::{
    fs,
    path::PathBuf,
    process::Command,
    time::{SystemTime, UNIX_EPOCH},
};
use url::Url;

fn temp_dir(prefix: &str) -> PathBuf {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("time went backwards")
        .as_nanos();
    let dir = std::env::temp_dir().join(format!("{prefix}-{}-{nanos}", std::process::id()));
    fs::create_dir_all(&dir).expect("create temp dir");
    dir
}

fn test_index_name() -> String {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("time went backwards")
        .as_nanos();
    format!("espipe-test-{}-{nanos}", std::process::id())
}

fn write_input_file(dir: &PathBuf, filename: &str) -> PathBuf {
    let path = dir.join(filename);
    let contents = r#"{"message":"hello"}
{"message":"world"}
"#;
    fs::write(&path, contents).expect("write input file");
    path
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cli_ingests_into_elasticsearch_if_available() -> Result<()> {
    let base_url = Url::parse("http://localhost:9200")?;
    let transport =
        TransportBuilder::new(SingleNodeConnectionPool::new(base_url.clone())).build()?;
    let client = Elasticsearch::new(transport);

    if !is_connected(&client).await.unwrap_or(false) {
        eprintln!("Skipping Elasticsearch integration test; local node not available.");
        return Ok(());
    }

    let temp_dir = temp_dir("espipe-es-it");
    let input_path = write_input_file(&temp_dir, "bulk_input.ndjson");
    let index = test_index_name();
    let output_url = format!("{}/{}", base_url.as_str().trim_end_matches('/'), index);

    let status = Command::new(env!("CARGO_BIN_EXE_espipe"))
        .arg(&input_path)
        .arg(&output_url)
        .status()
        .expect("run espipe");

    assert!(status.success(), "espipe exited with failure");

    client
        .indices()
        .refresh(IndicesRefreshParts::Index(&[&index]))
        .send()
        .await?;

    let response = client.count(CountParts::Index(&[&index])).send().await?;
    let body: Value = response.json().await?;
    let count = body.get("count").and_then(Value::as_u64).unwrap_or(0);
    assert_eq!(count, 2);

    client
        .indices()
        .delete(IndicesDeleteParts::Index(&[&index]))
        .send()
        .await?;

    Ok(())
}

async fn is_connected(client: &Elasticsearch) -> Result<bool> {
    let response = match client.info().send().await {
        Ok(response) => response,
        Err(_) => return Ok(false),
    };

    let body: Value = match response.json().await {
        Ok(body) => body,
        Err(_) => return Ok(false),
    };

    Ok(body
        .get("tagline")
        .and_then(Value::as_str)
        .is_some_and(|tagline| tagline == "You Know, for Search"))
}