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
|
use std::collections::HashMap;
use serde::{Serialize, Deserialize};
use hyper::{Request, Body, Method, Uri};
use http::request;
use url;
pub mod parse;
#[derive(Default, Deserialize, Debug, Clone)]
pub struct ProxyConfig {
pub bind_addr: Option<String>, // 127.0.0.1:9292
pub upstream_addr: Option<String>, // 127.0.0.1:9200
pub allow_all_indices: Option<bool>,
pub index: Vec<IndexConfig>
}
#[derive(Deserialize, Debug, Clone)]
pub struct IndexConfig {
pub name: String,
}
impl ProxyConfig {
pub fn allow_index(&self, name: &str) -> bool {
if self.allow_all_indices == Some(true) {
return true
}
for index in &self.index {
if index.name == name {
return true
}
}
false
}
}
#[derive(Debug)]
pub enum ProxyError {
Malformed(String),
ParseError(String),
NotAllowed(String),
NotSupported(String),
NotFound(String),
}
pub async fn parse_request(req: Request<Body>, config: &ProxyConfig) -> Result<Request<Body>, ProxyError> {
let (parts, body) = req.into_parts();
// split path into at most 3 chunks
let mut req_path = parts.uri.path();
if req_path.starts_with("/") {
req_path = &req_path[1..];
}
let path_chunks: Vec<&str> = req_path.split("/").collect();
if path_chunks.len() > 3 {
return Err(ProxyError::NotSupported("only request paths with up to three segments allowed".to_string()))
}
let raw_params: HashMap<String, String> = parts.uri.query()
.map(|v| {
url::form_urlencoded::parse(v.as_bytes())
.into_owned()
.collect()
})
.unwrap_or_else(HashMap::new);
// this is sort of like a router
let body = match (&parts.method, path_chunks.as_slice()) {
(&Method::GET, [""]) | (&Method::HEAD, [""]) => {
Body::empty()
},
(&Method::POST, ["_search", "scroll"]) | (&Method::DELETE, ["_search", "scroll"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
parse_request_scroll(None, &parts, &whole_body, config)?
},
(&Method::POST, ["_search", "scroll", key]) | (&Method::DELETE, ["_search", "scroll", key]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
parse_request_scroll(Some(key), &parts, &whole_body, config)?
},
(&Method::GET, [index, "_search"]) | (&Method::POST, [index, "_search"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
parse_request_search(index, &parts, &whole_body, config)?
},
(&Method::GET, [index, "_count"]) | (&Method::POST, [index, "_count"]) => {
let whole_body = hyper::body::to_bytes(body).await.unwrap();
parse_request_search(index, &parts, &whole_body, config)?
},
//(Method::GET, [index, "_count"]) => {
// parse_request_count(index, "_count", None, &parts, body, config)?
//},
(&Method::GET, [index, "_doc", key]) | (&Method::GET, [index, "_source", key]) => {
parse_request_read(index, path_chunks[1], key, &parts, config)?
},
_ => Err(ProxyError::NotSupported("unknown endpoint".to_string()))?,
};
// TODO: pass-through query parameters
let upstream_uri = Uri::builder()
.scheme("http")
.authority(config.upstream_addr.as_ref().unwrap_or(&"localhost:9200".to_string()).as_str())
.path_and_query(format!("{}", req_path).as_str())
.build()
.unwrap();
let upstream_req = Request::builder()
.uri(upstream_uri)
.method(&parts.method)
.body(body)
.unwrap();
Ok(upstream_req)
}
pub fn parse_request_scroll(key: Option<&str>, parts: &request::Parts, body: &[u8], config: &ProxyConfig) -> Result<Body, ProxyError> {
// XXX
//let _parsed: ScrollBody = serde_json::from_str(&body).unwrap();
Err(ProxyError::NotSupported("not yet implemented".to_string()))
}
pub fn parse_request_read(index: &str, endpoint: &str, key: &str, parts: &request::Parts, config: &ProxyConfig) -> Result<Body, ProxyError>{
if !config.allow_index(index) {
return Err(ProxyError::NotAllowed(format!("index doesn't exist or isn't proxied: {}", index)));
}
// XXX: no body needed?
Ok(Body::empty())
}
pub fn parse_request_search(index: &str, parts: &request::Parts, body: &[u8], config: &ProxyConfig) -> Result<Body, ProxyError> {
if !config.allow_index(index) {
return Err(ProxyError::NotAllowed(format!("index doesn't exist or isn't proxied: {}", index)));
}
// XXX: more checks
if body.len() > 0 {
let parsed: parse::ScrollBody = serde_json::from_slice(body).unwrap();
Ok(Body::from(serde_json::to_string(&parsed).unwrap()))
} else {
Ok(Body::empty())
}
}
|
