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

extern crate failure;
extern crate serde_json;
use futures::future::{done, ok, Either, Future};
use futures::stream::Stream;
use http::StatusCode;
use hyper::service::service_fn;
use hyper::Client;
use hyper::{Body, Request, Response, Server};
use log::{debug, error, trace};
use serde::de::DeserializeOwned;
use std::fmt::Debug;
use std::sync::Arc;
mod render_to_prometheus;
pub use render_to_prometheus::PrometheusCounter;

#[inline]
fn extract_body(
    req: hyper::client::ResponseFuture,
) -> impl Future<Item = String, Error = failure::Error> + Send {
    req.from_err().and_then(|resp| {
        debug!("response == {:?}", resp);
        let (_parts, body) = resp.into_parts();
        body.concat2()
            .from_err()
            .and_then(|complete_body| done(String::from_utf8(complete_body.to_vec())).from_err())
    })
}

pub fn create_string_future_from_hyper_request(
    request: hyper::Request<hyper::Body>,
) -> impl Future<Item = String, Error = failure::Error> {
    let https = hyper_rustls::HttpsConnector::new(4);
    let client = Client::builder().build::<_, hyper::Body>(https);

    extract_body(client.request(request))
        .from_err()
        .and_then(|text: String| {
            debug!("received_text == {:?}", text);
            ok(text)
        })
}

pub fn create_deserialize_future_from_hyper_request<T>(
    request: hyper::Request<hyper::Body>,
) -> impl Future<Item = T, Error = failure::Error>
where
    T: DeserializeOwned + std::fmt::Debug,
{
    create_string_future_from_hyper_request(request)
        .and_then(|text| done(serde_json::from_str(&text)).from_err())
        .and_then(|t: T| {
            debug!("deserialized object == {:?}", t);
            ok(t)
        })
}

fn check_compliance(req: &Request<Body>) -> Result<(), Response<Body>> {
    if req.uri() != "/metrics" {
        trace!("uri not allowed {}", req.uri());
        Err(Response::builder()
            .status(StatusCode::NOT_FOUND)
            .body(hyper::Body::empty())
            .unwrap())
    } else if req.method() != "GET" {
        trace!("method not allowed {}", req.method());
        Err(Response::builder()
            .status(StatusCode::METHOD_NOT_ALLOWED)
            .body(hyper::Body::empty())
            .unwrap())
    } else {
        Ok(())
    }
}

fn handle_request<O, P>(
    req: Request<Body>,
    options: Arc<O>,
    perform_request_box: P,
) -> impl Future<Item = Response<Body>, Error = failure::Error>
where
    O: Debug + Clone + Send + Sync + 'static,
    P: FnOnce(
            Request<Body>,
            &Arc<O>,
        )
            -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'static>
        + Send
        + Clone
        + 'static,
{
    trace!("{:?}", req);

    done(check_compliance(&req)).then(move |res| match res {
        Ok(_) => Either::A(perform_request_box(req, &options).then(|res| match res {
            Ok(body) => ok(body),
            Err(err) => {
                error!("internal server error: {:?}", err);
                ok(Response::builder()
                    .status(StatusCode::INTERNAL_SERVER_ERROR)
                    .body(hyper::Body::empty())
                    .unwrap())
            }
        })),
        Err(err) => Either::B(ok(err)),
    })
}

pub fn render_prometheus<O, P>(addr: &::std::net::SocketAddr, options: O, perform_request: P)
where
    O: Debug + Clone + Send + Sync + 'static,
    P: FnOnce(
            Request<Body>,
            &Arc<O>,
        )
            -> Box<Future<Item = Response<Body>, Error = failure::Error> + Send + 'static>
        + Send
        + Clone
        + 'static,
{
    // let's avoid unnecessary copies of our readonly data
    let options = Arc::new(options.clone());

    let new_svc = move || {
        let options = options.clone();
        let perform_request = perform_request.clone();
        service_fn(move |req| handle_request(req, options.clone(), perform_request.clone()))
    };

    let server = Server::bind(&addr)
        .serve(new_svc)
        .map_err(|e| eprintln!("server error: {}", e));
    hyper::rt::run(server);
}