Pingora网关

use super::*;
use pingora_cache::{key::HashBinary, CacheKey, CacheMeta, RespCacheable, RespCacheable::*};
use std::time::Duration;

/// The interface to control the HTTP proxy
/// 控制HTTP代理的接口
/// The methods in [ProxyHttp] are filters/callbacks which will be performed on all requests at their
/// particular stage (if applicable).
/// [ProxyHttp] 中的方法是过滤器/回调，它们将在特定阶段对所有请求执行（如果适用）。
/// If any of the filters returns [Result::Err], the request will fail, and the error will be logged.
/// 如果任何过滤器返回错误，请求就会失败，错误会被日志记录
#[cfg_attr(not(doc_async_trait), async_trait)]
pub trait ProxyHttp {
    /// The per request object to share state across the different filters
    /// 每个请求对象在不同的过滤器之间共享状态
    type CTX;

    /// Define how the `ctx` should be created.
    /// 创建上下文对象的方法
    fn new_ctx(&self) -> Self::CTX;

    /// Define where the proxy should send the request to.
    /// 定义代理应将请求发生到哪个后端服务器。
    /// The returned [HttpPeer] contains the information regarding where and how this request should
    /// be forwarded to.
    async fn upstream_peer(
        &self,
        session: &mut Session,
        ctx: &mut Self::CTX,
    ) -> Result<Box<HttpPeer>>;

    /// Set up downstream modules.
    ///
    /// In this phase, users can add or configure [HttpModules] before the server starts up.
    ///
    /// In the default implementation of this method, [ResponseCompressionBuilder] is added
    /// and disabled.
    /// 在默认实现的方法中，返回压缩被添加并设置为无效
    fn init_downstream_modules(&self, modules: &mut HttpModules) {
        // Add disabled downstream compression module by default
        modules.add_module(ResponseCompressionBuilder::enable(0));
    }

    /// Handle the incoming request.
    /// 处理进来的请求
    /// In this phase, users can parse, validate, rate limit, perform access control and/or
    /// return a response for this request.
    /// 在此阶段，用户可以解析、验证、限速、执行访问控制和/或返回此请求的响应。
    /// If the user already sent a response to this request, an `Ok(true)` should be returned so that
    /// the proxy would exit. The proxy continues to the next phases when `Ok(false)` is returned.
    /// 如果用户想对请求做拦截和处理，处理后返回Ok(true)。否则返回Ok(false)，让代理走后续流程
    /// By default this filter does nothing and returns `Ok(false)`.
    async fn request_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool>
    where
        Self::CTX: Send + Sync,
    {
        Ok(false)
    }

    /// Handle the incoming request before any downstream module is executed.
    /// 在执行任何下游模块之前处理传入的请求。
    /// This function is similar to [Self::request_filter()] but execute before any other logic
    /// especially the downstream modules. The main purpose of this function is to provide finer
    /// grained control of behavior of the modules.
    /// 此函数类似于 [Self::request_filter()]，但在任何其他逻辑（尤其是下游模块）之前执行。此函数的主要目的是提供对模块行为的更细粒度的控制。
    /// Note that because this function is executed before any module that might provide access
    /// control or rate limiting, logic should stay in request_filter() if it can in order to be
    /// protected by said modules.
    /// 请注意，由于此函数在任何可能提供访问控制或速率限制的模块之前执行，因此逻辑应该保留在 request_filter() 中（如果可以的话），以便受到所述模块的保护。
    async fn early_request_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> Result<()>
    where
        Self::CTX: Send + Sync,
    {
        Ok(())
    }

    /// Handle the incoming request body.
    /// 处理请求体
    /// This function will be called every time a piece of request body is received. The `body` is
    /// **not the entire request body**.
    /// 每次收到一段请求主体时都会调用此函数。`body` 不是整个请求主体。
    /// The async nature of this function allows to throttle the upload speed and/or executing
    /// heavy computation logic such as WAF rules on offloaded threads without blocking the threads
    /// who process the requests themselves.
    /// 该函数的异步特性允许限制上传速度和/或在卸载线程上执行繁重的计算逻辑（例如 WAF 规则），而不会阻塞处理请求本身的线程。
    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,
    {
        Ok(())
    }

