estunnel 1.7.1

Tool for downloading data from 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

use crate::common::Result;
use reqwest::{Client, Response};
use serde::de;
use serde::de::MapAccess;
use serde_derive::*;
use serde_json;
use serde_json::Value;
use std::collections::HashMap;
use std::fmt;
use std::result;

#[derive(Serialize, Deserialize)]
pub struct ScrollResponse {
    pub _scroll_id: String,
    pub took:       u32,
    pub hits:       Hits,
}

#[derive(Serialize, Deserialize)]
pub struct Hits {
    #[serde(deserialize_with = "parse_total")]
    pub total: u64,
    pub max_score: f64,
    pub hits: Vec<Hit>,
}

#[derive(Serialize, Deserialize)]
pub struct Hit {
    pub _source: serde_json::Value,
}

fn parse_total<'de, D>(deserializer: D) -> result::Result<u64, D::Error>
where
    D: de::Deserializer<'de>,
{
    struct TotalVisitor;

    impl<'de> de::Visitor<'de> for TotalVisitor {
        type Value = u64;

        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
            formatter.write_str("u64 or map contains u64 value")
        }

        fn visit_u64<E>(self, v: u64) -> result::Result<Self::Value, E>
        where
            E: serde::de::Error,
        {
            Ok(v)
        }

        fn visit_map<A>(self, access: A) -> result::Result<Self::Value, A::Error>
        where
            A: MapAccess<'de>,
        {
            let map: HashMap<&str, Value> =
                serde::Deserialize::deserialize(de::value::MapAccessDeserializer::new(access))?;
            let v = map
                .get("value")
                .ok_or_else(|| de::Error::missing_field("value"))?
                .as_u64()
                .expect("not a valid u64");
            Ok(v)
        }
    }
    deserializer.deserialize_any(TotalVisitor)
}

pub fn parse_response(mut res: Response) -> Result<(Vec<String>, String, u64)> {
    let res = res.text()?;
    // serde_json has bad performance on reader. So we first read body into a string.
    // See: https://github.com/serde-rs/json/issues/160
    let res: ScrollResponse = serde_json::from_str(&res)?;
    let docs = res.hits.hits.iter().map(|hit| hit._source.to_string()).collect();
    Ok((docs, res._scroll_id, res.hits.total))
}

pub fn request_elastic(
    client: &Client,
    url: &str,
    query: &Value,
    user: &str,
    pass: &Option<String>,
    params: Option<Vec<(&str, String)>>,
) -> Result<Response> {
    let res = client.post(url).basic_auth(user, pass.clone()).json(query);

    let res = match params {
        Some(params) => res.query(&params),
        _ => res,
    };

    let res = res.send()?.error_for_status()?;
    Ok(res)
}