cardinal_proxy/
lib.rs

1pub mod context_provider;
2pub mod req;
3pub mod retry;
4mod utils;
5
6use crate::context_provider::CardinalContextProvider;
7use crate::req::ReqCtx;
8use crate::retry::RetryState;
9use crate::utils::requests::{
10    compose_upstream_url, execution_context_from_request, parse_origin, rewrite_request_path,
11    set_upstream_host_headers,
12};
13use bytes::Bytes;
14use cardinal_base::context::CardinalContext;
15use cardinal_base::destinations::container::DestinationContainer;
16use cardinal_plugins::plugin_executor::CardinalPluginExecutor;
17use cardinal_plugins::request_context::RequestContext;
18use cardinal_plugins::runner::MiddlewareResult;
19use pingora::http::ResponseHeader;
20use pingora::prelude::*;
21use pingora::protocols::Digest;
22use pingora::upstreams::peer::Peer;
23use std::sync::Arc;
24use std::time::Duration;
25use tracing::{debug, error, info, warn};
26
27pub mod pingora {
28    pub use pingora::*;
29}
30
31#[derive(Debug, Clone)]
32pub enum HealthCheckStatus {
33    None,
34    Ready,
35    Unavailable {
36        status_code: u16,
37        reason: Option<String>,
38    },
39}
40
41#[derive(Clone)]
42pub struct StaticContextProvider {
43    context: Arc<CardinalContext>,
44}
45
46impl StaticContextProvider {
47    pub fn new(context: Arc<CardinalContext>) -> Self {
48        Self { context }
49    }
50}
51
52impl CardinalContextProvider for StaticContextProvider {
53    fn resolve(&self, _session: &Session, _ctx: &mut ReqCtx) -> Option<Arc<CardinalContext>> {
54        Some(self.context.clone())
55    }
56}
57
58#[async_trait::async_trait]
59impl CardinalPluginExecutor for StaticContextProvider {}
60
61pub struct CardinalProxy {
62    provider: Arc<dyn CardinalContextProvider>,
63    plugin_executor: Arc<dyn CardinalPluginExecutor>,
64}
65
66impl CardinalProxy {
67    pub fn new(context: Arc<CardinalContext>) -> Self {
68        Self::builder(context).build()
69    }
70
71    pub fn with_provider(
72        provider: Arc<dyn CardinalContextProvider>,
73        plugin_executor: Arc<dyn CardinalPluginExecutor>,
74    ) -> Self {
75        Self {
76            provider,
77            plugin_executor,
78        }
79    }
80
81    pub fn builder(context: Arc<CardinalContext>) -> CardinalProxyBuilder {
82        CardinalProxyBuilder::new(context)
83    }
84}
85
86pub struct CardinalProxyBuilder {
87    provider: Arc<dyn CardinalContextProvider>,
88    plugin_executor: Arc<dyn CardinalPluginExecutor>,
89}
90
91impl CardinalProxyBuilder {
92    pub fn new(context: Arc<CardinalContext>) -> Self {
93        Self {
94            provider: Arc::new(StaticContextProvider::new(context.clone())),
95            plugin_executor: Arc::new(StaticContextProvider::new(context)),
96        }
97    }
98
99    pub fn from_context_provider(
100        provider: Arc<dyn CardinalContextProvider>,
101        plugin_executor: Arc<dyn CardinalPluginExecutor>,
102    ) -> Self {
103        Self {
104            provider,
105            plugin_executor,
106        }
107    }
108
109    pub fn with_context_provider(
110        mut self,
111        provider: Arc<dyn CardinalContextProvider>,
112        plugin_executor: Arc<dyn CardinalPluginExecutor>,
113    ) -> Self {
114        self.provider = provider;
115        self.plugin_executor = plugin_executor;
116        self
117    }
118
119    pub fn build(self) -> CardinalProxy {
120        CardinalProxy::with_provider(self.provider, self.plugin_executor)
121    }
122}
123
124#[async_trait::async_trait]
125impl ProxyHttp for CardinalProxy {
126    type CTX = ReqCtx;
127
128    fn new_ctx(&self) -> Self::CTX {
129        self.provider.ctx()
130    }
131
132    async fn early_request_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> Result<()>
133    where
134        Self::CTX: Send + Sync,
135    {
136        self.provider.early_request_filter(_session, _ctx).await
137    }
138
139    async fn logging(&self, _session: &mut Session, _e: Option<&Error>, ctx: &mut Self::CTX)
140    where
141        Self::CTX: Send + Sync,
142    {
143        self.provider.logging(_session, _e, ctx);
144    }
145
146    async fn request_body_filter(
147        &self,
148        _session: &mut Session,
149        _body: &mut Option<Bytes>,
150        _end_of_stream: bool,
151        _ctx: &mut Self::CTX,
152    ) -> Result<()>
153    where
154        Self::CTX: Send + Sync,
155    {
156        self.provider
157            .request_body_filter(_session, _body, _end_of_stream, _ctx)
158            .await
159    }
160
161    fn response_body_filter(
162        &self,
163        _session: &mut Session,
164        _body: &mut Option<Bytes>,
165        _end_of_stream: bool,
166        _ctx: &mut Self::CTX,
167    ) -> Result<Option<Duration>>
168    where
169        Self::CTX: Send + Sync,
170    {
171        self.provider
172            .response_body_filter(_session, _body, _end_of_stream, _ctx)
173    }
174
175    async fn request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<bool> {
176        let path = session.req_header().uri.path().to_string();
177        info!(%path, "Request received");
178
179        match self.provider.health_check(session) {
180            HealthCheckStatus::None => {}
181            HealthCheckStatus::Ready => {
182                debug!(%path, "Health check ready");
183                // Build a 200 OK header
184                let mut resp = ResponseHeader::build(200, None)?;
185                resp.insert_header("Content-Type", "text/plain")?;
186                resp.set_content_length("healthy\n".len())?;
187
188                // Send header + body to the client
189                session
190                    .write_response_header(Box::new(resp), /*end_of_stream*/ false)
191                    .await?;
192                session
193                    .write_response_body(Some(Bytes::from_static(b"healthy\n")), /*end*/ true)
194                    .await?;
195
196                // Returning Ok(true) means "handled", stop further processing.
197                return Ok(true);
198            }
199            HealthCheckStatus::Unavailable {
200                status_code,
201                reason,
202            } => {
203                if let Some(reason) = reason {
204                    warn!(%path, status = status_code, reason = %reason, "Health check failed");
205                } else {
206                    warn!(%path, status = status_code, "Health check failed");
207                }
208                let _ = session.respond_error(status_code).await;
209                return Ok(true);
210            }
211        }
212
213        let context = match self.provider.resolve(session, ctx) {
214            Some(ctx) => ctx,
215            None => {
216                warn!(%path, "No context found for request host");
217                let _ = session.respond_error(421).await;
218                return Ok(true);
219            }
220        };
221
222        let destination_container = context
223            .get::<DestinationContainer>()
224            .await
225            .map_err(|_| Error::new_str("Destination Container is not present"))?;
226
227        let force_path = context.config.server.force_path_parameter;
228        let backend =
229            match destination_container.get_backend_for_request(session.req_header(), force_path) {
230                Some(b) => b,
231                None => {
232                    warn!(%path, "No matching backend, returning 404");
233                    let _ = session.respond_error(404).await;
234                    return Ok(true);
235                }
236            };
237
238        let destination_name = backend.destination.name.clone();
239        let _ = set_upstream_host_headers(session, &backend);
240        info!(backend_id = %destination_name, "Routing to backend");
241
242        rewrite_request_path(session.req_header_mut(), &destination_name, force_path);
243
244        let mut request_state = RequestContext::new(
245            context.clone(),
246            backend,
247            execution_context_from_request(session),
248            self.plugin_executor.clone(),
249        );
250
251        let plugin_runner = request_state.plugin_runner.clone();
252
253        let run_filters = plugin_runner
254            .run_request_filters(session, &mut request_state)
255            .await;
256
257        let res = match run_filters {
258            Ok(filter_result) => filter_result,
259            Err(err) => {
260                error!(%err, "Error running request filters");
261                let _ = session.respond_error(500).await;
262                return Ok(true);
263            }
264        };
265
266        ctx.set_resolved_request(request_state);
267
268        match res {
269            MiddlewareResult::Continue(resp_headers) => {
270                ctx.ctx_base
271                    .resolved_request
272                    .as_mut()
273                    .unwrap()
274                    .response_headers = Some(resp_headers);
275
276                Ok(false)
277            }
278            MiddlewareResult::Responded => Ok(true),
279        }
280    }
281
282    fn fail_to_connect(
283        &self,
284        _session: &mut Session,
285        _peer: &HttpPeer,
286        ctx: &mut Self::CTX,
287        mut e: Box<Error>,
288    ) -> Box<Error> {
289        let backend_config = ctx.req_unsafe().backend.destination.retry.clone();
290        if let Some(mut retry_state) = ctx.retry_state.take() {
291            retry_state.register_attempt();
292            if retry_state.can_retry() {
293                e.set_retry(true);
294                ctx.retry_state = Some(retry_state);
295            } else {
296                ctx.retry_state = None;
297            }
298        } else if let Some(retry_config) = backend_config {
299            let mut retry_state = RetryState::from(retry_config);
300            retry_state.register_attempt();
301            if retry_state.can_retry() {
302                e.set_retry(true);
303                ctx.retry_state = Some(retry_state);
304            } else {
305                ctx.retry_state = None;
306            }
307        }
308
309        e
310    }
311
312    async fn upstream_peer(
313        &self,
314        _session: &mut Session,
315        ctx: &mut Self::CTX,
316    ) -> Result<Box<HttpPeer>> {
317        if let Some(retry_state) = ctx.retry_state.as_mut() {
318            if !retry_state.sleep_if_retry_allowed().await {
319                ctx.retry_state = None;
320                return Err(Error::new_str("Retry attempts exhausted"));
321            }
322        }
323
324        let backend = &ctx.req_unsafe().backend;
325        // Determine origin parts for TLS and SNI
326        let (host, port, is_tls) = parse_origin(&backend.destination.url)
327            .map_err(|_| Error::new_str("Origin could not be parsed "))?;
328        let hostport = format!("{host}:{port}");
329
330        // Compose full upstream URL for logging with normalized scheme
331        let path_and_query = _session
332            .req_header()
333            .uri
334            .path_and_query()
335            .map(|pq| pq.as_str())
336            .unwrap_or("/");
337        let upstream_url = compose_upstream_url(is_tls, &host, port, path_and_query);
338
339        info!(%upstream_url, backend_id = %&backend.destination.name, is_tls, sni = %host, "Forwarding to upstream");
340        debug!(upstream_origin = %hostport, "Connecting to upstream origin");
341
342        let mut peer = HttpPeer::new(&hostport, is_tls, host);
343        if let Some(opts) = peer.get_mut_peer_options() {
344            // Allow both HTTP/1.1 and HTTP/2 so plain HTTP backends keep working.
345            opts.set_http_version(2, 1);
346            if let Some(timeout) = &backend.destination.timeout {
347                opts.idle_timeout = timeout
348                    .idle
349                    .as_ref()
350                    .map(|idle| Duration::from_millis(*idle));
351                opts.write_timeout = timeout
352                    .write
353                    .as_ref()
354                    .map(|idle| Duration::from_millis(*idle));
355                opts.total_connection_timeout = timeout
356                    .connect
357                    .as_ref()
358                    .map(|idle| Duration::from_millis(*idle));
359                opts.read_timeout = timeout
360                    .read
361                    .as_ref()
362                    .map(|idle| Duration::from_millis(*idle));
363            }
364        }
365        let peer = Box::new(peer);
366        Ok(peer)
367    }
368
369    async fn connected_to_upstream(
370        &self,
371        _session: &mut Session,
372        reused: bool,
373        peer: &HttpPeer,
374        #[cfg(unix)] _fd: std::os::unix::io::RawFd,
375        #[cfg(windows)] _sock: std::os::windows::io::RawSocket,
376        _digest: Option<&Digest>,
377        ctx: &mut Self::CTX,
378    ) -> Result<()> {
379        ctx.retry_state = None;
380        let backend_id = ctx.req_unsafe().backend.destination.name.to_string();
381
382        info!(backend_id, reused, peer = %peer, "Connected to upstream");
383        Ok(())
384    }
385
386    async fn response_filter(
387        &self,
388        session: &mut Session,
389        upstream_response: &mut ResponseHeader,
390        ctx: &mut Self::CTX,
391    ) -> Result<()> {
392        if let Some(resp_headers) = ctx.req_unsafe_mut().response_headers.take() {
393            for (key, val) in resp_headers {
394                let _ = upstream_response.insert_header(key, val);
395            }
396        }
397
398        {
399            // Run response filters first
400            {
401                let runner = {
402                    let req = ctx.req_unsafe_mut();
403                    req.plugin_runner.clone()
404                };
405
406                runner
407                    .run_response_filters(
408                        session,
409                        {
410                            let req = ctx.req_unsafe_mut();
411                            req
412                        },
413                        upstream_response,
414                    )
415                    .await;
416            }
417
418            ctx.set("status", upstream_response.status.as_str());
419
420            // Safe to get another mutable reference now
421            let req = ctx.req_unsafe_mut();
422
423            if !req.cardinal_context.config.server.log_upstream_response {
424                return Ok(());
425            }
426
427            let status = upstream_response.status.as_u16();
428            let location = upstream_response
429                .headers
430                .get("location")
431                .and_then(|v| v.to_str().ok())
432                .map(str::to_string);
433            let backend_id = &req.backend.destination.name;
434
435            match location {
436                Some(loc) => info!(backend_id, status, location = %loc, "Upstream responded"),
437                None => info!(backend_id, status, "Upstream responded"),
438            }
439        }
440
441        Ok(())
442    }
443}
cardinal_proxy/lib.rs

cardinal_proxy/
lib.rs