    /// This filter decides if the request is cacheable and what cache backend to use
    /// 此过滤器决定请求是否可缓存以及使用哪个缓存后端
    /// The caller can interact with `Session.cache` to enable caching.
    /// 调用者可以与`Session.cache`交互来启用缓存。
    /// By default this filter does nothing which effectively disables caching.
    // Ideally only session.cache should be modified, TODO: reflect that in this interface
    // 默认情况下，此过滤器不执行任何操作，从而有效地禁用缓存。理想情况下，只应修改 session.cache
    fn request_cache_filter(&self, _session: &mut Session, _ctx: &mut Self::CTX) -> Result<()> {
        Ok(())
    }

    /// This callback generates the cache key
    /// 在此生成缓存key
    /// This callback is called only when cache is enabled for this request
    /// 仅在启用缓存的时候才调用此方法
    /// By default this callback returns a default cache key generated from the request.
    /// 默认情况下，此回调返回从请求生成的默认缓存键。
    fn cache_key_callback(&self, session: &Session, _ctx: &mut Self::CTX) -> Result<CacheKey> {
        let req_header = session.req_header();
        Ok(CacheKey::default(req_header))
    }

    /// This callback is invoked when a cacheable response is ready to be admitted to cache
    /// 当可缓存响应准备好被缓存时，将调用此回调
    fn cache_miss(&self, session: &mut Session, _ctx: &mut Self::CTX) {
        session.cache.cache_miss();
    }

    /// This filter is called after a successful cache lookup and before the cache asset is ready to
    /// be used.
    /// 缓存命中之后，调用此方法
    /// This filter allow the user to log or force expire the asset.
    /// flex purge, other filtering, returns whether asset is should be force expired or not
    /// 在此方法中允许用户记录或者强制资源过期，返回资源是否需要被强制过期
    async fn cache_hit_filter(
        &self,
        _session: &Session,
        _meta: &CacheMeta,
        _ctx: &mut Self::CTX,
    ) -> Result<bool>
    where
        Self::CTX: Send + Sync,
    {
        Ok(false)
    }

    /// Decide if a request should continue to upstream after not being served from cache.
    /// 决定请求在未从缓存中获取服务后是否应继续向上游传输。
    /// returns: Ok(true) if the request should continue, Ok(false) if a response was written by the
    /// callback and the session should be finished, or an error
    /// 如果请求应该继续，则返回 Ok（true）；如果回调写入响应并且会话应该完成，则返回 Ok（false）；或者返回错误
    /// This filter can be used for deferring checks like rate limiting or access control to when they
    /// actually needed after cache miss.
    /// 该过滤器可用于将速率限制或访问控制等检查推迟到缓存未命中后真正需要时进行。
    async fn proxy_upstream_filter(
        &self,
        _session: &mut Session,
        _ctx: &mut Self::CTX,
    ) -> Result<bool>
    where
        Self::CTX: Send + Sync,
    {
        Ok(true)
    }

    /// Decide if the response is cacheable
    /// 决定返回是否需要缓存
    fn response_cache_filter(
        &self,
        _session: &Session,
        _resp: &ResponseHeader,
        _ctx: &mut Self::CTX,
    ) -> Result<RespCacheable> {
        Ok(Uncacheable(NoCacheReason::Custom("default")))
    }

    /// Decide how to generate cache vary key from both request and response
    /// 决定如何根据请求和响应生成缓存变化键
    /// None means no variance is needed.
    /// None表示不需要variance
    fn cache_vary_filter(
        &self,
        _meta: &CacheMeta,
        _ctx: &mut Self::CTX,
        _req: &RequestHeader,
    ) -> Option<HashBinary> {
        // default to None for now to disable vary feature
        None
    }

    /// Decide if the incoming request's condition _fails_ against the cached response.
    /// 确定传入请求的条件是否与缓存的响应不符。
    /// Returning `Ok(true)` means that the response does _not_ match against the condition, and
    /// that the proxy can return `304 Not Modified` downstream.
    /// 返回“Ok(true)”意味着响应与条件不匹配，并且代理可以向下游返回“304 Not Modified”。
    /// An example is a conditional GET request with `If-None-Match: "foobar"`. If the cached
    /// response contains the `ETag: "foobar"`, then the condition fails, and `304 Not Modified`
    /// should be returned. Else, the condition passes which means the full `200 OK` response must
    /// be sent.
    /// 一个例子是带有“If-None-Match: "foobar"”的条件 GET 请求。如果缓存的响应包含“ETag: "foobar"”，则条件失败，并且应返回“304 Not Modified”。否则，条件通过，这意味着必须发送完整的“200 OK”响应。
    fn cache_not_modified_filter(
        &self,
        session: &Session,
        resp: &ResponseHeader,
        _ctx: &mut Self::CTX,
    ) -> Result<bool> {
        Ok(
            pingora_core::protocols::http::conditional_filter::not_modified_filter(
                session.req_header(),
                resp,
            ),
        )
    }

