Repository: cloudflare/pingora (language: rust)

Issue:
Replace structopt by clap
## What is the problem your feature solves, or the need it fulfills?

structopt crate is unmaintained and superseded by clap v3 ([Maintenance note](https://docs.rs/structopt/latest/structopt/#maintenance)). It has bugs that clap fixed three years ago before even releasing them (https://github.com/TeXitoi/structopt/issues/539, https://github.com/clap-rs/clap/issues/2527). This creates issues when expanding Pingora’s [Opt structure](https://docs.rs/pingora-core/0.1.1/pingora_core/server/configuration/struct.Opt.html).

## Describe the solution you'd like

Pingora should use a current clap version.

## Describe alternatives you've considered

I implemented the following hack – adding a dummy field to the app’s options only to overwrite application description that has been overwritten by `Opt` above:

```rust
/// My description
#[derive(Debug, StructOpt)]
struct MyOpt {
    …

    #[structopt(flatten)]
    server: Opt,

    #[allow(dead_code)]
    #[structopt(flatten)]
    dummy: StructOptDummy,
}

#[derive(Debug, StructOpt)]
#[structopt(about = "My description", long_about = "My description")]
struct StructOptDummy {}
```


Discussion / hints:
No promises, but I'll try to make this happen
> No promises, but I'll try to make this happen

It isn’t complicated, structopt to clap is almost search&replace – I could do it if I knew that the pull request will be accepted rather than stuck because nobody has time to review it.
Yeah, the only hurdle is to get buy in from the team since it's a change a pretty central place. If you're willing to put in the work/pr. That will make it easier for me to champion 💪
@johnhurt Ok, then there is #239 now.

Fix patch:
diff --git a/.bleep b/.bleep
index 186f68f8e..5b3f05fd5 100644
--- a/.bleep
+++ b/.bleep
@@ -1,1 +1,1 @@
-2c9d4c55853235e908a1acd20454ebe7b979d246
\ No newline at end of file
+719b4dcafd471ecfa9e4ade3abfb217929afddc6
\ No newline at end of file
diff --git a/pingora-cache/src/lib.rs b/pingora-cache/src/lib.rs
index 74e4870f6..563727cfd 100644
--- a/pingora-cache/src/lib.rs
+++ b/pingora-cache/src/lib.rs
@@ -21,7 +21,7 @@ use key::{CacheHashKey, HashBinary};
 use lock::WritePermit;
 use pingora_error::Result;
 use pingora_http::ResponseHeader;
-use std::time::{Duration, SystemTime};
+use std::time::{Duration, Instant, SystemTime};
 use trace::CacheTraceCTX;
 
 pub mod cache_control;
@@ -222,6 +222,8 @@ struct HttpCacheInner {
     pub lock: Option<Locked>, // TODO: these 3 fields should come in 1 sub struct
     pub cache_lock: Option<&'static CacheLock>,
     pub lock_duration: Option<Duration>,
+    // time spent in cache lookup and reading the header
+    pub lookup_duration: Option<Duration>,
     pub traces: trace::CacheTraceCTX,
 }
 
@@ -369,6 +371,7 @@ impl HttpCache {
                     lock: None,
                     cache_lock,
                     lock_duration: None,
+                    lookup_duration: None,
                     traces: CacheTraceCTX::new(),
                 }));
             }
