Synchronize internally merged commits (2024-03-15)

.bleep

@@ -1 +1 @@
-20f290d8adc57ed489bcdd1b950833d0eb6bd8ed
+deb3c5409e938ec9c7d0da9b7a2d331eabbb2cd5

.gitignore

@@ -5,3 +5,4 @@ dhat-heap.json
 .vscode
 .idea
 .cover
+bleeper.user.toml

pingora-boringssl/src/lib.rs

@@ -15,7 +15,7 @@
 //! The BoringSSL API compatibility layer.
 //!
 //! This crate aims at making [boring] APIs exchangeable with [openssl-rs](https://docs.rs/openssl/latest/openssl/).
-//! In other words, this crate and `pingora-openssl` expose identical rust APIs.
+//! In other words, this crate and [`pingora-openssl`](https://docs.rs/pingora-openssl) expose identical rust APIs.
 
 #![warn(clippy::all)]
 

pingora-cache/src/cache_control.rs

@@ -20,8 +20,7 @@ use http::header::HeaderName;
 use http::HeaderValue;
 use indexmap::IndexMap;
 use once_cell::sync::Lazy;
-use pingora_error::{Error, ErrorType, Result};
-use pingora_http::ResponseHeader;
+use pingora_error::{Error, ErrorType};
 use regex::bytes::Regex;
 use std::num::IntErrorKind;
 use std::slice;