    /// Modify the request before it is sent to the upstream
    /// 在将请求发送到上游之前对其进行修改，仅能修改header
    /// Unlike [Self::request_filter()], this filter allows to change the request headers to send
    /// to the upstream.
    /// 与 [Self::request_filter()] 不同，此过滤器允许更改要发送到上游的请求头。
    async fn upstream_request_filter(
        &self,
        _session: &mut Session,
        _upstream_request: &mut RequestHeader,
        _ctx: &mut Self::CTX,
    ) -> Result<()>
    where
        Self::CTX: Send + Sync,
    {
        Ok(())
    }

    /// Modify the response header from the upstream
    /// 修改上游的响应头
    /// The modification is before caching, so any change here will be stored in the cache if enabled.
    /// 修改是在缓存之前，因此如果启用，此处的任何更改都将存储在缓存中。
    /// Responses served from cache won't trigger this filter. If the cache needed revalidation,
    /// only the 304 from upstream will trigger the filter (though it will be merged into the
    /// cached header, not served directly to downstream).
    /// 来自缓存的响应不会触发此过滤器。如果缓存需要重新验证，则只有来自上游的 304 才会触发过滤器（尽管它将合并到缓存的标头中，而不是直接提供给下游）。
    fn upstream_response_filter(
        &self,
        _session: &mut Session,
        _upstream_response: &mut ResponseHeader,
        _ctx: &mut Self::CTX,
    ) {
    }

    /// Modify the response header before it is send to the downstream
    /// 在发送给下游之前修改响应头
    /// The modification is after caching. This filter is called for all responses including
    /// responses served from cache.
    /// 修改是在缓存之后进行的。所有响应（包括从缓存提供的响应）都会调用此过滤器。
    async fn response_filter(
        &self,
        _session: &mut Session,
        _upstream_response: &mut ResponseHeader,
        _ctx: &mut Self::CTX,
    ) -> Result<()>
    where
        Self::CTX: Send + Sync,
    {
        Ok(())
    }

    /// Similar to [Self::upstream_response_filter()] but for response body
    /// 类似于 [Self::upstream_response_filter()]，但用于响应主体
    /// This function will be called every time a piece of response body is received. The `body` is
    /// **not the entire response body**.
    /// 每次收到一段响应体时都会调用此函数。`body`不是整个响应体。
    fn upstream_response_body_filter(
        &self,
        _session: &mut Session,
        _body: &mut Option<Bytes>,
        _end_of_stream: bool,
        _ctx: &mut Self::CTX,
    ) {
    }

    /// Similar to [Self::upstream_response_filter()] but for response trailers
    /// 类似于 [Self::upstream_response_filter()]，但用于响应尾部
    fn upstream_response_trailer_filter(
        &self,
        _session: &mut Session,
        _upstream_trailers: &mut header::HeaderMap,
        _ctx: &mut Self::CTX,
    ) -> Result<()> {
        Ok(())
    }

    /// Similar to [Self::response_filter()] but for response body chunks
    /// 类似于 [Self::response_filter()]，但用于响应主体块
    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,
    {
        Ok(None)
    }

    /// Similar to [Self::response_filter()] but for response trailers.
    /// Note, returning an Ok(Some(Bytes)) will result in the downstream response
    /// trailers being written to the response body.
    /// 类似于 [Self::response_filter()]，但用于响应尾部。请注意，返回 Ok(Some(Bytes)) 将导致下游响应尾部被写入响应主体。
    /// TODO: make this interface more intuitive
    async fn response_trailer_filter(
        &self,
        _session: &mut Session,
        _upstream_trailers: &mut header::HeaderMap,
        _ctx: &mut Self::CTX,
    ) -> Result<Option<Bytes>>
    where
        Self::CTX: Send + Sync,
    {
        Ok(None)
    }

