proksi 0.8.0

A batteries-included reverse proxy with automatic HTTPS using Cloudflare Pingora and Let's Encrypt.
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

use std::{sync::Arc, time::Duration};

use async_trait::async_trait;
use pingora::{
    lb::{selection::RoundRobin, LoadBalancer},
    server::{ListenFds, ShutdownWatch},
    services::Service,
};

use crate::stores::{self};

/// Health check service that will run health checks on all upstreams
/// And update the route store with the new healthy upstreams.
/// This service will run in a separate thread.
pub struct HealthService {}

impl HealthService {
    pub fn new() -> Self {
        Self {}
    }
}

/// How often upstream names are resolved again (ADR 0006)
const RESOLVE_INTERVAL: Duration = Duration::from_secs(5);

/// The load balancers of every location and `tcp` block, shared with the route table: health
/// checks and new addresses apply to the routes being served
fn load_balancers() -> Vec<(String, Arc<LoadBalancer<RoundRobin>>)> {
    stores::get_routes()
        .iter()
        .flat_map(|(host, container)| {
            container
                .locations
                .iter()
                .map(|location| (host.clone(), location.load_balancer.clone()))
                .collect::<Vec<_>>()
        })
        .chain(crate::tcp::load_balancers().iter().cloned())
        .collect()
}

async fn run_health_check_loop() {
    let mut interval = tokio::time::interval(Duration::from_secs(30));
    interval.tick().await;

    loop {
        interval.tick().await;

        for (host, load_balancer) in load_balancers() {
            tracing::trace!("Running health check for host {}", host);
            load_balancer.backends().run_health_check(false).await;
        }

        record_health();
    }
}

/// Resolves upstream names again, replacing their addresses when they changed
async fn run_resolve_loop() {
    let mut interval = tokio::time::interval(RESOLVE_INTERVAL);
    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    interval.tick().await;

    loop {
        interval.tick().await;
        for (_, load_balancer) in load_balancers() {
            load_balancer.update().await.ok();
        }
    }
}

/// Sets `proksi_upstream_healthy` from the health checks: an upstream is healthy when one of
/// its addresses is
fn record_health() {
    for (_, host) in &stores::get_routes() {
        for location in &host.locations {
            let Some(metrics) = &location.metrics else {
                continue;
            };
            for (index, healthy) in location.upstream_health().into_iter().enumerate() {
                if let Some(upstream) = metrics.upstream(index) {
                    upstream.set_healthy(healthy);
                }
            }
        }
    }
}

#[async_trait]
impl Service for HealthService {
    async fn start_service(
        &mut self,
        _fds: Option<ListenFds>,
        _shutdown: ShutdownWatch,
        _listeners_per_fd: usize,
    ) {
        tracing::info!("Starting health check service");

        tokio::join!(run_health_check_loop(), run_resolve_loop());
    }

    fn name(&self) -> &'static str {
        "health_check_service"
    }

    fn threads(&self) -> Option<usize> {
        Some(1)
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;

    use crate::{
        routing::{build::build_host_container, test_utils::host},
        stores,
    };

    #[test]
    fn test_upstream_health_metrics() {
        crate::metrics::enable();
        let container = build_host_container(&host("health-metrics.test", "10.0.0.9")).unwrap();
        let location = container.locations[0].clone();
        stores::insert_route("health-metrics.test".to_string(), Arc::new(container));
        let healthy = || {
            crate::metrics::gather().contains(
                "proksi_upstream_healthy{host=\"health-metrics.test\",location=\"/*\",upstream=\"10.0.0.9:3000\"} 1",
            )
        };
        assert!(healthy());

        // A health check failed
        let backends = location.load_balancer.backends();
        let backend = backends.get_backend().iter().next().unwrap().clone();
        backends.set_enable(&backend, false);
        super::record_health();
        assert!(!healthy());

        backends.set_enable(&backend, true);
        super::record_health();
        assert!(healthy());
        stores::remove_route("health-metrics.test");
    }
}