@@ -434,7 +433,6 @@ pub trait InterpretCacheControl {
 mod tests {
  use super::*;
  use http::header::CACHE_CONTROL;
- use http::HeaderValue;
  use http::{request, response};
 
  fn build_response(cc_key: HeaderName, cc_value: &str) -> response::Parts {

pingora-cache/src/eviction/lru.rs

@@ -233,7 +233,6 @@ impl<const N: usize> EvictionManager for Manager<N> {
 mod test {
  use super::*;
  use crate::CacheKey;
- use EvictionManager;
 
  // we use shard (N) = 1 for eviction consistency in all tests
 

pingora-cache/src/filters.rs

@@ -16,12 +16,12 @@
 
 use super::*;
 use crate::cache_control::{CacheControl, Cacheable, InterpretCacheControl};
-use crate::{RespCacheable, RespCacheable::*};
+use crate::RespCacheable::*;
 
 use http::{header, HeaderValue};
 use httpdate::HttpDate;
 use log::warn;
-use pingora_http::{RequestHeader, ResponseHeader};
+use pingora_http::RequestHeader;
 
 /// Decide if the request can be cacheable
 pub fn request_cacheable(req_header: &ReqHeader) -> bool {
@@ -206,6 +206,7 @@ pub mod upstream {
 #[cfg(test)]
 mod tests {
  use super::*;
+ use crate::RespCacheable::Cacheable;
  use http::header::{HeaderName, CACHE_CONTROL, EXPIRES, SET_COOKIE};
  use http::StatusCode;
  use httpdate::fmt_http_date;

pingora-cache/src/memory.rs

@@ -19,8 +19,8 @@
 //TODO: Mark this module #[test] only
 
 use super::*;
-use crate::key::{CacheHashKey, CompactCacheKey};
-use crate::storage::{HandleHit, HandleMiss, Storage};
+use crate::key::CompactCacheKey;
+use crate::storage::{HandleHit, HandleMiss};
 use crate::trace::SpanHandle;
 
 use async_trait::async_trait;
@@ -211,7 +211,7 @@ impl HandleHit for MemHitHandler {
 pub struct MemMissHandler {
  body: Arc<RwLock<Vec<u8>>>,
  bytes_written: Arc<watch::Sender<PartialState>>,
- // these are used only in finish() to to data from temp to cache
+ // these are used only in finish() to data from temp to cache
  key: String,
  cache: Arc<RwLock<HashMap<String, CacheObject>>>,
  temp: Arc<RwLock<HashMap<String, TempObject>>>,

pingora-cache/src/put.rs

@@ -264,14 +264,12 @@ mod test {
 
 mod parse_response {
  use super::*;
- use bytes::{Bytes, BytesMut};
+ use bytes::BytesMut;
  use httparse::Status;
  use pingora_error::{
  Error,
  ErrorType::{self, *},
- Result,
  };
- use pingora_http::ResponseHeader;
 
  pub const INVALID_CHUNK: ErrorType = ErrorType::new("InvalidChunk");
  pub const INCOMPLETE_BODY: ErrorType = ErrorType::new("IncompleteHttpBody");
@@ -280,7 +278,7 @@ mod parse_response {
  const INIT_HEADER_BUF_SIZE: usize = 4096;
  const CHUNK_DELIMITER_SIZE: usize = 2; // \r\n
 
- #[derive(Debug, Clone, Copy)]
+ #[derive(Debug, Clone, Copy, PartialEq)]
  enum ParseState {
  Init,
  PartialHeader,
@@ -561,6 +559,10 @@ mod parse_response {
  let output = parser.inject_data(input);
  // header is not complete
  assert!(output.is_err());
+ match parser.state {
+ ParseState::Invalid(httparse::Error::Version) => {}
+ _ => panic!("should have failed to parse"),
+ }
  }
 
  #[test]

pingora-core/src/apps/mod.rs

@@ -24,6 +24,7 @@ use std::sync::Arc;
 
 use crate::protocols::http::v2::server;
 use crate::protocols::http::ServerSession;
+use crate::protocols::Digest;
 use crate::protocols::Stream;
 use crate::protocols::ALPN;
 
@@ -91,6 +92,15 @@ where
  ) -> Option<Stream> {
  match stream.selected_alpn_proto() {
  Some(ALPN::H2) => {
+ // create a shared connection digest
+ let digest = Arc::new(Digest {
+ ssl_digest: stream.get_ssl_digest(),
+ // TODO: log h2 handshake time
+ timing_digest: stream.get_timing_digest(),
+ proxy_digest: stream.get_proxy_digest(),
+ socket_digest: stream.get_socket_digest(),
+ });
+
  let h2_options = self.h2_options();
  let h2_conn = server::handshake(stream, h2_options).await;
  let mut h2_conn = match h2_conn {
@@ -100,10 +110,12 @@ where
  }
  Ok(c) => c,
  };
+
  loop {
  // this loop ends when the client decides to close the h2 conn
  // TODO: add a timeout?
- let h2_stream = server::HttpSession::from_h2_conn(&mut h2_conn).await;
+ let h2_stream =
+ server::HttpSession::from_h2_conn(&mut h2_conn, digest.clone()).await;
  let h2_stream = match h2_stream {
  Err(e) => {
  // It is common for client to just disconnect TCP without properly

pingora-core/src/apps/prometheus_http_app.rs

@@ -15,7 +15,7 @@
 //! An HTTP application that reports Prometheus metrics.
 
 use async_trait::async_trait;
-use http::{self, Response};
+use http::Response;
 use prometheus::{Encoder, TextEncoder};
 
 use super::http_app::HttpServer;

pingora-core/src/connectors/http/v1.rs

@@ -65,6 +65,7 @@ impl Connector {
 #[cfg(test)]
 mod tests {
  use super::*;
+ use crate::protocols::l4::socket::SocketAddr;
  use crate::upstreams::peer::HttpPeer;
  use pingora_http::RequestHeader;
 
@@ -85,6 +86,8 @@ mod tests {
  let peer = HttpPeer::new(("1.1.1.1", 80), false, "".into());
  // make a new connection to 1.1.1.1
  let (http, reused) = connector.get_http_session(&peer).await.unwrap();
+ let server_addr = http.server_addr().unwrap();
+ assert_eq!(*server_addr, "1.1.1.1:80".parse::<SocketAddr>().unwrap());
  assert!(!reused);
 
  // this http is not even used, so not be able to reuse
@@ -104,6 +107,8 @@ mod tests {
  let peer = HttpPeer::new(("1.1.1.1", 443), true, "one.one.one.one".into());
  // make a new connection to https://1.1.1.1
  let (http, reused) = connector.get_http_session(&peer).await.unwrap();
+ let server_addr = http.server_addr().unwrap();
+ assert_eq!(*server_addr, "1.1.1.1:443".parse::<SocketAddr>().unwrap());
  assert!(!reused);
 
  // this http is not even used, so not be able to reuse

pingora-core/src/connectors/http/v2.rs

@@ -369,6 +369,7 @@ async fn handshake(
  // TODO: log h2 handshake time
  timing_digest: stream.get_timing_digest(),
  proxy_digest: stream.get_proxy_digest(),
+ socket_digest: stream.get_socket_digest(),
  };
  // TODO: make these configurable
  let (send_req, connection) = Builder::new()

pingora-core/src/connectors/l4.rs

@@ -16,10 +16,12 @@ use log::debug;
 use pingora_error::{Context, Error, ErrorType::*, OrErr, Result};
 use rand::seq::SliceRandom;
 use std::net::SocketAddr as InetSocketAddr;
+use std::os::unix::io::AsRawFd;
 
 use crate::protocols::l4::ext::{connect as tcp_connect, connect_uds, set_tcp_keepalive};
 use crate::protocols::l4::socket::SocketAddr;
 use crate::protocols::l4::stream::Stream;
+use crate::protocols::{GetSocketDigest, SocketDigest};
 use crate::upstreams::peer::Peer;
 
 /// Establish a connection (l4) to the given peer using its settings and an optional bind address.
@@ -32,7 +34,8 @@ where
  .await
  .err_context(|| format!("Fail to establish CONNECT proxy: {}", peer));
  }
- let mut stream: Stream = match peer.address() {
+ let peer_addr = peer.address();
+ let mut stream: Stream = match peer_addr {
  SocketAddr::Inet(addr) => {
  let connect_future = tcp_connect(addr, bind_to.as_ref());
  let conn_res = match peer.connection_timeout() {
@@ -97,6 +100,14 @@ where
  }
 
  stream.set_nodelay()?;
+
+ let digest = SocketDigest::from_raw_fd(stream.as_raw_fd());
+ digest
+ .peer_addr
+ .set(Some(peer_addr.clone()))
+ .expect("newly created OnceCell must be empty");
+ stream.set_socket_digest(digest);
+
  Ok(stream)
 }
 

pingora-core/src/connectors/mod.rs

@@ -360,9 +360,9 @@ fn test_reusable_stream(stream: &mut Stream) -> bool {
 #[cfg(test)]
 mod tests {
  use pingora_error::ErrorType;
- use pingora_openssl::ssl::SslMethod;
 
  use super::*;
+ use crate::tls::ssl::SslMethod;
  use crate::upstreams::peer::BasicPeer;
 
  // 192.0.2.1 is effectively a black hole
@@ -427,7 +427,7 @@ mod tests {
 
  let stream = connector.new_stream(&peer).await;
  let error = stream.unwrap_err();
- // XXX: some system will allow the socket to bind and connect without error, only to timeout
+ // XXX: some systems will allow the socket to bind and connect without error, only to timeout
  assert!(error.etype() == &ConnectError || error.etype() == &ConnectTimedout)
  }
 

pingora-core/src/lib.rs

@@ -34,7 +34,7 @@
 //! # Usage
 //! This crate provides low level service and protocol implementation and abstraction.
 //!
-//! If looking to build a (reverse) proxy, see `pingora-proxy` crate.
+//! If looking to build a (reverse) proxy, see [`pingora-proxy`](https://docs.rs/pingora-proxy) crate.
 //!
 //! # Optional features
 //! `boringssl`: Switch the internal TLS library from OpenSSL to BoringSSL.

pingora-core/src/protocols/digest.rs

@@ -17,18 +17,23 @@
 use std::sync::Arc;
 use std::time::SystemTime;
 
+use once_cell::sync::OnceCell;
+
+use super::l4::socket::SocketAddr;
 use super::raw_connect::ProxyDigest;
 use super::ssl::digest::SslDigest;
 
 /// The information can be extracted from a connection
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 pub struct Digest {
  /// Information regarding the TLS of this connection if any
  pub ssl_digest: Option<Arc<SslDigest>>,
  /// Timing information
  pub timing_digest: Vec<Option<TimingDigest>>,
  /// information regarding the CONNECT proxy this connection uses.
  pub proxy_digest: Option<Arc<ProxyDigest>>,
+ /// Information about underlying socket/fd of this connection
+ pub socket_digest: Option<Arc<SocketDigest>>,
 }
 
 /// The interface to return protocol related information
@@ -53,6 +58,38 @@ impl Default for TimingDigest {
  }
 }
 
+#[derive(Debug)]
+/// The interface to return socket-related information
+pub struct SocketDigest {
+ raw_fd: std::os::unix::io::RawFd,
+ /// Remote socket address
+ pub peer_addr: OnceCell<Option<SocketAddr>>,
+ /// Local socket address
+ pub local_addr: OnceCell<Option<SocketAddr>>,
+}
+
+impl SocketDigest {
+ pub fn from_raw_fd(raw_fd: std::os::unix::io::RawFd) -> SocketDigest {
+ SocketDigest {
+ raw_fd,
+ peer_addr: OnceCell::new(),
+ local_addr: OnceCell::new(),
+ }
+ }
+
+ pub fn peer_addr(&self) -> Option<&SocketAddr> {
+ self.peer_addr
+ .get_or_init(|| SocketAddr::from_raw_fd(self.raw_fd, true))
+ .as_ref()
+ }
+
+ pub fn local_addr(&self) -> Option<&SocketAddr> {
+ self.local_addr
+ .get_or_init(|| SocketAddr::from_raw_fd(self.raw_fd, false))
+ .as_ref()
+ }
+}
+
 /// The interface to return timing information
 pub trait GetTimingDigest {
  /// Return the timing for each layer from the lowest layer to upper
@@ -64,3 +101,9 @@ pub trait GetProxyDigest {
  fn get_proxy_digest(&self) -> Option<Arc<ProxyDigest>>;
  fn set_proxy_digest(&mut self, _digest: ProxyDigest) {}
 }
+
+/// The interface to set or return socket information
+pub trait GetSocketDigest {
+ fn get_socket_digest(&self) -> Option<Arc<SocketDigest>>;
+ fn set_socket_digest(&mut self, _socket_digest: SocketDigest) {}
+}

pingora-core/src/protocols/http/client.rs

@@ -19,7 +19,7 @@ use std::time::Duration;
 
 use super::v1::client::HttpSession as Http1Session;
 use super::v2::client::Http2Session;
-use crate::protocols::Digest;
+use crate::protocols::{Digest, SocketAddr};
 
 /// A type for Http client session. It can be either an Http1 connection or an Http2 stream.
 pub enum HttpSession {
@@ -151,11 +151,27 @@ impl HttpSession {
  /// Return the [Digest] of the connection
  ///
  /// For reused connection, the timing in the digest will reflect its initial handshakes
- /// The caller should check if the connection is reused to avoid misuse the timing field
+ /// The caller should check if the connection is reused to avoid misuse of the timing field
  pub fn digest(&self) -> Option<&Digest> {
  match self {
  Self::H1(s) => Some(s.digest()),
  Self::H2(s) => s.digest(),
  }
  }
+
+ /// Return the server (peer) address of the connection.
+ pub fn server_addr(&self) -> Option<&SocketAddr> {
+ match self {
+ Self::H1(s) => s.server_addr(),
+ Self::H2(s) => s.server_addr(),
+ }
+ }
+
+ /// Return the client (local) address of the connection.
+ pub fn client_addr(&self) -> Option<&SocketAddr> {
+ match self {
+ Self::H1(s) => s.client_addr(),
+ Self::H2(s) => s.client_addr(),
+ }
+ }
 }