    /// This filter is called when the entire response is sent to the downstream successfully or
    /// there is a fatal error that terminate the request.
    /// 当整个响应成功发送到下游或出现终止请求的致命错误时，将调用此过滤器。
    /// An error log is already emitted if there is any error. This phase is used for collecting
    /// metrics and sending access logs.
    /// 如果发生任何错误，则会发出错误日志。此阶段用于收集指标和发送访问日志。
    async fn logging(&self, _session: &mut Session, _e: Option<&Error>, _ctx: &mut Self::CTX)
    where
        Self::CTX: Send + Sync,
    {
    }

    /// A value of true means that the log message will be suppressed. The default value is false.
    /// true 表示将抑制日志消息。默认值为 false。
    fn suppress_error_log(&self, _session: &Session, _ctx: &Self::CTX, _error: &Error) -> bool {
        false
    }

    /// This filter is called when there is an error **after** a connection is established (or reused)
    /// to the upstream.
    /// 当与上游建立（或重用）连接后出现错误时，将调用此过滤器。
    fn error_while_proxy(
        &self,
        peer: &HttpPeer,
        session: &mut Session,
        e: Box<Error>,
        _ctx: &mut Self::CTX,
        client_reused: bool,
    ) -> Box<Error> {
        let mut e = e.more_context(format!("Peer: {}", peer));
        // only reused client connections where retry buffer is not truncated
        e.retry
            .decide_reuse(client_reused && !session.as_ref().retry_buffer_truncated());
        e
    }

    /// This filter is called when there is an error in the process of establishing a connection
    /// to the upstream.
    /// 当与上游建立连接的过程中出现错误时，将调用此过滤器。
    /// In this filter the user can decide whether the error is retry-able by marking the error `e`.
    /// 在这个过滤器中，用户可以通过标记错误“e”来决定该错误是否可以重试。
    /// If the error can be retried, [Self::upstream_peer()] will be called again so that the user
    /// can decide whether to send the request to the same upstream or another upstream that is possibly
    /// available.
    /// 如果错误可以重试，则会再次调用 [Self::upstream_peer()]，以便用户可以决定是否将请求发送到同一个上游或可能可用的另一个上游。
    fn fail_to_connect(
        &self,
        _session: &mut Session,
        _peer: &HttpPeer,
        _ctx: &mut Self::CTX,
        e: Box<Error>,
    ) -> Box<Error> {
        e
    }

    /// This filter is called when the request encounters a fatal error.
    /// 当请求遇到致命错误时，将调用此过滤器。
    /// Users may write an error response to the downstream if the downstream is still writable.
    /// 如果下游仍然可写，用户可以向下游写入错误响应。
    /// The response status code of the error response maybe returned for logging purpose.
    /// 错误响应的响应状态代码可能会出于日志记录目的而返回。
    async fn fail_to_proxy(&self, session: &mut Session, e: &Error, _ctx: &mut Self::CTX) -> u16
    where
        Self::CTX: Send + Sync,
    {
        let server_session = session.as_mut();
        let code = match e.etype() {
            HTTPStatus(code) => *code,
            _ => {
                match e.esource() {
                    ErrorSource::Upstream => 502,
                    ErrorSource::Downstream => {
                        match e.etype() {
                            WriteError | ReadError | ConnectionClosed => {
                                /* conn already dead */
                                0
                            }
                            _ => 400,
                        }
                    }
                    ErrorSource::Internal | ErrorSource::Unset => 500,
                }
            }
        };
        if code > 0 {
            server_session.respond_error(code).await
        }
        code
    }

    /// Decide whether should serve stale when encountering an error or during revalidation
    /// 决定在遇到错误时或重新验证期间是否应提供过时的服务
    /// An implementation should follow
    /// <https://datatracker.ietf.org/doc/html/rfc9111#section-4.2.4>
    /// <https://www.rfc-editor.org/rfc/rfc5861#section-4>
    ///
    /// This filter is only called if cache is enabled.
    /// 仅当启用缓存时才会调用此过滤器。
    // 5xx HTTP status will be encoded as ErrorType::HTTPStatus(code)
    fn should_serve_stale(
        &self,
        _session: &mut Session,
        _ctx: &mut Self::CTX,
        error: Option<&Error>, // None when it is called during stale while revalidate
    ) -> bool {
        // A cache MUST NOT generate a stale response unless
        // it is disconnected
        // or doing so is explicitly permitted by the client or origin server
        // (e.g. headers or an out-of-band contract)
        error.map_or(false, |e| e.esource() == &ErrorSource::Upstream)
    }

