pub mod context_provider;
pub mod req;
pub mod retry;
mod utils;
use crate::context_provider::CardinalContextProvider;
use crate::req::ReqCtx;
use crate::retry::RetryState;
use crate::utils::requests::{
compose_upstream_url, execution_context_from_request, parse_origin, rewrite_request_path,
set_upstream_host_headers,
};
use bytes::Bytes;
use cardinal_base::context::CardinalContext;
use cardinal_base::destinations::container::DestinationContainer;
use cardinal_plugins::plugin_executor::CardinalPluginExecutor;
use cardinal_plugins::request_context::RequestContext;
use cardinal_plugins::runner::MiddlewareResult;
use pingora::http::ResponseHeader;
use pingora::prelude::*;
use pingora::protocols::Digest;
use pingora::upstreams::peer::Peer;
use std::sync::Arc;
use std::time::Duration;
use tracing::{debug, error, info, warn};
pub mod pingora {
pub use pingora::*;
}
#[derive(Debug, Clone)]
pub enum HealthCheckStatus {
None,
Ready,
Unavailable {
status_code: u16,
reason: Option<String>,
},
}
#[derive(Clone)]
pub struct StaticContextProvider {
context: Arc<CardinalContext>,
}
impl StaticContextProvider {
pub fn new(context: Arc<CardinalContext>) -> Self {
Self { context }
}
}
impl CardinalContextProvider for StaticContextProvider {
fn resolve(&self, _session: &Session, _ctx: &mut ReqCtx) -> Option<Arc<CardinalContext>> {
Some(self.context.clone())
}
}
#[async_trait::async_trait]
impl CardinalPluginExecutor for StaticContextProvider {}
pub struct CardinalProxy {
provider: Arc<dyn CardinalContextProvider>,
plugin_executor: Arc<dyn CardinalPluginExecutor>,
}
impl CardinalProxy {
pub fn new(context: Arc<CardinalContext>) -> Self {
Self::builder(context).build()
}
pub fn with_provider(
provider: Arc<dyn CardinalContextProvider>,
plugin_executor: Arc<dyn CardinalPluginExecutor>,
) -> Self {
Self {
provider,
plugin_executor,
}
}
pub fn builder(context: Arc<CardinalContext>) -> CardinalProxyBuilder {
CardinalProxyBuilder::new(context)
}
}
pub struct CardinalProxyBuilder {
provider: Arc<dyn CardinalContextProvider>,
plugin_executor: Arc<dyn CardinalPluginExecutor>,
}
impl CardinalProxyBuilder {
pub fn new(context: Arc<CardinalContext>) -> Self {
Self {
provider: Arc::new(StaticContextProvider::new(context.clone())),
plugin_executor: Arc::new(StaticContextProvider::new(context)),
}
}
pub fn from_context_provider(
provider: Arc<dyn CardinalContextProvider>,
plugin_executor: Arc<dyn CardinalPluginExecutor>,
) -> Self {
Self {
provider,
plugin_executor,
}
}
pub fn with_context_provider(
mut self,
provider: Arc<dyn CardinalContextProvider>,
plugin_executor: Arc<dyn CardinalPluginExecutor>,
) -> Self {
self.provider = provider;
self.plugin_executor = plugin_executor;
self
}
pub fn build(self) -> CardinalProxy {
CardinalProxy::with_provider(self.provider, self.plugin_executor)
}
}
#[async_trait::async_trait]
impl ProxyHttp for CardinalProxy {
type CTX = ReqCtx;
fn new_ctx(&self) -> Self::CTX {
self.provider.ctx()
}
async fn early_request_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> Result<()>
where
Self::CTX: Send + Sync,
{
self.provider.early_request_filter(_session, _ctx).await
}
async fn logging(&self, _session: &mut Session, _e: Option<&Error>, ctx: &mut Self::CTX)
where
Self::CTX: Send + Sync,
{
self.provider.logging(_session, _e, ctx);
}
async fn request_body_filter(
&self,
_session: &mut Session,
_body: &mut Option<Bytes>,
_end_of_stream: bool,
_ctx: &mut Self::CTX,
) -> Result<()>
where
Self::CTX: Send + Sync,
{
self.provider
.request_body_filter(_session, _body, _end_of_stream, _ctx)
.await
}
fn response_body_filter(
&self,
_session: &mut Session,
_body: &mut Option<Bytes>,
_end_of_stream: bool,
_ctx: &mut Self::CTX,
) -> Result<Option<Duration>>
where
Self::CTX: Send + Sync,
{
self.provider
.response_body_filter(_session, _body, _end_of_stream, _ctx)
}
async fn request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<bool> {
let path = session.req_header().uri.path().to_string();
info!(%path, "Request received");
match self.provider.health_check(session) {
HealthCheckStatus::None => {}
HealthCheckStatus::Ready => {
debug!(%path, "Health check ready");
let mut resp = ResponseHeader::build(200, None)?;
resp.insert_header("Content-Type", "text/plain")?;
resp.set_content_length("healthy\n".len())?;
session
.write_response_header(Box::new(resp), false)
.await?;
session
.write_response_body(Some(Bytes::from_static(b"healthy\n")), true)
.await?;
return Ok(true);
}
HealthCheckStatus::Unavailable {
status_code,
reason,
} => {
if let Some(reason) = reason {
warn!(%path, status = status_code, reason = %reason, "Health check failed");
} else {
warn!(%path, status = status_code, "Health check failed");
}
let _ = session.respond_error(status_code).await;
return Ok(true);
}
}
let context = match self.provider.resolve(session, ctx) {
Some(ctx) => ctx,
None => {
warn!(%path, "No context found for request host");
let _ = session.respond_error(421).await;
return Ok(true);
}
};
let destination_container = context
.get::<DestinationContainer>()
.await
.map_err(|_| Error::new_str("Destination Container is not present"))?;
let force_path = context.config.server.force_path_parameter;
let backend =
match destination_container.get_backend_for_request(session.req_header(), force_path) {
Some(b) => b,
None => {
warn!(%path, "No matching backend, returning 404");
let _ = session.respond_error(404).await;
return Ok(true);
}
};
let destination_name = backend.destination.name.clone();
let _ = set_upstream_host_headers(session, &backend);
info!(backend_id = %destination_name, "Routing to backend");
rewrite_request_path(session.req_header_mut(), &destination_name, force_path);
let mut request_state = RequestContext::new(
context.clone(),
backend,
execution_context_from_request(session),
self.plugin_executor.clone(),
);
self.provider
.request_context_initialized(session, &mut request_state)
.await;
let plugin_runner = request_state.plugin_runner.clone();
let run_filters = plugin_runner
.run_request_filters(session, &mut request_state)
.await;
let res = match run_filters {
Ok(filter_result) => filter_result,
Err(err) => {
error!(%err, "Error running request filters");
let _ = session.respond_error(500).await;
return Ok(true);
}
};
ctx.set_resolved_request(request_state);
match res {
MiddlewareResult::Continue(resp_headers) => {
ctx.ctx_base
.resolved_request
.as_mut()
.unwrap()
.response_headers = Some(resp_headers);
Ok(false)
}
MiddlewareResult::Responded => Ok(true),
}
}
fn fail_to_connect(
&self,
_session: &mut Session,
_peer: &HttpPeer,
ctx: &mut Self::CTX,
mut e: Box<Error>,
) -> Box<Error> {
let backend_config = ctx.req_unsafe().backend.destination.retry.clone();
if let Some(mut retry_state) = ctx.retry_state.take() {
retry_state.register_attempt();
if retry_state.can_retry() {
e.set_retry(true);
ctx.retry_state = Some(retry_state);
} else {
ctx.retry_state = None;
}
} else if let Some(retry_config) = backend_config {
let mut retry_state = RetryState::from(retry_config);
retry_state.register_attempt();
if retry_state.can_retry() {
e.set_retry(true);
ctx.retry_state = Some(retry_state);
} else {
ctx.retry_state = None;
}
}
e
}
async fn upstream_peer(
&self,
_session: &mut Session,
ctx: &mut Self::CTX,
) -> Result<Box<HttpPeer>> {
if let Some(retry_state) = ctx.retry_state.as_mut() {
if !retry_state.sleep_if_retry_allowed().await {
ctx.retry_state = None;
return Err(Error::new_str("Retry attempts exhausted"));
}
}
let backend = &ctx.req_unsafe().backend;
let (host, port, is_tls) = parse_origin(&backend.destination.url)
.map_err(|_| Error::new_str("Origin could not be parsed "))?;
let hostport = format!("{host}:{port}");
let path_and_query = _session
.req_header()
.uri
.path_and_query()
.map(|pq| pq.as_str())
.unwrap_or("/");
let upstream_url = compose_upstream_url(is_tls, &host, port, path_and_query);
info!(%upstream_url, backend_id = %&backend.destination.name, is_tls, sni = %host, "Forwarding to upstream");
debug!(upstream_origin = %hostport, "Connecting to upstream origin");
let mut peer = HttpPeer::new(&hostport, is_tls, host);
if let Some(opts) = peer.get_mut_peer_options() {
opts.set_http_version(2, 1);
if let Some(timeout) = &backend.destination.timeout {
opts.idle_timeout = timeout
.idle
.as_ref()
.map(|idle| Duration::from_millis(*idle));
opts.write_timeout = timeout
.write
.as_ref()
.map(|idle| Duration::from_millis(*idle));
opts.total_connection_timeout = timeout
.connect
.as_ref()
.map(|idle| Duration::from_millis(*idle));
opts.read_timeout = timeout
.read
.as_ref()
.map(|idle| Duration::from_millis(*idle));
}
}
let peer = Box::new(peer);
Ok(peer)
}
async fn connected_to_upstream(
&self,
_session: &mut Session,
reused: bool,
peer: &HttpPeer,
#[cfg(unix)] _fd: std::os::unix::io::RawFd,
#[cfg(windows)] _sock: std::os::windows::io::RawSocket,
_digest: Option<&Digest>,
ctx: &mut Self::CTX,
) -> Result<()> {
ctx.retry_state = None;
let backend_id = ctx.req_unsafe().backend.destination.name.to_string();
info!(backend_id, reused, peer = %peer, "Connected to upstream");
Ok(())
}
async fn response_filter(
&self,
session: &mut Session,
upstream_response: &mut ResponseHeader,
ctx: &mut Self::CTX,
) -> Result<()> {
if let Some(resp_headers) = ctx.req_unsafe_mut().response_headers.take() {
for (key, val) in resp_headers {
let _ = upstream_response.insert_header(key, val);
}
}
{
{
let runner = {
let req = ctx.req_unsafe_mut();
req.plugin_runner.clone()
};
runner
.run_response_filters(
session,
{
let req = ctx.req_unsafe_mut();
req
},
upstream_response,
)
.await;
}
ctx.set("status", upstream_response.status.as_str());
let req = ctx.req_unsafe_mut();
if !req.cardinal_context.config.server.log_upstream_response {
return Ok(());
}
let status = upstream_response.status.as_u16();
let location = upstream_response
.headers
.get("location")
.and_then(|v| v.to_str().ok())
.map(str::to_string);
let backend_id = &req.backend.destination.name;
match location {
Some(loc) => info!(backend_id, status, location = %loc, "Upstream responded"),
None => info!(backend_id, status, "Upstream responded"),
}
}
Ok(())
}
}