roperator 0.3.0

Easily create Kubernetes Operators with Rust
Documentation 
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

use crate::runner::RuntimeConfig;

use hyper::server::Server;
use hyper::service::{make_service_fn, service_fn};
use hyper::{Body, Method, Request, Response};
use tokio::runtime::Handle;

use std::net::SocketAddr;
use std::sync::Arc;

pub(crate) async fn start(
    _executor: Handle,
    port: u16,
    runtime_config: Arc<RuntimeConfig>,
    serve_metrics: bool,
    serve_health: bool,
) {
    let address: SocketAddr = ([0u8; 4], port).into();
    log::info!(
        "Starting server on address: {}, exposing '/metrics': {}, '/health': {}",
        address,
        serve_metrics,
        serve_health
    );

    let svc = Svc::new(runtime_config.clone(), serve_metrics, serve_health);
    let service = make_service_fn(move |_| {
        let service = svc.clone();
        async move {
            let service = service;
            Ok::<_, hyper::Error>(service_fn(move |request| {
                futures_util::future::ready(service.handle_request(request))
            }))
        }
    });
    if let Err(err) = Server::bind(&address).serve(service).await {
        log::error!("Server failed with error: {:?}", err);
    }
}

type Error = Box<dyn std::error::Error + Send + Sync>;

#[derive(Debug, Clone)]
struct Svc {
    runtime_config: Arc<RuntimeConfig>,
    serve_metrics: bool,
    serve_health: bool,
}

impl Svc {
    fn new(runtime_config: Arc<RuntimeConfig>, serve_metrics: bool, serve_health: bool) -> Svc {
        Svc {
            runtime_config,
            serve_metrics,
            serve_health,
        }
    }

    fn not_found(&self, _request: &Request<Body>) -> Result<Response<Body>, Error> {
        let resp = Response::builder().status(404).body(Body::empty())?;
        Ok(resp)
    }

    fn health(&self, _request: &Request<Body>) -> Result<Response<Body>, Error> {
        let resp = Response::builder().status(200).body(Body::empty())?;
        Ok(resp)
    }
    fn metrics(&self, _request: &Request<Body>) -> Result<Response<Body>, Error> {
        let body = self.runtime_config.metrics.encode_as_text()?;
        let resp = Response::builder()
            .status(200)
            .header(http::header::CONTENT_TYPE, prometheus::TEXT_FORMAT)
            .body(Body::from(body))?;
        Ok(resp)
    }

    fn handle_request(&self, request: Request<Body>) -> Result<Response<Body>, Error> {
        let req_path = request.uri().path().trim_end_matches('/');
        let req_method = request.method();

        log::debug!("Got http request {} {}", req_method, request.uri());

        let result = match (req_method, req_path) {
            (&Method::GET, "/health") if self.serve_health => self.health(&request),
            (&Method::GET, "/metrics") if self.serve_metrics => self.metrics(&request),
            _ => self.not_found(&request),
        };
        match result.as_ref() {
            Ok(ref resp) => {
                log::debug!(
                    "Finished handling {} {} with response status: {}",
                    req_method,
                    request.uri(),
                    resp.status()
                );
            }
            Err(ref err) => {
                log::error!(
                    "Error handling {} {} , error: {:?}",
                    req_method,
                    request.uri(),
                    err
                );
            }
        }
        result
    }
}