通过 CTX 在阶段之间共享状态

使用 CTX

用户在请求的不同阶段实现的自定义过滤器并不直接相互交互。为了在过滤器之间共享信息和状态，用户可以定义一个 CTX 结构体。每个请求拥有一个单独的 CTX 对象。所有过滤器都能够读取和更新 CTX 对象的成员。CTX 对象将在请求结束时被丢弃。

示例

在以下示例中，代理在 request_filter 阶段解析请求头部，它存储了一个布尔标志，以便在稍后的 upstream_peer 阶段使用该标志来决定将流量路由到哪个服务器。（技术上，头部可以在 upstream_peer 阶段解析，但我们只是为了演示而在更早的阶段进行解析。）

  1. pub struct MyProxy();
  3. pub struct MyCtx {
  4.     beta_user: bool,
  5. }
  7. fn check_beta_user(req: &pingora_http::RequestHeader) -> bool {
  8.     // 一些简单的逻辑来检查用户是否是 beta 用户
  9.     req.headers.get("beta-flag").is_some()
  10. }
  12. #[async_trait]
  13. impl ProxyHttp for MyProxy {
  14.     type CTX = MyCtx;
  15.     fn new_ctx(&self) -> Self::CTX {
  16.         MyCtx { beta_user: false }
  17.     }
  19.     async fn request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<bool> {
  20.         ctx.beta_user = check_beta_user(session.req_header());
  21.         Ok(false)
  22.     }
  24.     async fn upstream_peer(
  25.         &self,
  26.         _session: &mut Session,
  27.         ctx: &mut Self::CTX,
  28.     ) -> Result<Box<HttpPeer>> {
  29.         let addr = if ctx.beta_user {
  30.             info!("I'm a beta user");
  31.             ("1.0.0.1", 443)
  32.         } else {
  33.             ("1.1.1.1", 443)
  34.         };
  36.         let peer = Box::new(HttpPeer::new(addr, true, "one.one.one.one".to_string()));
  37.         Ok(peer)
  38.     }
  39. }

在请求之间共享状态

共享诸如计数器、缓存和其他信息等状态在请求之间是常见的。在 Pingora 中共享请求之间的资源和数据并没有什么特别需要的。可以使用 Arc、static 或任何其他机制。

示例

让我们修改上面的示例，以跟踪 beta 访客的数量以及总访客的数量。计数器可以定义在 MyProxy 结构体本身中，也可以定义为全局变量。由于计数器可以被并发访问，这里使用了 Mutex。

  1. // 全局计数器
  2. static REQ_COUNTER: Mutex<usize> = Mutex::new(0);
  4. pub struct MyProxy {
  5.     // 服务的计数器
  6.     beta_counter: Mutex<usize>, // AtomicUsize 也可以
  7. }
  9. pub struct MyCtx {
  10.     beta_user: bool,
  11. }
  13. fn check_beta_user(req: &pingora_http::RequestHeader) -> bool {
  14.     // 一些简单的逻辑来检查用户是否是 beta
  15.     req.headers.get("beta-flag").is_some()
  16. }
  18. #[async_trait]
  19. impl ProxyHttp for MyProxy {
  20.     type CTX = MyCtx;
  21.     fn new_ctx(&self) -> Self::CTX {
  22.         MyCtx { beta_user: false }
  23.     }
  25.     async fn request_filter(&self, session: &mut Session, ctx: &mut Self::CTX) -> Result<bool> {
  26.         ctx.beta_user = check_beta_user(session.req_header());
  27.         Ok(false)
  28.     }
  30.     async fn upstream_peer(
  31.         &self,
  32.         _session: &mut Session,
  33.         ctx: &mut Self::CTX,
  34.     ) -> Result<Box<HttpPeer>> {
  35.         let mut req_counter = REQ_COUNTER.lock().unwrap();
  36.         *req_counter += 1;
  38.         let addr = if ctx.beta_user {
  39.             let mut beta_count = self.beta_counter.lock().unwrap();
  40.             *beta_count += 1;
  41.             info!("I'm a beta user #{beta_count}");
  42.             ("1.0.0.1", 443)
  43.         } else {
  44.             info!("I'm a user #{req_counter}");
  45.             ("1.1.1.1", 443)
  46.         };
  48.         let peer = Box::new(HttpPeer::new(addr, true, "one.one.one.one".to_string()));
  49.         Ok(peer)
  50.     }
  51. }

完整的示例可以在 pingora-proxy/examples/ctx.rs 下找到。您可以使用 cargo 运行它：

  1. RUST_LOG=INFO cargo run --example ctx