@@ -896,7 +899,15 @@ impl HttpCache {
                 let inner = self.inner_mut();
                 let mut span = inner.traces.child("lookup");
                 let key = inner.key.as_ref().unwrap(); // safe, this phase should have cache key
+                let now = Instant::now();
                 let result = inner.storage.lookup(key, &span.handle()).await?;
+                let lookup_duration = now.elapsed();
+                // one request may have multiple lookups
+                inner.lookup_duration = Some(
+                    inner
+                        .lookup_duration
+                        .map_or(lookup_duration, |d| d + lookup_duration),
+                );
                 let result = result.and_then(|(meta, header)| {
                     if let Some(ts) = inner.valid_after {
                         if meta.created() < ts {
@@ -1015,7 +1026,7 @@ impl HttpCache {
         let _span = inner.traces.child("cache_lock");
         let lock = inner.lock.take(); // remove the lock from self
         if let Some(Locked::Read(r)) = lock {
-            let now = std::time::Instant::now();
+            let now = Instant::now();
             r.wait().await;
             let lock_duration = now.elapsed();
             // it's possible for a request to be locked more than once
@@ -1037,6 +1048,12 @@ impl HttpCache {
         self.inner.as_ref().and_then(|i| i.lock_duration)
     }
 
+    /// How long did this request spent on cache lookup and reading the header
+    pub fn lookup_duration(&self) -> Option<Duration> {
+        // FIXME: this duration is lost when cache is disabled
+        self.inner.as_ref().and_then(|i| i.lookup_duration)
+    }
+
     /// Delete the asset from the cache storage
     /// # Panic
     /// Need to be called after the cache key is set. Panic otherwise.
diff --git a/pingora-core/Cargo.toml b/pingora-core/Cargo.toml
index 5ced49dbd..ebbe4c085 100644
--- a/pingora-core/Cargo.toml
+++ b/pingora-core/Cargo.toml
@@ -36,7 +36,7 @@ log = { workspace = true }
 h2 = { workspace = true }
 lru = { workspace = true }
 nix = "~0.24.3"
-structopt = "0.3"
+clap = { version = "3.2.25", features = ["derive"] }
 once_cell = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 serde_yaml = "0.8"
@@ -54,7 +54,7 @@ sentry = { version = "0.26", features = [
 ], default-features = false }
 regex = "1"
 percent-encoding = "2.1"
-parking_lot = "0.12"
+parking_lot = { version = "0.12", features = ["arc_lock"] }
 socket2 = { version = "0", features = ["all"] }
 flate2 = { version = "1", features = ["zlib-ng"], default-features = false }
 sfv = "0"
diff --git a/pingora-core/src/connectors/http/v2.rs b/pingora-core/src/connectors/http/v2.rs
index 6f26b462c..2c11ede17 100644
--- a/pingora-core/src/connectors/http/v2.rs
+++ b/pingora-core/src/connectors/http/v2.rs
@@ -21,8 +21,8 @@ use crate::upstreams::peer::{Peer, ALPN};
 
 use bytes::Bytes;
 use h2::client::SendRequest;
-use log::debug;
-use parking_lot::RwLock;
+use log::{debug, warn};
+use parking_lot::{Mutex, RwLock};
 use pingora_error::{Error, ErrorType::*, OrErr, Result};
 use pingora_pool::{ConnectionMeta, ConnectionPool, PoolNode};
 use std::collections::HashMap;
@@ -55,6 +55,8 @@ pub(crate) struct ConnectionRefInner {
     // because `SendRequest` doesn't actually have access to the underlying Stream,
     // we log info about timing and tcp info here.
     pub(crate) digest: Digest,
+    // To serialize certain operations when trying to release the connect back to the pool,
+    pub(crate) release_lock: Arc<Mutex<()>>,
 }
 
 #[derive(Clone)]
@@ -77,6 +79,7 @@ impl ConnectionRef {
             max_streams,
             current_streams: AtomicUsize::new(0),
             digest,
+            release_lock: Arc::new(Mutex::new(())),
         }))
     }
     pub fn more_streams_allowed(&self) -> bool {
@@ -269,14 +272,16 @@ impl Connector {
             .get(reuse_hash)
             .or_else(|| self.idle_pool.get(&reuse_hash));
         if let Some(conn) = maybe_conn {
-            let h2_stream = conn
-                .spawn_stream()
-                .await?
-                .expect("connection from the pools should have free stream to allocate");
+            let h2_stream = conn.spawn_stream().await?;
+            if h2_stream.is_none() {
+                warn!("connection from the pools should have free stream to allocate, current in use {}, max {}",
+                    conn.0.current_streams.load(Ordering::Relaxed),
+                    conn.0.max_streams);
+            }
             if conn.more_streams_allowed() {
                 self.in_use_pool.insert(reuse_hash, conn);
             }
-            Ok(Some(h2_stream))
+            Ok(h2_stream)
         } else {
             Ok(None)
         }
@@ -298,6 +303,12 @@ impl Connector {
         let reuse_hash = peer.reuse_hash();
         // get a ref to the connection, which we might need below, before dropping the h2
         let conn = session.conn();
+
+        // The lock here is to make sure that in_use_pool.insert() below cannot be called after
+        // in_use_pool.release(), which would have put the conn entry in both pools.
+        // It also makes sure that only one conn will trigger the conn.is_idle() condition, which
+        // avoids putting the same conn into the idle_pool more than once.
+        let locked = conn.0.release_lock.lock_arc();
         // this drop() will both drop the actual stream and call the conn.release_stream()
         drop(session);
         // find and remove the conn stored in in_use_pool so that it could be put in the idle pool
@@ -308,6 +319,7 @@ impl Connector {
             return;
         }
         if conn.is_idle() {
+            drop(locked);
             let meta = ConnectionMeta {
                 key: reuse_hash,
                 id,
@@ -324,6 +336,7 @@ impl Connector {
             }
         } else {
             self.in_use_pool.insert(reuse_hash, conn);
+            drop(locked);
         }
     }
 
diff --git a/pingora-core/src/connectors/l4.rs b/pingora-core/src/connectors/l4.rs
index b623d9f3d..ffbf084d3 100644
--- a/pingora-core/src/connectors/l4.rs
+++ b/pingora-core/src/connectors/l4.rs
@@ -20,7 +20,6 @@ use std::os::unix::io::AsRawFd;
 
 use crate::protocols::l4::ext::{
     connect_uds, connect_with as tcp_connect, set_recv_buf, set_tcp_fastopen_connect,
-    set_tcp_keepalive,
 };
 use crate::protocols::l4::socket::SocketAddr;
 use crate::protocols::l4::stream::Stream;
@@ -61,10 +60,6 @@ where
             match conn_res {
                 Ok(socket) => {
                     debug!("connected to new server: {}", peer.address());
-                    if let Some(ka) = peer.tcp_keepalive() {
-                        debug!("Setting tcp keepalive");
-                        set_tcp_keepalive(&socket, ka)?;
-                    }
                     Ok(socket.into())
                 }
                 Err(e) => {
@@ -92,7 +87,6 @@ where
             match conn_res {
                 Ok(socket) => {
                     debug!("connected to new server: {}", peer.address());
-                    // no SO_KEEPALIVE for UDS
                     Ok(socket.into())
                 }
                 Err(e) => {
@@ -111,6 +105,10 @@ where
         stream.tracer = Some(t);
     }
 
+    // settings applied based on stream type
+    if let Some(ka) = peer.tcp_keepalive() {
+        stream.set_keepalive(ka)?;
+    }
     stream.set_nodelay()?;
 
     let digest = SocketDigest::from_raw_fd(stream.as_raw_fd());
diff --git a/pingora-core/src/listeners/l4.rs b/pingora-core/src/listeners/l4.rs
index 3718f14a3..406190e7b 100644
--- a/pingora-core/src/listeners/l4.rs
+++ b/pingora-core/src/listeners/l4.rs
@@ -28,6 +28,7 @@ use tokio::net::TcpSocket;
 use crate::protocols::l4::ext::set_tcp_fastopen_backlog;
 use crate::protocols::l4::listener::Listener;
 pub use crate::protocols::l4::stream::Stream;
+use crate::protocols::TcpKeepalive;
 use crate::server::ListenFds;
 
 const TCP_LISTENER_MAX_TRY: usize = 30;
@@ -51,7 +52,17 @@ impl AsRef<str> for ServerAddress {
     }
 }
 
-/// TCP socket configuration options.
+impl ServerAddress {
+    fn tcp_sock_opts(&self) -> Option<&TcpSocketOptions> {
+        match &self {
+            Self::Tcp(_, op) => op.into(),
+            _ => None,
+        }
+    }
+}
+
+/// TCP socket configuration options, this is used for setting options on
+/// listening sockets and accepted connections.
 #[non_exhaustive]
 #[derive(Clone, Debug, Default)]
 pub struct TcpSocketOptions {
@@ -62,6 +73,9 @@ pub struct TcpSocketOptions {
     /// Enable TCP fast open and set the backlog size of it.
     /// See the [man page](https://man7.org/linux/man-pages/man7/tcp.7.html) for more information.
     pub tcp_fastopen: Option<usize>,
+    /// Enable TCP keepalive on accepted connections.
+    /// See the [man page](https://man7.org/linux/man-pages/man7/tcp.7.html) for more information.
+    pub tcp_keepalive: Option<TcpKeepalive>,
     // TODO: allow configuring reuseaddr, backlog, etc. from here?
 }
 
@@ -257,6 +271,18 @@ impl ListenerEndpoint {
         Ok(())
     }
 
+    fn apply_stream_settings(&self, stream: &mut Stream) -> Result<()> {
+        // settings are applied based on whether the underlying stream supports it
+        stream.set_nodelay()?;
+        let Some(op) = self.listen_addr.tcp_sock_opts() else {
+            return Ok(());
+        };
+        if let Some(ka) = op.tcp_keepalive.as_ref() {
+            stream.set_keepalive(ka)?;
+        }
+        Ok(())
+    }
+
     pub async fn accept(&mut self) -> Result<Stream> {
         let Some(listener) = self.listener.as_mut() else {
             // panic otherwise this thing dead loop
@@ -266,7 +292,7 @@ impl ListenerEndpoint {
             .accept()
             .await
             .or_err(AcceptError, "Fail to accept()")?;
-        stream.set_nodelay()?;
+        self.apply_stream_settings(&mut stream)?;
         Ok(stream)
     }
 }
diff --git a/pingora-core/src/protocols/http/v1/server.rs b/pingora-core/src/protocols/http/v1/server.rs
index 714e9906a..6f8c96ce2 100644
--- a/pingora-core/src/protocols/http/v1/server.rs
+++ b/pingora-core/src/protocols/http/v1/server.rs
@@ -386,7 +386,7 @@ impl HttpSession {
     /// This function can be called more than once to send 1xx informational headers excluding 101.
     pub async fn write_response_header(&mut self, mut header: Box<ResponseHeader>) -> Result<()> {
         if let Some(resp) = self.response_written.as_ref() {
-            if !resp.status.is_informational() {
+            if !resp.status.is_informational() || self.upgraded {
                 warn!("Respond header is already sent, cannot send again");
                 return Ok(());
             }
@@ -1431,6 +1431,14 @@ mod tests_stream {
             .unwrap();
         let n = http_stream.write_body(wire_body).await.unwrap().unwrap();
         assert_eq!(wire_body.len(), n);
+        // simulate upgrade
+        http_stream.upgraded = true;
+        // this write should be ignored
+        let response_502 = ResponseHeader::build(StatusCode::BAD_GATEWAY, None).unwrap();
+        http_stream
+            .write_response_header_ref(&response_502)
+            .await
+            .unwrap();
     }
 
     #[tokio::test]
diff --git a/pingora-core/src/protocols/l4/stream.rs b/pingora-core/src/protocols/l4/stream.rs
index e2efdbfd5..6169d3115 100644
--- a/pingora-core/src/protocols/l4/stream.rs
+++ b/pingora-core/src/protocols/l4/stream.rs
@@ -26,6 +26,7 @@ use std::time::SystemTime;
 use tokio::io::{self, AsyncRead, AsyncWrite, AsyncWriteExt, BufStream, ReadBuf};
 use tokio::net::{TcpStream, UnixStream};
 
+use crate::protocols::l4::ext::{set_tcp_keepalive, TcpKeepalive};
 use crate::protocols::raw_connect::ProxyDigest;
 use crate::protocols::{
     GetProxyDigest, GetSocketDigest, GetTimingDigest, Shutdown, SocketDigest, Ssl, TimingDigest,
@@ -151,6 +152,15 @@ impl Stream {
         }
         Ok(())
     }
+
+    /// set TCP keepalive settings for this connection if `self` is TCP
+    pub fn set_keepalive(&mut self, ka: &TcpKeepalive) -> Result<()> {
+        if let RawStream::Tcp(s) = &self.stream.get_ref() {
+            debug!("Setting tcp keepalive");
+            set_tcp_keepalive(s, ka)?;
+        }
+        Ok(())
+    }
 }
 
 impl From<TcpStream> for Stream {
diff --git a/pingora-core/src/protocols/mod.rs b/pingora-core/src/protocols/mod.rs
index 6b7a35755..2c25e8953 100644
--- a/pingora-core/src/protocols/mod.rs
+++ b/pingora-core/src/protocols/mod.rs
@@ -24,6 +24,7 @@ pub use digest::{
     Digest, GetProxyDigest, GetSocketDigest, GetTimingDigest, ProtoDigest, SocketDigest,
     TimingDigest,
 };
+pub use l4::ext::TcpKeepalive;
 pub use ssl::ALPN;
 
 use async_trait::async_trait;
diff --git a/pingora-core/src/server/configuration/mod.rs b/pingora-core/src/server/configuration/mod.rs
index b556cc72d..5b7609dcb 100644
--- a/pingora-core/src/server/configuration/mod.rs
+++ b/pingora-core/src/server/configuration/mod.rs
@@ -19,11 +19,11 @@
 //! * Number of threads per service
 //! * Error log file path
 
+use clap::Parser;
 use log::{debug, trace};
 use pingora_error::{Error, ErrorType::*, OrErr, Result};
 use serde::{Deserialize, Serialize};
 use std::fs;
-use structopt::StructOpt;
 
 /// The configuration file
 ///
@@ -118,48 +118,54 @@ impl Default for ServerConf {
 /// Command-line options
 ///
 /// Call `Opt::from_args()` to build this object from the process's command line arguments.
-#[derive(StructOpt, Debug)]
-#[structopt(name = "basic")]
+#[derive(Parser, Debug)]
+#[clap(name = "basic")]
 pub struct Opt {
     /// Whether this server should try to upgrade from a running old server
-    ///
-    /// `-u` or `--upgrade` can be used
-    #[structopt(short, long)]
+    #[clap(
+        short,
+        long,
+        help = "This is the base set of command line arguments for a pingora-based service"
+    )]
     pub upgrade: bool,
+
     /// Whether should run this server in the background
-    ///
-    /// `-d` or `--daemon` can be used
-    #[structopt(short, long)]
+    #[clap(short, long)]
     pub daemon: bool,
+
     /// Not actually used. This flag is there so that the server is not upset seeing this flag
     /// passed from `cargo test` sometimes
-    #[structopt(long)]
+    #[clap(long, hidden = true)]
     pub nocapture: bool,
+
     /// Test the configuration and exit
     ///
     /// When this flag is set, calling `server.bootstrap()` will exit the process without errors
     ///
     /// This flag is useful for upgrading service where the user wants to make sure the new
     /// service can start before shutting down the old server process.
-    ///
-    /// `-t` or `--test` can be used
-    #[structopt(short, long)]
+    #[clap(
+        short,
+        long,
+        help = "This flag is useful for upgrading service where the user wants \
+                to make sure the new service can start before shutting down \
+                the old server process."
+    )]
     pub test: bool,
+
     /// The path to the configuration file.
     ///
     /// See [`ServerConf`] for more details of the configuration file.
-    ///
-    /// `-c` or `--conf` can be used
-    #[structopt(short, long)]
+    #[clap(short, long, help = "The path to the configuration file.")]
     pub conf: Option<String>,
 }
 
 /// Create the default instance of Opt based on the current command-line args.
-/// This is equivalent to running `Opt::from_args` but does not require the
-/// caller to have included the `structopt::StructOpt`
+/// This is equivalent to running `Opt::parse` but does not require the
+/// caller to have included the `clap::Parser`
 impl Default for Opt {
     fn default() -> Self {
-        Opt::from_args()
+        Opt::parse()
     }
 }
 
diff --git a/pingora-core/src/upstreams/peer.rs b/pingora-core/src/upstreams/peer.rs
index cc6dfc73d..21f737b8a 100644
--- a/pingora-core/src/upstreams/peer.rs
+++ b/pingora-core/src/upstreams/peer.rs
@@ -26,9 +26,9 @@ use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 
-pub use crate::protocols::l4::ext::TcpKeepalive;
 use crate::protocols::l4::socket::SocketAddr;
 use crate::protocols::ConnFdReusable;
+use crate::protocols::TcpKeepalive;
 use crate::tls::x509::X509;
 use crate::utils::{get_organization_unit, CertKey};
 
diff --git a/pingora-proxy/Cargo.toml b/pingora-proxy/Cargo.toml
index c39119612..2e75cc695 100644
--- a/pingora-proxy/Cargo.toml
+++ b/pingora-proxy/Cargo.toml
@@ -31,7 +31,7 @@ async-trait = { workspace = true }
 log = { workspace = true }
 h2 = { workspace = true }
 once_cell = { workspace = true }
-structopt = "0.3"
+clap = { version = "3.2.25", features = ["derive"] }
 regex = "1"
 
 [dev-dependencies]
diff --git a/pingora-proxy/examples/ctx.rs b/pingora-proxy/examples/ctx.rs
index 36169e21a..4838391e3 100644
--- a/pingora-proxy/examples/ctx.rs
+++ b/pingora-proxy/examples/ctx.rs
@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use async_trait::async_trait;
+use clap::Parser;
 use log::info;
 use std::sync::Mutex;
-use structopt::StructOpt;
 
 use pingora_core::server::configuration::Opt;
 use pingora_core::server::Server;
@@ -82,7 +82,7 @@ fn main() {
     env_logger::init();
 
     // read command line arguments
-    let opt = Opt::from_args();
+    let opt = Opt::parse();
     let mut my_server = Server::new(Some(opt)).unwrap();
     my_server.bootstrap();
 
diff --git a/pingora-proxy/examples/gateway.rs b/pingora-proxy/examples/gateway.rs
index 27c502049..0bd53306d 100644
--- a/pingora-proxy/examples/gateway.rs
+++ b/pingora-proxy/examples/gateway.rs
@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use async_trait::async_trait;
+use clap::Parser;
 use log::info;
 use prometheus::register_int_counter;
-use structopt::StructOpt;
 
 use pingora_core::server::configuration::Opt;
 use pingora_core::server::Server;
@@ -114,7 +114,7 @@ fn main() {
     env_logger::init();
 
     // read command line arguments
-    let opt = Opt::from_args();
+    let opt = Opt::parse();
     let mut my_server = Server::new(Some(opt)).unwrap();
     my_server.bootstrap();
 
diff --git a/pingora-proxy/examples/load_balancer.rs b/pingora-proxy/examples/load_balancer.rs
index 425b6ece5..614981d60 100644
--- a/pingora-proxy/examples/load_balancer.rs
+++ b/pingora-proxy/examples/load_balancer.rs
@@ -13,10 +13,10 @@
 // limitations under the License.
 
 use async_trait::async_trait;
+use clap::Parser;
 use log::info;
 use pingora_core::services::background::background_service;
 use std::{sync::Arc, time::Duration};
-use structopt::StructOpt;
 
 use pingora_core::server::configuration::Opt;
 use pingora_core::server::Server;
@@ -62,7 +62,7 @@ fn main() {
     env_logger::init();
 
     // read command line arguments
-    let opt = Opt::from_args();
+    let opt = Opt::parse();
     let mut my_server = Server::new(Some(opt)).unwrap();
     my_server.bootstrap();
 
diff --git a/pingora-proxy/examples/modify_response.rs b/pingora-proxy/examples/modify_response.rs
index 5166dc64b..6730f7186 100644
--- a/pingora-proxy/examples/modify_response.rs
+++ b/pingora-proxy/examples/modify_response.rs
@@ -14,9 +14,9 @@
 
 use async_trait::async_trait;
 use bytes::Bytes;
+use clap::Parser;
 use serde::{Deserialize, Serialize};
 use std::net::ToSocketAddrs;
-use structopt::StructOpt;
 
 use pingora_core::server::configuration::Opt;
 use pingora_core::server::Server;
@@ -117,7 +117,7 @@ impl ProxyHttp for Json2Yaml {
 fn main() {
     env_logger::init();
 
-    let opt = Opt::from_args();
+    let opt = Opt::parse();
     let mut my_server = Server::new(Some(opt)).unwrap();
     my_server.bootstrap();
 
diff --git a/pingora-proxy/src/proxy_cache.rs b/pingora-proxy/src/proxy_cache.rs
index 00b61daae..1611f5937 100644
--- a/pingora-proxy/src/proxy_cache.rs
+++ b/pingora-proxy/src/proxy_cache.rs
@@ -116,11 +116,7 @@ impl<SV> HttpProxy<SV> {
                             // (this is a soft purge which tries to revalidate,
                             // vs. hard purge which forces miss)
                             // TODO: allow hard purge
-                            match self
-                                .inner
-                                .cache_hit_filter(&meta, ctx, session.req_header())
-                                .await
-                            {
+                            match self.inner.cache_hit_filter(session, &meta, ctx).await {
                                 Err(e) => {
                                     error!(
                                         "Failed to filter cache hit: {e}, {}",
diff --git a/pingora-proxy/src/proxy_h1.rs b/pingora-proxy/src/proxy_h1.rs
index c702af3dc..feac860c5 100644
--- a/pingora-proxy/src/proxy_h1.rs
+++ b/pingora-proxy/src/proxy_h1.rs
@@ -223,7 +223,14 @@ impl<SV> HttpProxy<SV> {
                 .reserve()
                 .await
                 .or_err(InternalError, "reserving body pipe")?;
-            send_body_to_pipe(buffer, downstream_state.is_done(), send_permit).await;
+            self.send_body_to_pipe(
+                session,
+                buffer,
+                downstream_state.is_done(),
+                send_permit,
+                ctx,
+            )
+            .await?;
         }
 
         let mut response_state = ResponseStateMachine::new();
@@ -288,12 +295,15 @@ impl<SV> HttpProxy<SV> {
                         response_state.maybe_set_upstream_done(true);
                     }
                     // TODO: consider just drain this if serve_from_cache is set
-                    let request_done = send_body_to_pipe(
+                    let is_body_done = session.is_body_done();
+                    let request_done = self.send_body_to_pipe(
+                        session,
                         body,
-                        session.is_body_done(),
+                        is_body_done,
                         send_permit.unwrap(), // safe because we checked is_ok()
+                        ctx,
                     )
-                    .await;
+                    .await?;
                     downstream_state.maybe_finished(request_done);
                 },
 
@@ -520,30 +530,48 @@ impl<SV> HttpProxy<SV> {
             HttpTask::Failed(_) => Ok(task), // Do nothing just pass the error down
         }
     }
-}
 
-// TODO:: use this function to replace send_body_to2
-pub(crate) async fn send_body_to_pipe(
-    data: Option<Bytes>,
-    end_of_body: bool,
-    tx: mpsc::Permit<'_, HttpTask>,
-) -> bool {
-    match data {
-        Some(data) => {
-            debug!("Read {} bytes body from downstream", data.len());
-            if data.is_empty() && !end_of_body {
-                /* it is normal to get 0 bytes because of multi-chunk
-                 * don't write 0 bytes to downstream since it will be
-                 * misread as the terminating chunk */
-                return false;
-            }
-            tx.send(HttpTask::Body(Some(data), end_of_body));
-            end_of_body
-        }
-        None => {
-            tx.send(HttpTask::Body(None, true));
-            true
+    // TODO:: use this function to replace send_body_to2
+    async fn send_body_to_pipe(
+        &self,
+        session: &mut Session,
+        mut data: Option<Bytes>,
+        end_of_body: bool,
+        tx: mpsc::Permit<'_, HttpTask>,
+        ctx: &mut SV::CTX,
+    ) -> Result<bool>
+    where
+        SV: ProxyHttp + Send + Sync,
+        SV::CTX: Send + Sync,
+    {
+        // None: end of body
+        // this var is to signal if downstream finish sending the body, which shouldn't be
+        // affected by the request_body_filter
+        let end_of_body = end_of_body || data.is_none();
+
+        self.inner
+            .request_body_filter(session, &mut data, end_of_body, ctx)
+            .await?;
+
+        // the flag to signal to upstream
+        let upstream_end_of_body = end_of_body || data.is_none();
+
+        /* It is normal to get 0 bytes because of multi-chunk or request_body_filter decides not to
+         * output anything yet.
+         * Don't write 0 bytes to the network since it will be
+         * treated as the terminating chunk */
+        if !upstream_end_of_body && data.as_ref().map_or(false, |d| d.is_empty()) {
+            return Ok(false);
         }
+
+        debug!(
+            "Read {} bytes body from downstream",
+            data.as_ref().map_or(-1, |d| d.len() as isize)
+        );
+
+        tx.send(HttpTask::Body(data, upstream_end_of_body));
+
+        Ok(end_of_body)
     }
 }
 
diff --git a/pingora-proxy/src/proxy_h2.rs b/pingora-proxy/src/proxy_h2.rs
index 8f0dffacd..534564fbc 100644
--- a/pingora-proxy/src/proxy_h2.rs
+++ b/pingora-proxy/src/proxy_h2.rs
@@ -223,7 +223,14 @@ impl<SV> HttpProxy<SV> {
 
         // retry, send buffer if it exists
         if let Some(buffer) = session.as_mut().get_retry_buffer() {
-            send_body_to2(Ok(Some(buffer)), downstream_state.is_done(), client_body)?;
+            self.send_body_to2(
+                session,
+                Some(buffer),
+                downstream_state.is_done(),
+                client_body,
+                ctx,
+            )
+            .await?;
         }
 
         let mut response_state = ResponseStateMachine::new();
@@ -260,7 +267,10 @@ impl<SV> HttpProxy<SV> {
                            }
                         }
                     };
-                    let request_done = send_body_to2(Ok(body), session.is_body_done(), client_body)?;
+                    let is_body_done = session.is_body_done();
+                    let request_done =
+                        self.send_body_to2(session, body, is_body_done, client_body, ctx)
+                        .await?;
                     downstream_state.maybe_finished(request_done);
                 },
 
@@ -497,38 +507,40 @@ impl<SV> HttpProxy<SV> {
             HttpTask::Failed(_) => Ok(task), // Do nothing just pass the error down
         }
     }
-}
 
-pub(crate) fn send_body_to2(
-    data: Result<Option<Bytes>>,
-    end_of_body: bool,
-    client_body: &mut h2::SendStream<bytes::Bytes>,
-) -> Result<bool> {
-    match data {
-        Ok(res) => match res {
-            Some(data) => {
-                let data_len = data.len();
-                debug!(
-                    "Read {} bytes body from downstream, body end: {}",
-                    data_len, end_of_body
-                );
-                if data_len == 0 && !end_of_body {
-                    /* it is normal to get 0 bytes because of multi-chunk parsing */
-                    return Ok(false);
-                }
-                write_body(client_body, data, end_of_body).map_err(|e| e.into_up())?;
-                debug!("Write {} bytes body to h2 upstream", data_len);
-                Ok(end_of_body)
-            }
-            None => {
-                debug!("Read downstream body done");
-                /* send a standalone END_STREAM flag */
-                write_body(client_body, Bytes::new(), true).map_err(|e| e.into_up())?;
-                debug!("Write END_STREAM to h2 upstream");
-                Ok(true)
-            }
-        },
-        Err(e) => e.into_down().into_err(),
+    async fn send_body_to2(
+        &self,
+        session: &mut Session,
+        mut data: Option<Bytes>,
+        end_of_body: bool,
+        client_body: &mut h2::SendStream<bytes::Bytes>,
+        ctx: &mut SV::CTX,
+    ) -> Result<bool>
+    where
+        SV: ProxyHttp + Send + Sync,
+        SV::CTX: Send + Sync,
+    {
+        self.inner
+            .request_body_filter(session, &mut data, end_of_body, ctx)
+            .await?;
+
+        /* it is normal to get 0 bytes because of multi-chunk parsing or request_body_filter.
+         * Although there is no harm writing empty byte to h2, unlike h1, we ignore it
+         * for consistency */
+        if !end_of_body && data.as_ref().map_or(false, |d| d.is_empty()) {
+            return Ok(false);
+        }
+
+        if let Some(data) = data {
+            debug!("Write {} bytes body to h2 upstream", data.len());
+            write_body(client_body, data, end_of_body).map_err(|e| e.into_up())?;
+        } else {
+            debug!("Read downstream body done");
+            /* send a standalone END_STREAM flag */
+            write_body(client_body, Bytes::new(), true).map_err(|e| e.into_up())?;
+        }
+
+        Ok(end_of_body)
     }
 }
 
diff --git a/pingora-proxy/src/proxy_trait.rs b/pingora-proxy/src/proxy_trait.rs
index 01586810d..307280cdb 100644
--- a/pingora-proxy/src/proxy_trait.rs
+++ b/pingora-proxy/src/proxy_trait.rs
@@ -14,6 +14,7 @@
 
 use super::*;
 use pingora_cache::{key::HashBinary, CacheKey, CacheMeta, RespCacheable, RespCacheable::*};
+use std::time::Duration;
 
 /// The interface to control the HTTP proxy
 ///
@@ -55,6 +56,27 @@ pub trait ProxyHttp {
         Ok(false)
     }
 
+    /// 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**.
+    ///
+    /// 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.
+    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.
@@ -87,9 +109,9 @@ pub trait ProxyHttp {
     // 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,
-        _req: &RequestHeader,
     ) -> Result<bool>
     where
         Self::CTX: Send + Sync,
@@ -238,7 +260,7 @@ pub trait ProxyHttp {
         _body: &mut Option<Bytes>,
         _end_of_stream: bool,
         _ctx: &mut Self::CTX,
-    ) -> Result<Option<std::time::Duration>>
+    ) -> Result<Option<Duration>>
     where
         Self::CTX: Send + Sync,
     {
diff --git a/pingora/Cargo.toml b/pingora/Cargo.toml
index 8ae3aba5a..09ae1e7e3 100644
--- a/pingora/Cargo.toml
+++ b/pingora/Cargo.toml
@@ -30,7 +30,7 @@ pingora-proxy = { version = "0.2.0", path = "../pingora-proxy", optional = true,
 pingora-cache = { version = "0.2.0", path = "../pingora-cache", optional = true, default-features = false }
 
 [dev-dependencies]
-structopt = "0.3"
+clap = { version = "3.2.25", features = ["derive"] }
 tokio = { workspace = true, features = ["rt-multi-thread", "signal"] }
 matches = "0.1"
 env_logger = "0.9"
diff --git a/pingora/examples/server.rs b/pingora/examples/server.rs
index 546c1dce7..a2a092e9d 100644
--- a/pingora/examples/server.rs
+++ b/pingora/examples/server.rs
@@ -15,13 +15,14 @@
 #[global_allocator]
 static GLOBAL: jemallocator::Jemalloc = jemallocator::Jemalloc;
 
+use pingora::protocols::TcpKeepalive;
 use pingora::server::configuration::Opt;
 use pingora::server::{Server, ShutdownWatch};
 use pingora::services::background::{background_service, BackgroundService};
 use pingora::services::{listening::Service as ListeningService, Service};
 
 use async_trait::async_trait;
-use structopt::StructOpt;
+use clap::Parser;
 use tokio::time::interval;
 
 use std::time::Duration;
@@ -105,7 +106,7 @@ pub fn main() {
 
     print!("{USAGE}");
 
-    let opt = Some(Opt::from_args());
+    let opt = Some(Opt::parse());
     let mut my_server = Server::new(opt).unwrap();
     my_server.bootstrap();
 
@@ -119,8 +120,15 @@ pub fn main() {
         .unwrap();
 
     let mut echo_service_http = service::echo::echo_service_http();
+
     let mut options = pingora::listeners::TcpSocketOptions::default();
     options.tcp_fastopen = Some(10);
+    options.tcp_keepalive = Some(TcpKeepalive {
+        idle: Duration::from_secs(60),
+        interval: Duration::from_secs(5),
+        count: 5,
+    });
+
     echo_service_http.add_tcp_with_settings("0.0.0.0:6145", options);
     echo_service_http.add_uds("/tmp/echo.sock", None);
 


Test patch:
diff --git a/pingora-core/tests/utils/mod.rs b/pingora-core/tests/utils/mod.rs
index 1555b7dbc..af5de8544 100644
--- a/pingora-core/tests/utils/mod.rs
+++ b/pingora-core/tests/utils/mod.rs
@@ -15,11 +15,11 @@
 use once_cell::sync::Lazy;
 use std::{thread, time};
 
+use clap::Parser;
 use pingora_core::listeners::Listeners;
 use pingora_core::server::configuration::Opt;
 use pingora_core::server::Server;
 use pingora_core::services::listening::Service;
-use structopt::StructOpt;
 
 use async_trait::async_trait;
 use bytes::Bytes;
@@ -106,7 +106,7 @@ impl MyServer {
             "tests/pingora_conf.yaml".into(),
         ];
         let server_handle = thread::spawn(|| {
-            entry_point(Some(Opt::from_iter(opts)));
+            entry_point(Some(Opt::parse_from(opts)));
         });
         // wait until the server is up
         thread::sleep(time::Duration::from_secs(2));
diff --git a/pingora-proxy/tests/utils/server_utils.rs b/pingora-proxy/tests/utils/server_utils.rs
index e6ffe5834..02034fe31 100644
--- a/pingora-proxy/tests/utils/server_utils.rs
+++ b/pingora-proxy/tests/utils/server_utils.rs
@@ -14,6 +14,7 @@
 
 use super::cert;
 use async_trait::async_trait;
+use clap::Parser;
 use http::header::VARY;
 use http::HeaderValue;
 use once_cell::sync::Lazy;
@@ -36,7 +37,6 @@ use pingora_proxy::{ProxyHttp, Session};
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use std::thread;
-use structopt::StructOpt;
 
 pub struct ExampleProxyHttps {}
 
@@ -469,7 +469,7 @@ fn test_main() {
         "-c".into(),
         "tests/pingora_conf.yaml".into(),
     ];
-    let mut my_server = pingora_core::server::Server::new(Some(Opt::from_iter(opts))).unwrap();
+    let mut my_server = pingora_core::server::Server::new(Some(Opt::parse_from(opts))).unwrap();
     my_server.bootstrap();
 
     let mut proxy_service_http =