    /// This filter is called when the request just established or reused a connection to the upstream
    /// 当请求刚刚建立或重新使用与上游的连接时，将调用此过滤器
    /// This filter allows user to log timing and connection related info.
    /// 该过滤器允许用户记录时间和连接相关的信息。
    async fn connected_to_upstream(
        &self,
        _session: &mut Session,
        _reused: bool,
        _peer: &HttpPeer,
        _fd: std::os::unix::io::RawFd,
        _digest: Option<&Digest>,
        _ctx: &mut Self::CTX,
    ) -> Result<()>
    where
        Self::CTX: Send + Sync,
    {
        Ok(())
    }

    /// This callback is invoked every time request related error log needs to be generated
    ///
    /// Users can define what is important to be written about this request via the returned string.
    fn request_summary(&self, session: &Session, _ctx: &Self::CTX) -> String {
        session.as_ref().request_summary()
    }

    /// Whether the request should be used to invalidate(delete) the HTTP cache
    /// 请求是否应该用于使HTTP缓存失效（删除）。
    /// - `true`: this request will be used to invalidate the cache.
    /// true：设置缓存失效
    /// - `false`: this request is a treated as a normal request
    fn is_purge(&self, _session: &Session, _ctx: &Self::CTX) -> bool {
        false
    }

    /// This filter is called after the proxy cache generates the downstream response to the purge
    /// request (to invalidate or delete from the HTTP cache), based on the purge status, which
    /// indicates whether the request succeeded or failed.
    /// 在代理缓存根据清除状态（指示请求是成功还是失败）生成对清除请求的下游响应（使 HTTP 缓存无效或删除）后，将调用此过滤器。
    /// The filter allows the user to modify or replace the generated downstream response.
    /// If the filter returns `Err`, the proxy will instead send a 500 response.
    /// 该过滤器允许用户修改或替换生成的下游响应。如果过滤器返回“Err”，则代理将发送 500 响应。
    fn purge_response_filter(
        &self,
        _session: &Session,
        _ctx: &mut Self::CTX,
        _purge_status: PurgeStatus,
        _purge_response: &mut std::borrow::Cow<'static, ResponseHeader>,
    ) -> Result<()> {
        Ok(())
    }
}

//upstream_peer：每个代理都必须实现，每个请求都会调用到，常用来做负载均衡、灰度发布、请求计数等等
pub struct LB(Arc<LoadBalancer<RoundRobin>>);
#[async_trait]
impl ProxyHttp for LB {
    type CTX = ();
    fn new_ctx(&self) -> Self::CTX {}

    async fn upstream_peer(&self, _session: &mut Session, _ctx: &mut ()) -> Result<Box<HttpPeer>> {
        let upstream = self
            .0
            .select(b"", 256) 
            .unwrap();

        info!("后端服务地址是: {:?}", upstream);

        let peer = Box::new(HttpPeer::new(upstream, true, "".to_string()));
        Ok(peer)
    }
}
//request_filter：解析、验证、请求限速、访问控制等等。
async fn request_filter(&self, session: &mut Session, _ctx: &mut Self::CTX) -> Result<bool> {
        if let Some(token) = session.req_header().headers.get("token") {
            // 业务逻辑，校验token，获得id、name等信息
            let header = session.req_header_mut();
            header.remove_header("token");
            header.insert_header("id","xxx")?;
            header.insert_header("name","yyy")?;
            return Ok(false);
        }
        Ok(true)
    }

// response_body_filter 修改响应体，一般用来做返回转换。json->xml，pb->json,json->yaml等等
fn response_body_filter(
        &self,
        _session: &mut Session,
        body: &mut Option<Bytes>,
        end_of_stream: bool,
        ctx: &mut Self::CTX,
    ) -> Result<Option<std::time::Duration>>
    where
        Self::CTX: Send + Sync,
    {
        // 返回未完成的时候，将流存储在本地
        if let Some(b) = body {
            ctx.buffer.extend(&b[..]);
            // 丢弃原始的body
            b.clear();
        }
        if end_of_stream {
            // 这是最后一块，我们现在可以处理数据了
            let json_body: Resp = serde_json::de::from_slice(&ctx.buffer).unwrap();
            let yaml_body = serde_yaml::to_string(&json_body).unwrap();
            *body = Some(Bytes::copy_from_slice(yaml_body.as_bytes()));
        }

        Ok(None)
    }