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::{Instrument as _, 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, record_response_span_attributes, request_body_filter, request_filter,
32    response_body_filter, response_filter, 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        let span = ctx.request_span.clone();
187        request_body_filter::execute(&pipeline, session, body, end_of_stream, ctx)
188            .instrument(span)
189            .await
190    }
191
192    fn response_body_filter(
193        &self,
194        _session: &mut Session,
195        body: &mut Option<Bytes>,
196        end_of_stream: bool,
197        ctx: &mut Self::CTX,
198    ) -> Result<Option<Duration>>
199    where
200        Self::CTX: Send + Sync,
201    {
202        let span = ctx.request_span.clone();
203        let _entered = span.enter();
204        let pipeline = ctx.pipeline(&self.pipeline);
205        response_body_filter::execute(&pipeline, body, end_of_stream, ctx)
206    }
207
208    fn fail_to_connect(
209        &self,
210        _session: &mut Session,
211        _peer: &HttpPeer,
212        ctx: &mut Self::CTX,
213        e: Box<pingora_core::Error>,
214    ) -> Box<pingora_core::Error> {
215        let span = ctx.request_span.clone();
216        let _entered = span.enter();
217        handle_connect_failure(ctx, e)
218    }
219
220    async fn fail_to_proxy(&self, session: &mut Session, e: &pingora_core::Error, ctx: &mut Self::CTX) -> FailToProxy
221    where
222        Self::CTX: Send + Sync,
223    {
224        let span = ctx.request_span.clone();
225        fail_to_proxy::execute(session, e, ctx).instrument(span).await
226    }
227
228    async fn connected_to_upstream(
229        &self,
230        _session: &mut Session,
231        reused: bool,
232        _peer: &HttpPeer,
233        #[cfg(unix)] _fd: std::os::unix::io::RawFd,
234        #[cfg(windows)] _sock: std::os::windows::io::RawSocket,
235        _digest: Option<&pingora_core::protocols::Digest>,
236        ctx: &mut Self::CTX,
237    ) -> Result<()>
238    where
239        Self::CTX: Send + Sync,
240    {
241        let span = ctx.request_span.clone();
242        let _entered = span.enter();
243        let cluster = ctx.metrics_cluster_shared.clone().unwrap_or_else(metrics::cluster_none);
244        if !reused && let Some(start) = ctx.upstream_connect_start.take() {
245            metrics::record_upstream_connect_duration(cluster.clone(), start.elapsed().as_secs_f64());
246        }
247        if ctx.retries > 0 {
248            metrics::record_upstream_retry(cluster, metrics::RETRY_RESULT_SUCCESS);
249        }
250        Ok(())
251    }
252
253    async fn upstream_request_filter(
254        &self,
255        session: &mut Session,
256        upstream_request: &mut pingora_http::RequestHeader,
257        ctx: &mut Self::CTX,
258    ) -> Result<()>
259    where
260        Self::CTX: Send + Sync,
261    {
262        let span = ctx.request_span.clone();
263        let _entered = span.enter();
264        let is_upgrade = session.is_upgrade_req();
265        upstream_request::strip_hop_by_hop(upstream_request, is_upgrade);
266        upstream_request.strip_reserved_internal();
267        upstream_request::apply_rewritten_path(upstream_request, ctx)?;
268        upstream_request::apply_mutated_content_length(upstream_request, ctx);
269        let client_ver = ctx.client_http_version.unwrap_or(http::Version::HTTP_11);
270        via::append_request_via(upstream_request, client_ver);
271        Ok(())
272    }
273
274    async fn response_filter(
275        &self,
276        session: &mut Session,
277        upstream_response: &mut pingora_http::ResponseHeader,
278        ctx: &mut Self::CTX,
279    ) -> Result<()>
280    where
281        Self::CTX: Send + Sync,
282    {
283        let pipeline = ctx.pipeline(&self.pipeline);
284        let span = ctx.request_span.clone();
285        let result = response_filter::execute(&pipeline, upstream_response, ctx)
286            .instrument(span)
287            .await;
288        if result.is_ok() {
289            let client_ver = ctx.client_http_version.unwrap_or(http::Version::HTTP_11);
290            via::append_response_via(upstream_response, client_ver);
291            adjust_compression(session, upstream_response, pipeline.compression_config());
292        }
293        result
294    }
295
296    async fn upstream_peer(&self, _session: &mut Session, ctx: &mut Self::CTX) -> Result<Box<HttpPeer>> {
297        let span = ctx.request_span.clone();
298        upstream_peer::execute(ctx).instrument(span).await
299    }
300
301    async fn logging(&self, session: &mut Session, e: Option<&pingora_core::Error>, ctx: &mut Self::CTX) {
302        record_response_span_attributes(session, ctx);
303        let span = std::mem::replace(&mut ctx.request_span, tracing::Span::none());
304        async {
305            let pipeline = ctx.pipeline(&self.pipeline);
306            emit_request_metrics(session, ctx);
307            record_passive_health(&pipeline, e, ctx);
308            logging_cleanup(&pipeline, ctx).await;
309        }
310        .instrument(span)
311        .await;
312    }
313}
314
315// -----------------------------------------------------------------------------
316// Utilities
317// -----------------------------------------------------------------------------
318
319/// Write a 503 response with `Retry-After` and return the corresponding error.
320async fn reject_503(session: &mut Session, retry_after: &'static str, reason: &'static str) -> Result<()> {
321    tracing::warn!(reason, "rejecting request");
322    let mut header = pingora_http::ResponseHeader::build(503, None)?;
323    header.append_header("Retry-After", retry_after)?;
324    session.write_response_header(Box::new(header), true).await?;
325    Err(pingora_core::Error::explain(
326        pingora_core::ErrorType::HTTPStatus(503),
327        reason,
328    ))
329}
praxis_protocol/http/pingora/handler/with_body.rs

praxis_protocol/http/pingora/handler/
with_body.rs
