autumn_plugin_pingora/
plugin.rs

1//! [`PingoraPlugin`]: the builder and the Autumn `Plugin` implementation.
2
3use std::borrow::Cow;
4use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
5use std::sync::{Arc, OnceLock};
6
7use autumn_web::AppState;
8use autumn_web::app::AppBuilder;
9use autumn_web::config::AutumnConfig;
10use autumn_web::plugin::Plugin;
11use autumn_web::route_listing::{RouteClassification, RouteInfo};
12use pingora_core::server::configuration::ServerConf;
13
14use crate::config::{
15    ConfigError, DEFAULT_SECTION, Fallback, PingoraConfig, Resolved, Route, normalize_prefix,
16};
17use crate::error::PingoraError;
18use crate::forwarded::IpRange;
19use crate::health::ProxyHealth;
20use crate::metrics::{Metrics, ProxyMetrics};
21use crate::proxy::{Gateway, Timeouts};
22use crate::router::Router;
23use crate::server::{Launch, PingoraHandle, Shared, Started};
24use crate::upstream::Pool;
25
26/// The name in `Plugin::name`, logs and errors.
27pub const PLUGIN_NAME: &str = "autumn-plugin-pingora";
28
29/// Attempts for a request after a pooled upstream connection was stale.
30const REUSE_RETRIES: usize = 2;
31
32/// Time to close aborted connections, inside Autumn's shutdown budget.
33const KILL_WAIT_MS: u64 = 500;
34
35type Override = Arc<dyn Fn(&mut PingoraConfig) + Send + Sync>;
36
37/// Runs a Pingora reverse proxy inside an Autumn app.
38///
39/// ```rust,ignore
40/// use autumn_plugin_pingora::{PingoraPlugin, Route};
41///
42/// autumn_web::app()
43///     .routes(routes![index])
44///     .plugin(PingoraPlugin::new().route(
45///         Route::new("billing").path_prefix("/billing").upstream("10.0.0.7:8080"),
46///     ))
47///     .run()
48///     .await;
49/// ```
50///
51/// The plugin reads `[pingora]` from `autumn.toml` (see [`PingoraConfig`]).
52/// Fluent setters apply on top of the file values.
53///
54/// On boot, the plugin:
55///
56/// - resolves the upstreams,
57/// - binds the listener (a failure stops boot),
58/// - starts the health checks,
59/// - adds a `pingora` health indicator and `pingora_proxy_*` metrics,
60/// - puts a [`PingoraHandle`] into `AppState`.
61///
62/// The plugin drains when Autumn's shutdown signal fires.
63pub struct PingoraPlugin {
64    explicit: Option<Box<PingoraConfig>>,
65    overrides: Vec<Override>,
66    development: Option<bool>,
67    public: bool,
68    resolved: OnceLock<Result<Resolved, ConfigError>>,
69    shared: Arc<Shared>,
70}
71
72impl Default for PingoraPlugin {
73    fn default() -> Self {
74        Self::new()
75    }
76}
77
78impl std::fmt::Debug for PingoraPlugin {
79    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
80        f.debug_struct("PingoraPlugin")
81            .field("config", &self.effective_config())
82            .finish_non_exhaustive()
83    }
84}
85
86impl PingoraPlugin {
87    /// A plugin with no routes, configured from `[pingora]`.
88    #[must_use]
89    pub fn new() -> Self {
90        Self {
91            explicit: None,
92            overrides: Vec::new(),
93            development: None,
94            public: false,
95            resolved: OnceLock::new(),
96            shared: Arc::new(Shared::default()),
97        }
98    }
99
100    /// Add a route after the routes from config.
101    #[must_use]
102    pub fn route(self, route: Route) -> Self {
103        self.configure(move |c| c.routes.push(route.clone()))
104    }
105
106    /// Change the configuration in code, after files and environment.
107    #[must_use]
108    pub fn configure(mut self, apply: impl Fn(&mut PingoraConfig) + Send + Sync + 'static) -> Self {
109        self.overrides.push(Arc::new(apply));
110        self.resolved = OnceLock::new();
111        self
112    }
113
114    /// Use `config` and read no config files. The profile still comes from
115    /// the environment (see [`development`](Self::development)). Fluent
116    /// setters still apply on top.
117    #[must_use]
118    pub fn config(mut self, config: PingoraConfig) -> Self {
119        self.explicit = Some(Box::new(config));
120        self.resolved = OnceLock::new();
121        self
122    }
123
124    /// Force development (`true`) or production (`false`) defaults. By
125    /// default, the active profile decides.
126    #[must_use]
127    pub fn development(mut self, development: bool) -> Self {
128        self.development = Some(development);
129        self.resolved = OnceLock::new();
130        self
131    }
132
133    /// Set the listen address, `IP:port`.
134    #[must_use]
135    pub fn bind(self, addr: impl Into<String>) -> Self {
136        let addr = addr.into();
137        self.configure(move |c| c.bind.clone_from(&addr))
138    }
139
140    /// Set where unmatched requests go.
141    #[must_use]
142    pub fn fallback(self, fallback: Fallback) -> Self {
143        self.configure(move |c| c.fallback = fallback)
144    }
145
146    /// Mark the proxied prefixes as public in `autumn routes`. If you do
147    /// not, they are unclassified. Then `autumn routes audit` fails.
148    #[must_use]
149    pub const fn public(mut self) -> Self {
150        self.public = true;
151        self
152    }
153
154    /// A handle to this plugin's proxy. It is valid before and after boot.
155    #[must_use]
156    pub fn handle(&self) -> PingoraHandle {
157        PingoraHandle::new(self.shared.clone())
158    }
159
160    /// The proxied prefixes as `autumn routes` lists them: method `PROXY`,
161    /// path `[host]/prefix/*`. The fallback is not listed: the app routes
162    /// cover it.
163    #[must_use]
164    pub fn route_infos(&self) -> Vec<RouteInfo> {
165        let Ok(resolved) = self.resolved() else {
166            return Vec::new();
167        };
168        if !resolved.config.enabled {
169            return Vec::new();
170        }
171        let classification = if self.public {
172            RouteClassification::Public
173        } else {
174            RouteClassification::Unclassified
175        };
176        resolved
177            .config
178            .routes
179            .iter()
180            .map(|route| RouteInfo {
181                method: "PROXY".to_owned(),
182                path: format!("{}{}/*", route.host, normalize_prefix(&route.path_prefix)),
183                handler: format!("{PLUGIN_NAME}::{}", route.name),
184                classification,
185                ..RouteInfo::default()
186            })
187            .collect()
188    }
189
190    /// The configuration after files, environment and code.
191    ///
192    /// # Errors
193    ///
194    /// The [`ConfigError`] that aborts boot.
195    pub fn effective_config(&self) -> Result<&PingoraConfig, &ConfigError> {
196        self.resolved().as_ref().map(|r| &r.config)
197    }
198
199    fn resolved(&self) -> &Result<Resolved, ConfigError> {
200        self.resolved.get_or_init(|| {
201            let mut resolved = match &self.explicit {
202                Some(config) => Resolved::explicit((**config).clone()),
203                None => PingoraConfig::resolve(DEFAULT_SECTION)?,
204            };
205            for apply in &self.overrides {
206                apply(&mut resolved.config);
207            }
208            if let Some(development) = self.development {
209                if development { "dev" } else { "prod" }.clone_into(&mut resolved.profile);
210            }
211            resolved.config.validate()?;
212            Ok(resolved)
213        })
214    }
215}
216
217impl Plugin for PingoraPlugin {
218    fn name(&self) -> Cow<'static, str> {
219        Cow::Borrowed(PLUGIN_NAME)
220    }
221
222    fn build(self, app: AppBuilder) -> AppBuilder {
223        let app = app.config_section(DEFAULT_SECTION);
224        let shared = self.shared.clone();
225        let (config, development) = match self.resolved() {
226            Ok(resolved) => (resolved.config.clone(), resolved.is_development()),
227            Err(error) => {
228                return fail_at_startup(app, shared, &PingoraError::Config(error.clone()));
229            }
230        };
231        if !config.enabled {
232            tracing::info!("pingora proxy disabled by configuration");
233            return app;
234        }
235        let declared = self.route_infos();
236        let app = if config.metrics {
237            app.metrics_source(PLUGIN_NAME, Arc::new(ProxyMetrics(shared.clone())))
238        } else {
239            app
240        };
241        let hook_shared = shared.clone();
242        let stop_shared = shared.clone();
243        app.declare_plugin_routes(declared)
244            .health_indicator("pingora", Arc::new(ProxyHealth(shared)))
245            .on_startup(move |state| {
246                let shared = hook_shared.clone();
247                let config = config.clone();
248                async move {
249                    match launch(&state, config, development, &shared).await {
250                        Ok(Started::Serving(addr)) => {
251                            tracing::info!(%addr, "pingora proxy listening");
252                            state.insert_extension(PingoraHandle::new(shared.clone()));
253                            drain_on_autumn_shutdown(&state, shared);
254                            Ok(())
255                        }
256                        Ok(Started::Stopped) => {
257                            tracing::info!("pingora proxy stopped before it started");
258                            Ok(())
259                        }
260                        Err(error) => {
261                            shared.fail();
262                            Err(startup_error(&error.to_string()))
263                        }
264                    }
265                }
266            })
267            .on_shutdown(move || {
268                let handle = PingoraHandle::new(stop_shared.clone());
269                async move { handle.shutdown().await }
270            })
271    }
272}
273
274/// Fail at startup, where Autumn reports hook errors.
275fn fail_at_startup(app: AppBuilder, shared: Arc<Shared>, error: &PingoraError) -> AppBuilder {
276    let message = error.to_string();
277    app.on_startup(move |_state| {
278        let shared = shared.clone();
279        let message = message.clone();
280        async move {
281            shared.fail();
282            Err(startup_error(&message))
283        }
284    })
285}
286
287fn startup_error(message: &str) -> autumn_web::AutumnError {
288    autumn_web::AutumnError::internal_server_error_msg(format!("{PLUGIN_NAME}: {message}"))
289}
290
291/// Start the drain when Autumn's shutdown signal fires (ADR 0003).
292fn drain_on_autumn_shutdown(state: &AppState, shared: Arc<Shared>) {
293    let token = state.shutdown_token();
294    tokio::spawn(async move {
295        token.cancelled().await;
296        PingoraHandle::new(shared).shutdown().await;
297    });
298}
299
300async fn launch(
301    state: &AppState,
302    mut config: PingoraConfig,
303    development: bool,
304    shared: &Arc<Shared>,
305) -> Result<Started, PingoraError> {
306    if shared.lifecycle.get() != crate::Lifecycle::Idle {
307        return Err(PingoraError::AlreadyStarted);
308    }
309    let autumn = state.config_arc();
310    let bind = config.bind_addr(development)?;
311    let fallback = match config.fallback {
312        Fallback::App => Some(app_target(&autumn, bind).await?),
313        Fallback::NotFound => None,
314    };
315    if fallback.is_some_and(|target| !trusts_proxy(&autumn, target)) {
316        tracing::warn!(
317            "pingora proxy: the Autumn app does not trust the proxy. The app sees 127.0.0.1 as the client. Add 127.0.0.1/32 to `security.trusted_proxies.ranges`. Set `security.trusted_proxies.trust_forwarded_headers = true`"
318        );
319    }
320    fit_grace(&mut config, &autumn);
321
322    let mut pools = Vec::with_capacity(config.routes.len());
323    for route in &config.routes {
324        pools.push(Pool::build(route, config.health_check_timeout()).await?);
325    }
326    let pools = Arc::new(pools);
327    let route_names: Vec<String> = config.routes.iter().map(|r| r.name.clone()).collect();
328    let metrics = config.metrics.then(|| {
329        let metrics = Arc::new(Metrics::new(route_names.iter().map(String::as_str)));
330        let _ = shared.metrics.set(metrics.clone());
331        metrics
332    });
333
334    let listener = tokio::net::TcpListener::bind(bind)
335        .await
336        .map_err(|source| PingoraError::Bind { addr: bind, source })?;
337
338    let server_conf = Arc::new(ServerConf {
339        // Pingora counts all attempts: the first one, connect retries
340        // (capped in `fail_to_connect`) and retries on a stale pooled
341        // connection.
342        max_retries: config
343            .max_retries
344            .saturating_add(1)
345            .saturating_add(REUSE_RETRIES),
346        ..ServerConf::default()
347    });
348    let gateway = Gateway {
349        router: Router::new(&config.routes),
350        routes: config.routes.clone(),
351        pools: pools.clone(),
352        fallback,
353        trusted: config
354            .trusted_proxies
355            .iter()
356            .filter_map(|range| IpRange::parse(range))
357            .collect(),
358        max_body: config.max_request_body_bytes,
359        max_retries: config.max_retries,
360        timeouts: Timeouts {
361            connect: config.connect_timeout(),
362            read: config.read_timeout(),
363            write: config.write_timeout(),
364        },
365        metrics,
366    };
367    let proxy = pingora_proxy::http_proxy(&server_conf, gateway);
368    crate::server::start(
369        shared,
370        Launch {
371            listener,
372            proxy,
373            pools,
374            route_names,
375            max_connections: config.max_connections,
376            max_connections_per_ip: config.max_connections_per_ip,
377            grace: config.shutdown_grace(),
378            health_interval: config.health_check_interval(),
379        },
380    )
381}
382
383/// The Autumn app address for `fallback = "app"` (ADR 0002).
384async fn app_target(autumn: &AutumnConfig, bind: SocketAddr) -> Result<SocketAddr, PingoraError> {
385    if autumn.server.tls.is_some() {
386        return Err(PingoraError::FallbackTls);
387    }
388    let host = autumn.server.host.trim();
389    let port = autumn.server.port;
390    if port == 0 {
391        return Err(PingoraError::AppAddress(format!(
392            "{host}:0 (the port is not known before boot; set `server.port`)"
393        )));
394    }
395    let target = match host.parse::<IpAddr>() {
396        Ok(ip) => SocketAddr::new(loopback_for(ip), port),
397        Err(_) if host.eq_ignore_ascii_case("localhost") => {
398            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port)
399        }
400        Err(_) => tokio::net::lookup_host((host, port))
401            .await
402            .ok()
403            .and_then(|mut addrs| addrs.next())
404            .ok_or_else(|| PingoraError::AppAddress(format!("{host}:{port}")))?,
405    };
406    if loops(bind, target) {
407        return Err(PingoraError::FallbackLoop(target));
408    }
409    Ok(target)
410}
411
412/// An unspecified address (`0.0.0.0`, `::`) becomes loopback.
413const fn loopback_for(ip: IpAddr) -> IpAddr {
414    match ip {
415        IpAddr::V4(v4) if v4.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
416        IpAddr::V6(v6) if v6.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
417        other => other,
418    }
419}
420
421/// `true` when the proxy listener would receive the fallback requests.
422/// Port `0` binds a new port, so it never loops.
423fn loops(bind: SocketAddr, target: SocketAddr) -> bool {
424    let wildcard_bind = bind.ip().is_unspecified();
425    bind.port() != 0
426        && bind.port() == target.port()
427        && (bind.ip() == target.ip() || (wildcard_bind && target.ip().is_loopback()))
428}
429
430/// `true` when Autumn takes the client from the proxy's `X-Forwarded-For`.
431///
432/// The proxy connects from loopback and sends one `X-Forwarded-For` entry.
433/// Autumn trusts all peers when `ranges` is empty. `trusted_hops = 0`
434/// takes the last entry; a larger count skips the client.
435fn trusts_proxy(autumn: &AutumnConfig, target: SocketAddr) -> bool {
436    let policy = &autumn.security.trusted_proxies;
437    let proxy_ip = if target.is_ipv6() {
438        IpAddr::V6(Ipv6Addr::LOCALHOST)
439    } else {
440        IpAddr::V4(Ipv4Addr::LOCALHOST)
441    };
442    policy.trust_forwarded_headers
443        && policy.trusted_hops.map_or_else(
444            || {
445                policy.ranges.is_empty()
446                    || policy
447                        .ranges
448                        .iter()
449                        .filter_map(|range| IpRange::parse(range))
450                        .any(|range| range.contains(proxy_ip))
451            },
452            |hops| hops == 0,
453        )
454}
455
456/// Autumn runs plugin shutdown hooks inside `server.shutdown_timeout_secs`.
457/// Cap the grace so the drain and the aborts fit.
458fn fit_grace(config: &mut PingoraConfig, autumn: &AutumnConfig) {
459    let budget = autumn.server.shutdown_timeout_secs.saturating_mul(1000);
460    let limit = budget.saturating_sub(KILL_WAIT_MS.min(budget / 2)).max(1);
461    if config.shutdown_grace_ms > limit {
462        tracing::warn!(
463            shutdown_grace_ms = config.shutdown_grace_ms,
464            limit_ms = limit,
465            "pingora `shutdown_grace_ms` is more than `server.shutdown_timeout_secs` permits. The plugin uses the limit"
466        );
467        config.shutdown_grace_ms = limit;
468    }
469}
470
471#[cfg(test)]
472mod tests {
473    use super::*;
474
475    fn autumn(host: &str, port: u16) -> AutumnConfig {
476        let mut config = AutumnConfig::default();
477        host.clone_into(&mut config.server.host);
478        config.server.port = port;
479        config
480    }
481
482    #[tokio::test]
483    async fn the_app_target_uses_loopback_for_unspecified_hosts() {
484        let bind: SocketAddr = "127.0.0.1:8080".parse().unwrap();
485        for (host, want) in [
486            ("0.0.0.0", "127.0.0.1:3000"),
487            ("::", "[::1]:3000"),
488            ("localhost", "127.0.0.1:3000"),
489            ("10.1.2.3", "10.1.2.3:3000"),
490        ] {
491            let target = app_target(&autumn(host, 3000), bind).await.unwrap();
492            assert_eq!(target.to_string(), want, "{host}");
493        }
494        let error = app_target(&autumn("no-such-host.invalid", 3000), bind)
495            .await
496            .unwrap_err();
497        assert!(
498            error.to_string().contains("no-such-host.invalid"),
499            "{error}"
500        );
501    }
502
503    #[tokio::test]
504    async fn tls_on_the_app_is_an_error() {
505        let mut config = autumn("127.0.0.1", 3000);
506        config.server.tls =
507            Some(toml::from_str("cert_path = \"c.pem\"\nkey_path = \"k.pem\"").unwrap());
508        let error = app_target(&config, "127.0.0.1:8080".parse().unwrap())
509            .await
510            .unwrap_err();
511        assert!(matches!(error, PingoraError::FallbackTls), "{error}");
512    }
513
514    #[test]
515    fn loops_are_found() {
516        let at = |s: &str| s.parse::<SocketAddr>().unwrap();
517        assert!(loops(at("127.0.0.1:3000"), at("127.0.0.1:3000")));
518        assert!(loops(at("0.0.0.0:3000"), at("127.0.0.1:3000")));
519        assert!(!loops(at("0.0.0.0:8080"), at("127.0.0.1:3000")));
520        assert!(!loops(at("127.0.0.2:3000"), at("127.0.0.1:3000")));
521        assert!(!loops(at("127.0.0.1:0"), at("127.0.0.1:0")));
522    }
523
524    #[test]
525    fn trust_needs_the_switch_and_loopback_or_zero_hops() {
526        let v4: SocketAddr = "127.0.0.1:3000".parse().unwrap();
527        let v6: SocketAddr = "[::1]:3000".parse().unwrap();
528        let mut config = AutumnConfig::default();
529        let policy = &mut config.security.trusted_proxies;
530        policy.trust_forwarded_headers = false;
531        policy.ranges = vec!["127.0.0.1".to_owned()];
532        assert!(!trusts_proxy(&config, v4), "switch off");
533        config.security.trusted_proxies.trust_forwarded_headers = true;
534        assert!(trusts_proxy(&config, v4));
535        assert!(!trusts_proxy(&config, v6), "no ::1 in the ranges");
536        config.security.trusted_proxies.ranges = vec!["10.0.0.0/8".to_owned()];
537        assert!(!trusts_proxy(&config, v4));
538        config.security.trusted_proxies.ranges.clear();
539        assert!(trusts_proxy(&config, v4), "empty ranges trust all peers");
540        config.security.trusted_proxies.trusted_hops = Some(1);
541        assert!(!trusts_proxy(&config, v4), "one hop skips the client");
542        config.security.trusted_proxies.trusted_hops = Some(0);
543        assert!(trusts_proxy(&config, v4));
544    }
545
546    #[test]
547    fn the_grace_fits_the_autumn_budget() {
548        let mut autumn = AutumnConfig::default();
549        autumn.server.shutdown_timeout_secs = 2;
550        let mut config = PingoraConfig::default();
551        fit_grace(&mut config, &autumn);
552        assert_eq!(config.shutdown_grace_ms, 1_500);
553        config.shutdown_grace_ms = 100;
554        fit_grace(&mut config, &autumn);
555        assert_eq!(config.shutdown_grace_ms, 100);
556        autumn.server.shutdown_timeout_secs = 0;
557        fit_grace(&mut config, &autumn);
558        assert_eq!(config.shutdown_grace_ms, 1);
559    }
560}
autumn_plugin_pingora/plugin.rs

autumn_plugin_pingora/
plugin.rs
