espipe 0.4.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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

use serde::Deserialize;
use std::collections::HashMap;

#[derive(Deserialize)]
pub struct BulkResponse {
    error: Option<ErrorType>,
    //took: u64,
    errors: Option<bool>,
    items: Option<Vec<BulkAction>>,
}

#[derive(Deserialize)]
#[serde(untagged)]
enum ErrorType {
    Object(ErrorCause),
    String(String),
}

impl std::fmt::Display for ErrorType {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match self {
            ErrorType::Object(e) => write!(f, "{}", e),
            ErrorType::String(s) => write!(f, "{}", s),
        }
    }
}

impl BulkResponse {
    pub fn error_cause(&self) -> String {
        match &self.error {
            Some(cause) => format!("{cause}"),
            None => "unknown".to_string(),
        }
    }

    pub fn error_counts(&self) -> String {
        let mut error_types: HashMap<String, u64> = HashMap::new();
        if let (Some(true), Some(items)) = (self.errors, &self.items) {
            items.iter().for_each(|item| {
                item.error_message().map(|e| {
                    *error_types.entry(e).or_insert(0) += 1;
                });
            })
        }

        error_types
            .into_iter()
            .map(|(k, v)| format!("({v}) {k}"))
            .collect::<Vec<String>>()
            .join(", ")
    }

    pub fn has_errors(&self) -> bool {
        match self.errors {
            Some(true) => true,
            _ => false,
        }
    }

    pub fn success_count(&self) -> usize {
        match &self.items {
            Some(items) => items.iter().filter(|item| item.is_success()).count(),
            None => 0,
        }
    }
}

#[derive(Deserialize)]
struct ErrorCause {
    r#type: String,
    //reason: String,
}

impl std::fmt::Display for ErrorCause {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{}", self.r#type)
    }
}

#[derive(Deserialize)]
#[serde(untagged)]
enum BulkAction {
    Create { create: BulkResponseItem },
    Index { index: BulkResponseItem },
}

impl BulkAction {
    fn is_success(&self) -> bool {
        match self {
            BulkAction::Create { create } => create.status == 201,
            BulkAction::Index { index } => index.status == 200 || index.status == 201,
        }
    }

    fn error_type(&self) -> Option<String> {
        match self {
            BulkAction::Create { create } => create.error.as_ref().map(|e| e.to_string()),
            BulkAction::Index { index } => index.error.as_ref().map(|e| e.to_string()),
        }
    }

    fn index(&self) -> String {
        match self {
            BulkAction::Create { create } => create._index.clone(),
            BulkAction::Index { index } => index._index.clone(),
        }
    }

    fn error_message(&self) -> Option<String> {
        self.error_type().map(|e| format!("<{}> {e}", self.index()))
    }
}

#[derive(Deserialize)]
struct BulkResponseItem {
    _index: String,
    _id: String,
    status: u16,
    error: Option<ResponseError>,
}

#[derive(Deserialize)]
struct ResponseError {
    //r#type: String,
    //reason: String,
    caused_by: CausedBy,
}

#[derive(Deserialize)]
struct CausedBy {
    r#type: String,
    reason: String,
}

impl std::fmt::Display for ResponseError {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{} - {}", self.caused_by.r#type, self.caused_by.reason)
    }
}

impl TryFrom<serde_json::Value> for BulkResponse {
    type Error = eyre::Report;
    fn try_from(value: serde_json::Value) -> Result<Self, Self::Error> {
        let response: BulkResponse = serde_json::from_value(value)
            .map_err(|e| eyre::eyre!("Failed to parse BulkResponse: {:?}", e))?;
        Ok(response)
    }
}