praxis_protocol/http/pingora/handler/
with_body.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2024 Praxis Contributors
3
4//! Pingora HTTP handler with body filter hooks enabled.
5//!
6//! [`PingoraHttpHandler`] is the full-featured `ProxyHttp` implementation
7//! used when the pipeline's [`BodyCapabilities`] declare request or
8//! response body access. It delegates each Pingora lifecycle hook to
9//! the corresponding submodule and enables Pingora's compression
10//! module when a compression filter is configured.
11//!
12//! [`BodyCapabilities`]: praxis_filter::body::BodyCapabilities
13
14use std::{sync::Arc, time::Duration};
15
16use arc_swap::ArcSwap;
17use async_trait::async_trait;
18use bytes::Bytes;
19use pingora_core::{
20    Result,
21    modules::http::{HttpModules, compression::ResponseCompressionBuilder},
22    upstreams::peer::HttpPeer,
23};
24use pingora_proxy::{FailToProxy, ProxyHttp, Session};
25use praxis_filter::{CompressionConfig, FilterPipeline};
26use tokio::sync::Semaphore;
27use tracing::debug;
28
29use super::{
30    adjust_compression, emit_request_metrics, fail_to_proxy, handle_connect_failure, hop_by_hop::RemoveHeader as _,
31    logging_cleanup, record_passive_health, request_body_filter, request_filter, response_body_filter, response_filter,
32    upstream_peer, upstream_request, via,
33};
34use crate::http::pingora::{context::PingoraRequestCtx, metrics};
35
36// -----------------------------------------------------------------------------
37// PingoraHttpHandler
38// -----------------------------------------------------------------------------
39
40/// Pingora HTTP handler that overrides body filter hooks.
41///
42/// Used when the pipeline contains filters that declare
43/// body access via [`BodyAccess`].
44///
45/// The pipeline is held behind [`ArcSwap`] so it can be
46/// atomically replaced by hot config reload without
47/// disrupting in-flight requests.
48///
49/// ```ignore
50/// // Requires a `FilterPipeline` and Pingora server runtime.
51/// use std::sync::Arc;
52///
53/// use arc_swap::ArcSwap;
54/// use praxis_protocol::http::pingora::handler::PingoraHttpHandler;
55///
56/// let handler = PingoraHttpHandler::new(
57///     Arc::new(ArcSwap::from_pointee(pipeline)),
58///     None,
59///     None,
60///     ::metrics::SharedString::const_str("http"),
61/// );
62/// ```
63///
64/// [`BodyAccess`]: praxis_filter::BodyAccess
65/// [`ArcSwap`]: arc_swap::ArcSwap
66pub struct PingoraHttpHandler {
67    /// Compression configuration snapshot for module registration.
68    ///
69    /// Used only by [`init_downstream_modules`] to register the
70    /// compression module at startup. Per-request compression
71    /// levels are read from the live pipeline via [`ArcSwap`]
72    /// so that hot-reload updates take effect immediately.
73    ///
74    /// Module registration itself is one-shot in Pingora;
75    /// adding compression to a listener that had none at
76    /// startup requires a restart.
77    ///
78    /// [`init_downstream_modules`]: Self::init_downstream_modules
79    /// [`ArcSwap`]: arc_swap::ArcSwap
80    compression: Option<CompressionConfig>,
81
82    /// Per-listener connection semaphore for max connections.
83    connection_semaphore: Option<Arc<Semaphore>>,
84
85    /// Per-listener downstream read timeout.
86    downstream_read_timeout: Option<Duration>,
87
88    /// Listener name for connection metrics.
89    listener_name: ::metrics::SharedString,
90
91    /// Swappable filter pipeline.
92    pipeline: Arc<ArcSwap<FilterPipeline>>,
93}
94
95impl PingoraHttpHandler {
96    /// Create a handler with body filter support.
97    pub(super) fn new(
98        pipeline: Arc<ArcSwap<FilterPipeline>>,
99        downstream_read_timeout: Option<Duration>,
100        connection_semaphore: Option<Arc<Semaphore>>,
101        listener_name: ::metrics::SharedString,
102    ) -> Self {
103        let compression = pipeline.load().compression_config().cloned();
104        Self {
105            compression,
106            connection_semaphore,
107            downstream_read_timeout,
108            listener_name,
109            pipeline,
110        }
111    }
112}
113
114#[async_trait]
115impl ProxyHttp for PingoraHttpHandler {
116    type CTX = PingoraRequestCtx;
117
118    fn new_ctx(&self) -> Self::CTX {
119        PingoraRequestCtx::default()
120    }
121
122    /// Registers Pingora's compression module when compression is
123    /// configured. Otherwise skips module registration to avoid
124    /// per-request `Box` allocation overhead.
125    fn init_downstream_modules(&self, modules: &mut HttpModules) {
126        if let Some(cfg) = &self.compression {
127            debug!(level = cfg.default_level, "registering compression module");
128            modules.add_module(ResponseCompressionBuilder::enable(cfg.default_level));
129        }
130    }
131
132    #[expect(clippy::cast_possible_truncation, reason = "millis fit u64")]
133    async fn early_request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<()>
134    where
135        Self::CTX: Send + Sync,
136    {
137        if praxis_core::memory::is_exceeded() {
138            metrics::record_overload_reject(metrics::OVERLOAD_REASON_MEMORY);
139            return reject_503(session, "5", "memory pressure exceeded").await;
140        }
141
142        let (exceeded, permit) = crate::connections::try_acquire_global();
143        ctx._global_connection_permit = permit;
144        if exceeded {
145            metrics::record_overload_reject(metrics::OVERLOAD_REASON_GLOBAL_CONNECTIONS);
146            return reject_503(session, "1", "global max connections exceeded").await;
147        }
148
149        if let Some(sem) = &self.connection_semaphore {
150            if let Ok(permit) = Arc::clone(sem).try_acquire_owned() {
151                ctx._connection_permit = Some(permit);
152            } else {
153                metrics::record_overload_reject(metrics::OVERLOAD_REASON_LISTENER_CONNECTIONS);
154                return reject_503(session, "1", "max connections exceeded").await;
155            }
156        }
157
158        ctx._active_connection = Some(metrics::ActiveConnectionGuard::acquire(self.listener_name.clone()));
159
160        if let Some(timeout) = self.downstream_read_timeout {
161            debug!(
162                timeout_ms = timeout.as_millis() as u64,
163                "applying downstream read timeout"
164            );
165            session.set_read_timeout(Some(timeout));
166        }
167        Ok(())
168    }
169
170    async fn request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<bool> {
171        let pipeline = ctx.pin_pipeline(&self.pipeline);
172        request_filter::execute(&pipeline, session, ctx).await
173    }
174
175    async fn request_body_filter(
176        &self,
177        session: &mut Session,
178        body: &mut Option<Bytes>,
179        end_of_stream: bool,
180        ctx: &mut Self::CTX,
181    ) -> Result<()>
182    where
183        Self::CTX: Send + Sync,
184    {
185        let pipeline = ctx.pipeline(&self.pipeline);
186        request_body_filter::execute(&pipeline, session, body, end_of_stream, ctx).await
187    }
188
189    fn response_body_filter(
190        &self,
191        _session: &mut Session,
192        body: &mut Option<Bytes>,
193        end_of_stream: bool,
194        ctx: &mut Self::CTX,
195    ) -> Result<Option<Duration>>
196    where
197        Self::CTX: Send + Sync,
198    {
199        let pipeline = ctx.pipeline(&self.pipeline);
200        response_body_filter::execute(&pipeline, body, end_of_stream, ctx)
201    }
202
203    fn fail_to_connect(
204        &self,
205        _session: &mut Session,
206        _peer: &HttpPeer,
207        ctx: &mut Self::CTX,
208        e: Box<pingora_core::Error>,
209    ) -> Box<pingora_core::Error> {
210        handle_connect_failure(ctx, e)
211    }
212
213    async fn fail_to_proxy(&self, session: &mut Session, e: &pingora_core::Error, ctx: &mut Self::CTX) -> FailToProxy
214    where
215        Self::CTX: Send + Sync,
216    {
217        fail_to_proxy::execute(session, e, ctx).await
218    }
219
220    async fn connected_to_upstream(
221        &self,
222        _session: &mut Session,
223        reused: bool,
224        _peer: &HttpPeer,
225        #[cfg(unix)] _fd: std::os::unix::io::RawFd,
226        #[cfg(windows)] _sock: std::os::windows::io::RawSocket,
227        _digest: Option<&pingora_core::protocols::Digest>,
228        ctx: &mut Self::CTX,
229    ) -> Result<()>
230    where
231        Self::CTX: Send + Sync,
232    {
233        let cluster = ctx.metrics_cluster_shared.clone().unwrap_or_else(metrics::cluster_none);
234        if !reused && let Some(start) = ctx.upstream_connect_start.take() {
235            metrics::record_upstream_connect_duration(cluster.clone(), start.elapsed().as_secs_f64());
236        }
237        if ctx.retries > 0 {
238            metrics::record_upstream_retry(cluster, metrics::RETRY_RESULT_SUCCESS);
239        }
240        Ok(())
241    }
242
243    async fn upstream_request_filter(
244        &self,
245        session: &mut Session,
246        upstream_request: &mut pingora_http::RequestHeader,
247        ctx: &mut Self::CTX,
248    ) -> Result<()>
249    where
250        Self::CTX: Send + Sync,
251    {
252        let is_upgrade = session.is_upgrade_req();
253        upstream_request::strip_hop_by_hop(upstream_request, is_upgrade);
254        upstream_request.strip_reserved_internal();
255        upstream_request::apply_rewritten_path(upstream_request, ctx)?;
256        upstream_request::apply_mutated_content_length(upstream_request, ctx);
257        let client_ver = ctx.client_http_version.unwrap_or(http::Version::HTTP_11);
258        via::append_request_via(upstream_request, client_ver);
259        Ok(())
260    }
261
262    async fn response_filter(
263        &self,
264        session: &mut Session,
265        upstream_response: &mut pingora_http::ResponseHeader,
266        ctx: &mut Self::CTX,
267    ) -> Result<()>
268    where
269        Self::CTX: Send + Sync,
270    {
271        let pipeline = ctx.pipeline(&self.pipeline);
272        let result = response_filter::execute(&pipeline, upstream_response, ctx).await;
273        if result.is_ok() {
274            let client_ver = ctx.client_http_version.unwrap_or(http::Version::HTTP_11);
275            via::append_response_via(upstream_response, client_ver);
276            adjust_compression(session, upstream_response, pipeline.compression_config());
277        }
278        result
279    }
280
281    async fn upstream_peer(&self, _session: &mut Session, ctx: &mut Self::CTX) -> Result<Box<HttpPeer>> {
282        upstream_peer::execute(ctx).await
283    }
284
285    async fn logging(&self, session: &mut Session, e: Option<&pingora_core::Error>, ctx: &mut Self::CTX) {
286        let pipeline = ctx.pipeline(&self.pipeline);
287        emit_request_metrics(session, ctx);
288        record_passive_health(&pipeline, e, ctx);
289        logging_cleanup(&pipeline, ctx).await;
290    }
291}
292
293// -----------------------------------------------------------------------------
294// Utilities
295// -----------------------------------------------------------------------------
296
297/// Write a 503 response with `Retry-After` and return the corresponding error.
298async fn reject_503(session: &mut Session, retry_after: &'static str, reason: &'static str) -> Result<()> {
299    tracing::warn!(reason, "rejecting request");
300    let mut header = pingora_http::ResponseHeader::build(503, None)?;
301    header.append_header("Retry-After", retry_after)?;
302    session.write_response_header(Box::new(header), true).await?;
303    Err(pingora_core::Error::explain(
304        pingora_core::ErrorType::HTTPStatus(503),
305        reason,
306    ))
307}
praxis_protocol/http/pingora/handler/with_body.rs

praxis_protocol/http/pingora/handler/
with_body.rs
