mirror of https://github.com/istio/ztunnel.git
Compare commits
33 Commits
Author | SHA1 | Date |
---|---|---|
|
85a94b6cc4 | |
|
3fa6335035 | |
|
dfa3b58bbc | |
|
c2d2534edb | |
|
84f0e52e64 | |
|
f030073f2f | |
|
3233bb1017 | |
|
7df8cf5d08 | |
|
7cddb868e9 | |
|
ac477c15a8 | |
|
5d0352588c | |
|
b86fd9989b | |
|
facd9a28a0 | |
|
224b2c34ac | |
|
c52e0bbdbf | |
|
442923910b | |
|
c616a29092 | |
|
d6d3b606ed | |
|
8d9a56a416 | |
|
615277a05a | |
|
3d1223af09 | |
|
79dfd10249 | |
|
9f6ae51005 | |
|
46acf76463 | |
|
58cf2a0f94 | |
|
9c01d1276d | |
|
d9ea32ce21 | |
|
c96dd032da | |
|
903cf079de | |
|
ad8bea43ef | |
|
6eaa32e8ac | |
|
3470f4bba2 | |
|
93a0973175 |
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"name": "istio build-tools",
|
||||
"image": "gcr.io/istio-testing/build-tools:master-fcd42145fc132acd1e8f607e9e7aca15058e9fb9",
|
||||
"image": "gcr.io/istio-testing/build-tools:master-8e6480403f5cf4c9a4cd9d65174d01850e632e1a",
|
||||
"privileged": true,
|
||||
"remoteEnv": {
|
||||
"USE_GKE_GCLOUD_AUTH_PLUGIN": "True",
|
||||
|
|
|
@ -112,29 +112,13 @@ version = "0.7.6"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs"
|
||||
version = "0.6.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5493c3bedbacf7fd7382c6346bbd66687d12bbaad3a89a2d2c303ee6cf20b048"
|
||||
dependencies = [
|
||||
"asn1-rs-derive 0.5.1",
|
||||
"asn1-rs-impl",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"num-traits",
|
||||
"rusticata-macros",
|
||||
"thiserror 1.0.69",
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs"
|
||||
version = "0.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "56624a96882bb8c26d61312ae18cb45868e5a9992ea73c58e45c3101e56a1e60"
|
||||
dependencies = [
|
||||
"asn1-rs-derive 0.6.0",
|
||||
"asn1-rs-derive",
|
||||
"asn1-rs-impl",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
|
@ -144,18 +128,6 @@ dependencies = [
|
|||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs-derive"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "965c2d33e53cb6b267e148a4cb0760bc01f4904c1cd4bb4002a085bb016d1490"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.100",
|
||||
"synstructure",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "asn1-rs-derive"
|
||||
version = "0.6.0"
|
||||
|
@ -761,27 +733,13 @@ dependencies = [
|
|||
"const-oid",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "der-parser"
|
||||
version = "9.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553"
|
||||
dependencies = [
|
||||
"asn1-rs 0.6.2",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"num-bigint",
|
||||
"num-traits",
|
||||
"rusticata-macros",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "der-parser"
|
||||
version = "10.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "07da5016415d5a3c4dd39b11ed26f915f52fc4e0dc197d87908bc916e51bc1a6"
|
||||
dependencies = [
|
||||
"asn1-rs 0.7.1",
|
||||
"asn1-rs",
|
||||
"displaydoc",
|
||||
"nom",
|
||||
"num-bigint",
|
||||
|
@ -1844,7 +1802,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"windows-targets 0.52.6",
|
||||
"windows-targets 0.48.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
@ -2226,22 +2184,13 @@ dependencies = [
|
|||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "oid-registry"
|
||||
version = "0.7.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a8d8034d9489cdaf79228eb9f6a3b8d7bb32ba00d6645ebd48eef4077ceb5bd9"
|
||||
dependencies = [
|
||||
"asn1-rs 0.6.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "oid-registry"
|
||||
version = "0.8.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "12f40cff3dde1b6087cc5d5f5d4d65712f34016a03ed60e9c08dcc392736b5b7"
|
||||
dependencies = [
|
||||
"asn1-rs 0.7.1",
|
||||
"asn1-rs",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
@ -2815,16 +2764,16 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "rcgen"
|
||||
version = "0.13.2"
|
||||
version = "0.13.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2"
|
||||
checksum = "887a643fa081058097896d87764863994f6c32a1716e76adc479bd283974a825"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"pem",
|
||||
"ring",
|
||||
"rustls-pki-types",
|
||||
"time",
|
||||
"x509-parser 0.16.0",
|
||||
"x509-parser",
|
||||
"yasna",
|
||||
]
|
||||
|
||||
|
@ -4305,36 +4254,18 @@ version = "0.5.5"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51"
|
||||
|
||||
[[package]]
|
||||
name = "x509-parser"
|
||||
version = "0.16.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fcbc162f30700d6f3f82a24bf7cc62ffe7caea42c0b2cba8bf7f3ae50cf51f69"
|
||||
dependencies = [
|
||||
"asn1-rs 0.6.2",
|
||||
"data-encoding",
|
||||
"der-parser 9.0.0",
|
||||
"lazy_static",
|
||||
"nom",
|
||||
"oid-registry 0.7.1",
|
||||
"ring",
|
||||
"rusticata-macros",
|
||||
"thiserror 1.0.69",
|
||||
"time",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "x509-parser"
|
||||
version = "0.17.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4569f339c0c402346d4a75a9e39cf8dad310e287eef1ff56d4c68e5067f53460"
|
||||
dependencies = [
|
||||
"asn1-rs 0.7.1",
|
||||
"asn1-rs",
|
||||
"data-encoding",
|
||||
"der-parser 10.0.0",
|
||||
"der-parser",
|
||||
"lazy_static",
|
||||
"nom",
|
||||
"oid-registry 0.8.1",
|
||||
"oid-registry",
|
||||
"ring",
|
||||
"rusticata-macros",
|
||||
"thiserror 2.0.12",
|
||||
|
@ -4510,7 +4441,8 @@ dependencies = [
|
|||
"matches",
|
||||
"netns-rs",
|
||||
"nix 0.29.0",
|
||||
"oid-registry 0.8.1",
|
||||
"num_cpus",
|
||||
"oid-registry",
|
||||
"once_cell",
|
||||
"openssl",
|
||||
"pin-project-lite",
|
||||
|
@ -4553,6 +4485,6 @@ dependencies = [
|
|||
"tracing-log",
|
||||
"tracing-subscriber",
|
||||
"url",
|
||||
"x509-parser 0.17.0",
|
||||
"x509-parser",
|
||||
"ztunnel",
|
||||
]
|
||||
|
|
|
@ -71,8 +71,9 @@ itertools = "0.14"
|
|||
keyed_priority_queue = "0.4"
|
||||
libc = "0.2"
|
||||
log = "0.4"
|
||||
nix = { version = "0.29", features = ["socket", "sched", "uio", "fs", "ioctl", "user", "net", "mount"] }
|
||||
nix = { version = "0.29", features = ["socket", "sched", "uio", "fs", "ioctl", "user", "net", "mount", "resource" ] }
|
||||
once_cell = "1.21"
|
||||
num_cpus = "1.16"
|
||||
ppp = "2.3"
|
||||
prometheus-client = { version = "0.23" }
|
||||
prometheus-parse = "0.2"
|
||||
|
|
|
@ -1 +1 @@
|
|||
a1d5c4198ab79a14c09c034f2d95245efa3e2bcb
|
||||
d46067e1a8ba3db4abe2635af5807f00ba1981e6
|
||||
|
|
|
@ -32,7 +32,7 @@ set -x
|
|||
####################################################################
|
||||
|
||||
# DEFAULT_KIND_IMAGE is used to set the Kubernetes version for KinD unless overridden in params to setup_kind_cluster(s)
|
||||
DEFAULT_KIND_IMAGE="gcr.io/istio-testing/kind-node:v1.32.0"
|
||||
DEFAULT_KIND_IMAGE="gcr.io/istio-testing/kind-node:v1.33.1"
|
||||
|
||||
# the default kind cluster should be ipv4 if not otherwise specified
|
||||
KIND_IP_FAMILY="${KIND_IP_FAMILY:-ipv4}"
|
||||
|
|
|
@ -75,7 +75,7 @@ fi
|
|||
TOOLS_REGISTRY_PROVIDER=${TOOLS_REGISTRY_PROVIDER:-gcr.io}
|
||||
PROJECT_ID=${PROJECT_ID:-istio-testing}
|
||||
if [[ "${IMAGE_VERSION:-}" == "" ]]; then
|
||||
IMAGE_VERSION=master-fcd42145fc132acd1e8f607e9e7aca15058e9fb9
|
||||
IMAGE_VERSION=master-8e6480403f5cf4c9a4cd9d65174d01850e632e1a
|
||||
fi
|
||||
if [[ "${IMAGE_NAME:-}" == "" ]]; then
|
||||
IMAGE_NAME=build-tools
|
||||
|
|
|
@ -3843,6 +3843,7 @@ dependencies = [
|
|||
"log",
|
||||
"netns-rs",
|
||||
"nix 0.29.0",
|
||||
"num_cpus",
|
||||
"once_cell",
|
||||
"pin-project-lite",
|
||||
"pingora-pool",
|
||||
|
|
22
src/app.rs
22
src/app.rs
|
@ -136,6 +136,25 @@ pub async fn build_with_cert(
|
|||
|
||||
if config.proxy_mode == config::ProxyMode::Shared {
|
||||
tracing::info!("shared proxy mode - in-pod mode enabled");
|
||||
|
||||
// Create ztunnel inbound listener only if its specific identity and workload info are configured.
|
||||
if let Some(inbound) = proxy_gen.create_ztunnel_self_proxy_listener().await? {
|
||||
// Run the inbound listener in the data plane worker pool
|
||||
let mut xds_rx_for_inbound = xds_rx.clone();
|
||||
data_plane_pool.send(DataPlaneTask {
|
||||
block_shutdown: true,
|
||||
fut: Box::pin(async move {
|
||||
tracing::info!("Starting ztunnel inbound listener task");
|
||||
let _ = xds_rx_for_inbound.changed().await;
|
||||
tokio::task::spawn(async move {
|
||||
inbound.run().in_current_span().await;
|
||||
})
|
||||
.await?;
|
||||
Ok(())
|
||||
}),
|
||||
})?;
|
||||
}
|
||||
|
||||
let run_future = init_inpod_proxy_mgr(
|
||||
&mut registry,
|
||||
&mut admin_server,
|
||||
|
@ -247,7 +266,8 @@ fn new_data_plane_pool(num_worker_threads: usize) -> mpsc::Sender<DataPlaneTask>
|
|||
.thread_name_fn(|| {
|
||||
static ATOMIC_ID: AtomicUsize = AtomicUsize::new(0);
|
||||
let id = ATOMIC_ID.fetch_add(1, Ordering::SeqCst);
|
||||
format!("ztunnel-proxy-{id}")
|
||||
// Thread name can only be 16 chars so keep it short
|
||||
format!("ztunnel-{id}")
|
||||
})
|
||||
.enable_all()
|
||||
.build()
|
||||
|
|
160
src/config.rs
160
src/config.rs
|
@ -54,10 +54,12 @@ const LOCAL_XDS_PATH: &str = "LOCAL_XDS_PATH";
|
|||
const LOCAL_XDS: &str = "LOCAL_XDS";
|
||||
const XDS_ON_DEMAND: &str = "XDS_ON_DEMAND";
|
||||
const XDS_ADDRESS: &str = "XDS_ADDRESS";
|
||||
const PREFERED_SERVICE_NAMESPACE: &str = "PREFERED_SERVICE_NAMESPACE";
|
||||
const CA_ADDRESS: &str = "CA_ADDRESS";
|
||||
const SECRET_TTL: &str = "SECRET_TTL";
|
||||
const FAKE_CA: &str = "FAKE_CA";
|
||||
const ZTUNNEL_WORKER_THREADS: &str = "ZTUNNEL_WORKER_THREADS";
|
||||
const ZTUNNEL_CPU_LIMIT: &str = "ZTUNNEL_CPU_LIMIT";
|
||||
const POOL_MAX_STREAMS_PER_CONNECTION: &str = "POOL_MAX_STREAMS_PER_CONNECTION";
|
||||
const POOL_UNUSED_RELEASE_TIMEOUT: &str = "POOL_UNUSED_RELEASE_TIMEOUT";
|
||||
// CONNECTION_TERMINATION_DEADLINE configures an explicit deadline
|
||||
|
@ -70,6 +72,10 @@ const ENABLE_ORIG_SRC: &str = "ENABLE_ORIG_SRC";
|
|||
const PROXY_CONFIG: &str = "PROXY_CONFIG";
|
||||
const IPV6_ENABLED: &str = "IPV6_ENABLED";
|
||||
|
||||
const HTTP2_STREAM_WINDOW_SIZE: &str = "HTTP2_STREAM_WINDOW_SIZE";
|
||||
const HTTP2_CONNECTION_WINDOW_SIZE: &str = "HTTP2_CONNECTION_WINDOW_SIZE";
|
||||
const HTTP2_FRAME_SIZE: &str = "HTTP2_FRAME_SIZE";
|
||||
|
||||
const UNSTABLE_ENABLE_SOCKS5: &str = "UNSTABLE_ENABLE_SOCKS5";
|
||||
|
||||
const DEFAULT_WORKER_THREADS: u16 = 2;
|
||||
|
@ -237,6 +243,12 @@ pub struct Config {
|
|||
// Allow custom alternative XDS hostname verification
|
||||
pub alt_xds_hostname: Option<String>,
|
||||
|
||||
/// Prefered service namespace to use for service resolution.
|
||||
/// If unset, local namespaces is preferred and other namespaces have equal priority.
|
||||
/// If set, the local namespace is preferred, then the defined prefered_service_namespace
|
||||
/// and finally other namespaces at an equal priority.
|
||||
pub prefered_service_namespace: Option<String>,
|
||||
|
||||
/// TTL for CSR requests
|
||||
pub secret_ttl: Duration,
|
||||
/// YAML config for local XDS workloads
|
||||
|
@ -293,6 +305,10 @@ pub struct Config {
|
|||
|
||||
// If true, when AppTunnel is set for
|
||||
pub localhost_app_tunnel: bool,
|
||||
|
||||
pub ztunnel_identity: Option<identity::Identity>,
|
||||
|
||||
pub ztunnel_workload: Option<state::WorkloadInfo>,
|
||||
}
|
||||
|
||||
#[derive(serde::Serialize, Clone, Copy, Debug)]
|
||||
|
@ -399,6 +415,60 @@ fn parse_headers(prefix: &str) -> Result<MetadataVector, Error> {
|
|||
Ok(metadata)
|
||||
}
|
||||
|
||||
fn get_cpu_count() -> Result<usize, Error> {
|
||||
// Allow overriding the count with an env var. This can be used to pass the CPU limit on Kubernetes
|
||||
// from the downward API.
|
||||
// Note the downward API will return the total thread count ("logical cores") if no limit is set,
|
||||
// so it is really the same as num_cpus.
|
||||
// We allow num_cpus for cases its not set (not on Kubernetes, etc).
|
||||
match parse::<usize>(ZTUNNEL_CPU_LIMIT)? {
|
||||
Some(limit) => Ok(limit),
|
||||
// This is *logical cores*
|
||||
None => Ok(num_cpus::get()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse worker threads configuration, supporting both fixed numbers and percentages
|
||||
fn parse_worker_threads(default: usize) -> Result<usize, Error> {
|
||||
match parse::<String>(ZTUNNEL_WORKER_THREADS)? {
|
||||
Some(value) => {
|
||||
if let Some(percent_str) = value.strip_suffix('%') {
|
||||
// Parse as percentage
|
||||
let percent: f64 = percent_str.parse().map_err(|e| {
|
||||
Error::EnvVar(
|
||||
ZTUNNEL_WORKER_THREADS.to_string(),
|
||||
value.clone(),
|
||||
format!("invalid percentage: {}", e),
|
||||
)
|
||||
})?;
|
||||
|
||||
if percent <= 0.0 || percent > 100.0 {
|
||||
return Err(Error::EnvVar(
|
||||
ZTUNNEL_WORKER_THREADS.to_string(),
|
||||
value,
|
||||
"percentage must be between 0 and 100".to_string(),
|
||||
));
|
||||
}
|
||||
|
||||
let cpu_count = get_cpu_count()?;
|
||||
// Round up, minimum of 1
|
||||
let threads = ((cpu_count as f64 * percent / 100.0).ceil() as usize).max(1);
|
||||
Ok(threads)
|
||||
} else {
|
||||
// Parse as fixed number
|
||||
value.parse::<usize>().map_err(|e| {
|
||||
Error::EnvVar(
|
||||
ZTUNNEL_WORKER_THREADS.to_string(),
|
||||
value,
|
||||
format!("invalid number: {}", e),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
None => Ok(default),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn parse_config() -> Result<Config, Error> {
|
||||
let pc = parse_proxy_config()?;
|
||||
construct_config(pc)
|
||||
|
@ -438,6 +508,14 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
.or_else(|| Some(default_istiod_address.clone())),
|
||||
))?;
|
||||
|
||||
let prefered_service_namespace = match parse::<String>(PREFERED_SERVICE_NAMESPACE) {
|
||||
Ok(ns) => ns,
|
||||
Err(e) => {
|
||||
warn!(err=?e, "failed to parse {PREFERED_SERVICE_NAMESPACE}, continuing with default behavior");
|
||||
None
|
||||
}
|
||||
};
|
||||
|
||||
let istio_meta_cluster_id = ISTIO_META_PREFIX.to_owned() + CLUSTER_ID;
|
||||
let cluster_id: String = match parse::<String>(&istio_meta_cluster_id)? {
|
||||
Some(id) => id,
|
||||
|
@ -600,6 +678,29 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
|
||||
let socket_config_defaults = SocketConfig::default();
|
||||
|
||||
// Read ztunnel identity and workload info from Downward API if available
|
||||
let (ztunnel_identity, ztunnel_workload) = match (
|
||||
parse::<String>("POD_NAMESPACE")?,
|
||||
parse::<String>("SERVICE_ACCOUNT")?,
|
||||
parse::<String>("POD_NAME")?,
|
||||
) {
|
||||
(Some(namespace), Some(service_account), Some(pod_name)) => {
|
||||
let trust_domain = std::env::var("TRUST_DOMAIN")
|
||||
.unwrap_or_else(|_| crate::identity::manager::DEFAULT_TRUST_DOMAIN.to_string());
|
||||
|
||||
let identity = identity::Identity::from_parts(
|
||||
trust_domain.into(),
|
||||
namespace.clone().into(),
|
||||
service_account.clone().into(),
|
||||
);
|
||||
|
||||
let workload = state::WorkloadInfo::new(pod_name, namespace, service_account);
|
||||
|
||||
(Some(identity), Some(workload))
|
||||
}
|
||||
_ => (None, None),
|
||||
};
|
||||
|
||||
validate_config(Config {
|
||||
proxy: parse_default(ENABLE_PROXY, true)?,
|
||||
// Enable by default; running the server is not an issue, clients still need to opt-in to sending their
|
||||
|
@ -619,9 +720,15 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
DEFAULT_POOL_UNUSED_RELEASE_TIMEOUT,
|
||||
)?,
|
||||
|
||||
window_size: 4 * 1024 * 1024,
|
||||
connection_window_size: 4 * 1024 * 1024,
|
||||
frame_size: 1024 * 1024,
|
||||
// window size: per-stream limit
|
||||
window_size: parse_default(HTTP2_STREAM_WINDOW_SIZE, 4 * 1024 * 1024)?,
|
||||
// connection window size: per connection.
|
||||
// Setting this to the same value as window_size can introduce deadlocks in some applications
|
||||
// where clients do not read data on streamA until they receive data on streamB.
|
||||
// If streamA consumes the entire connection window, we enter a deadlock.
|
||||
// A 4x limit should be appropriate without introducing too much potential buffering.
|
||||
connection_window_size: parse_default(HTTP2_CONNECTION_WINDOW_SIZE, 16 * 1024 * 1024)?,
|
||||
frame_size: parse_default(HTTP2_FRAME_SIZE, 1024 * 1024)?,
|
||||
|
||||
self_termination_deadline: match parse_duration(CONNECTION_TERMINATION_DEADLINE)? {
|
||||
Some(period) => period,
|
||||
|
@ -675,6 +782,7 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
|
||||
xds_address,
|
||||
xds_root_cert,
|
||||
prefered_service_namespace,
|
||||
ca_address,
|
||||
ca_root_cert,
|
||||
alt_xds_hostname: parse(ALT_XDS_HOSTNAME)?,
|
||||
|
@ -688,8 +796,7 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
fake_ca,
|
||||
auth,
|
||||
|
||||
num_worker_threads: parse_default(
|
||||
ZTUNNEL_WORKER_THREADS,
|
||||
num_worker_threads: parse_worker_threads(
|
||||
pc.concurrency.unwrap_or(DEFAULT_WORKER_THREADS).into(),
|
||||
)?,
|
||||
|
||||
|
@ -753,6 +860,8 @@ pub fn construct_config(pc: ProxyConfig) -> Result<Config, Error> {
|
|||
ca_headers: parse_headers(ISTIO_CA_HEADER_PREFIX)?,
|
||||
|
||||
localhost_app_tunnel: parse_default(LOCALHOST_APP_TUNNEL, true)?,
|
||||
ztunnel_identity,
|
||||
ztunnel_workload,
|
||||
})
|
||||
}
|
||||
|
||||
|
@ -1055,4 +1164,45 @@ pub mod tests {
|
|||
assert!(metadata.vec.contains(&(key, value)));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_parse_worker_threads() {
|
||||
unsafe {
|
||||
// Test fixed number
|
||||
env::set_var(ZTUNNEL_WORKER_THREADS, "4");
|
||||
assert_eq!(parse_worker_threads(2).unwrap(), 4);
|
||||
|
||||
// Test percentage with CPU limit
|
||||
env::set_var(ZTUNNEL_CPU_LIMIT, "8");
|
||||
env::set_var(ZTUNNEL_WORKER_THREADS, "50%");
|
||||
assert_eq!(parse_worker_threads(2).unwrap(), 4); // 50% of 8 CPUs = 4 threads
|
||||
|
||||
// Test percentage with CPU limit
|
||||
env::set_var(ZTUNNEL_CPU_LIMIT, "16");
|
||||
env::set_var(ZTUNNEL_WORKER_THREADS, "30%");
|
||||
assert_eq!(parse_worker_threads(2).unwrap(), 5); // Round up to 5
|
||||
|
||||
// Test low percentage that rounds up to 1
|
||||
env::set_var(ZTUNNEL_CPU_LIMIT, "4");
|
||||
env::set_var(ZTUNNEL_WORKER_THREADS, "10%");
|
||||
assert_eq!(parse_worker_threads(2).unwrap(), 1); // 10% of 4 CPUs = 0.4, rounds up to 1
|
||||
|
||||
// Test default when no env var is set
|
||||
env::remove_var(ZTUNNEL_WORKER_THREADS);
|
||||
assert_eq!(parse_worker_threads(2).unwrap(), 2);
|
||||
|
||||
// Test without CPU limit (should use system CPU count)
|
||||
env::remove_var(ZTUNNEL_CPU_LIMIT);
|
||||
let system_cpus = num_cpus::get();
|
||||
assert_eq!(get_cpu_count().unwrap(), system_cpus);
|
||||
|
||||
// Test with CPU limit
|
||||
env::set_var(ZTUNNEL_CPU_LIMIT, "12");
|
||||
assert_eq!(get_cpu_count().unwrap(), 12);
|
||||
|
||||
// Clean up
|
||||
env::remove_var(ZTUNNEL_WORKER_THREADS);
|
||||
env::remove_var(ZTUNNEL_CPU_LIMIT);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -47,7 +47,7 @@ use crate::drain::{DrainMode, DrainWatcher};
|
|||
use crate::metrics::{DeferRecorder, IncrementRecorder, Recorder};
|
||||
use crate::proxy::Error;
|
||||
use crate::state::DemandProxyState;
|
||||
use crate::state::service::IpFamily;
|
||||
use crate::state::service::{IpFamily, Service};
|
||||
use crate::state::workload::Workload;
|
||||
use crate::state::workload::address::Address;
|
||||
use crate::{config, dns};
|
||||
|
@ -85,6 +85,7 @@ impl Server {
|
|||
drain: DrainWatcher,
|
||||
socket_factory: &(dyn SocketFactory + Send + Sync),
|
||||
local_workload_information: Arc<LocalWorkloadFetcher>,
|
||||
prefered_service_namespace: Option<String>,
|
||||
) -> Result<Self, Error> {
|
||||
// if the address we got from config is supposed to be v6-enabled,
|
||||
// actually check if the local pod context our socketfactory operates in supports V6.
|
||||
|
@ -102,6 +103,7 @@ impl Server {
|
|||
forwarder,
|
||||
metrics,
|
||||
local_workload_information,
|
||||
prefered_service_namespace,
|
||||
);
|
||||
let store = Arc::new(store);
|
||||
let handler = dns::handler::Handler::new(store.clone());
|
||||
|
@ -191,6 +193,7 @@ struct Store {
|
|||
svc_domain: Name,
|
||||
metrics: Arc<Metrics>,
|
||||
local_workload: Arc<LocalWorkloadFetcher>,
|
||||
prefered_service_namespace: Option<String>,
|
||||
}
|
||||
|
||||
impl Store {
|
||||
|
@ -200,6 +203,7 @@ impl Store {
|
|||
forwarder: Arc<dyn Forwarder>,
|
||||
metrics: Arc<Metrics>,
|
||||
local_workload_information: Arc<LocalWorkloadFetcher>,
|
||||
prefered_service_namespace: Option<String>,
|
||||
) -> Self {
|
||||
let domain = as_name(domain);
|
||||
let svc_domain = append_name(as_name("svc"), &domain);
|
||||
|
@ -211,6 +215,7 @@ impl Store {
|
|||
svc_domain,
|
||||
metrics,
|
||||
local_workload: local_workload_information,
|
||||
prefered_service_namespace,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -359,7 +364,7 @@ impl Store {
|
|||
let search_name_str = search_name.to_string().into();
|
||||
search_name.set_fqdn(true);
|
||||
|
||||
let service = state
|
||||
let services: Vec<Arc<Service>> = state
|
||||
.services
|
||||
.get_by_host(&search_name_str)
|
||||
.iter()
|
||||
|
@ -382,13 +387,30 @@ impl Store {
|
|||
})
|
||||
// Get the service matching the client namespace. If no match exists, just
|
||||
// return the first service.
|
||||
.find_or_first(|service| service.namespace == client.namespace)
|
||||
.cloned();
|
||||
// .find_or_first(|service| service.namespace == client.namespace)
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
// TODO: ideally we'd sort these by creation time so that the oldest would be used if there are no namespace matches
|
||||
// presently service doesn't have creation time in WDS, but we could add it
|
||||
// TODO: if the local namespace doesn't define a service, kube service should be prioritized over se
|
||||
let service = match services
|
||||
.iter()
|
||||
.find(|service| service.namespace == client.namespace)
|
||||
{
|
||||
Some(service) => Some(service),
|
||||
None => match self.prefered_service_namespace.as_ref() {
|
||||
Some(prefered_namespace) => services.iter().find_or_first(|service| {
|
||||
service.namespace == prefered_namespace.as_str()
|
||||
}),
|
||||
None => services.first(),
|
||||
},
|
||||
};
|
||||
|
||||
// First, lookup the host as a service.
|
||||
if let Some(service) = service {
|
||||
return Some(ServerMatch {
|
||||
server: Address::Service(service),
|
||||
server: Address::Service(service.clone()),
|
||||
name: search_name,
|
||||
alias,
|
||||
});
|
||||
|
@ -564,7 +586,12 @@ impl Resolver for Store {
|
|||
}
|
||||
Err(e) => {
|
||||
// Forwarding failed. Just return the error.
|
||||
access_log(request, Some(&client), "forwarding failed", 0);
|
||||
access_log(
|
||||
request,
|
||||
Some(&client),
|
||||
&format!("forwarding failed ({e})"),
|
||||
0,
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
|
@ -589,7 +616,12 @@ impl Resolver for Store {
|
|||
}
|
||||
Err(e) => {
|
||||
// Forwarding failed. Just return the error.
|
||||
access_log(request, Some(&client), "forwarding failed", 0);
|
||||
access_log(
|
||||
request,
|
||||
Some(&client),
|
||||
&format!("forwarding failed ({e})"),
|
||||
0,
|
||||
);
|
||||
return Err(e);
|
||||
}
|
||||
}
|
||||
|
@ -946,6 +978,7 @@ mod tests {
|
|||
|
||||
const NS1: &str = "ns1";
|
||||
const NS2: &str = "ns2";
|
||||
const PREFERRED: &str = "preferred-ns";
|
||||
const NW1: Strng = strng::literal!("nw1");
|
||||
const NW2: Strng = strng::literal!("nw2");
|
||||
|
||||
|
@ -1053,6 +1086,7 @@ mod tests {
|
|||
forwarder,
|
||||
metrics: test_metrics(),
|
||||
local_workload,
|
||||
prefered_service_namespace: None,
|
||||
};
|
||||
|
||||
let namespaced_domain = n(format!("{}.svc.cluster.local", c.client_namespace));
|
||||
|
@ -1368,6 +1402,18 @@ mod tests {
|
|||
expect_code: ResponseCode::NXDomain,
|
||||
..Default::default()
|
||||
},
|
||||
Case {
|
||||
name: "success: preferred namespace is chosen if local namespace is not defined",
|
||||
host: "preferred.io.",
|
||||
expect_records: vec![a(n("preferred.io."), ipv4("10.10.10.211"))],
|
||||
..Default::default()
|
||||
},
|
||||
Case {
|
||||
name: "success: external service resolves to local namespace's address",
|
||||
host: "everywhere.io.",
|
||||
expect_records: vec![a(n("everywhere.io."), ipv4("10.10.10.112"))],
|
||||
..Default::default()
|
||||
},
|
||||
];
|
||||
|
||||
// Create and start the proxy.
|
||||
|
@ -1385,6 +1431,7 @@ mod tests {
|
|||
drain,
|
||||
&factory,
|
||||
local_workload,
|
||||
Some(PREFERRED.to_string()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -1471,6 +1518,7 @@ mod tests {
|
|||
drain,
|
||||
&factory,
|
||||
local_workload,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -1520,6 +1568,7 @@ mod tests {
|
|||
}),
|
||||
state.clone(),
|
||||
),
|
||||
prefered_service_namespace: None,
|
||||
};
|
||||
|
||||
let ip4n6_client_ip = ip("::ffff:202:202");
|
||||
|
@ -1553,6 +1602,7 @@ mod tests {
|
|||
drain,
|
||||
&factory,
|
||||
local_workload,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
@ -1669,6 +1719,16 @@ mod tests {
|
|||
xds_external_service("www.google.com", &[na(NW1, "1.1.1.1")]),
|
||||
xds_service("productpage", NS1, &[na(NW1, "9.9.9.9")]),
|
||||
xds_service("example", NS2, &[na(NW1, "10.10.10.10")]),
|
||||
// Service with the same name in another namespace
|
||||
// This should not be used if the preferred service namespace is set
|
||||
xds_namespaced_external_service("everywhere.io", NS2, &[na(NW1, "10.10.10.110")]),
|
||||
xds_namespaced_external_service("preferred.io", NS2, &[na(NW1, "10.10.10.210")]),
|
||||
// Preferred service namespace
|
||||
xds_namespaced_external_service("everywhere.io", PREFERRED, &[na(NW1, "10.10.10.111")]),
|
||||
xds_namespaced_external_service("preferred.io", PREFERRED, &[na(NW1, "10.10.10.211")]),
|
||||
// Service with the same name in the same namespace
|
||||
// Client in NS1 should use this service
|
||||
xds_namespaced_external_service("everywhere.io", NS1, &[na(NW1, "10.10.10.112")]),
|
||||
with_fqdn(
|
||||
"details.ns2.svc.cluster.remote",
|
||||
xds_service(
|
||||
|
@ -1819,9 +1879,17 @@ mod tests {
|
|||
}
|
||||
|
||||
fn xds_external_service<S: AsRef<str>>(hostname: S, addrs: &[NetworkAddress]) -> XdsService {
|
||||
xds_namespaced_external_service(hostname, NS1, addrs)
|
||||
}
|
||||
|
||||
fn xds_namespaced_external_service<S1: AsRef<str>, S2: AsRef<str>>(
|
||||
hostname: S1,
|
||||
ns: S2,
|
||||
vips: &[NetworkAddress],
|
||||
) -> XdsService {
|
||||
with_fqdn(
|
||||
hostname.as_ref(),
|
||||
xds_service(hostname.as_ref(), NS1, addrs),
|
||||
xds_service(hostname.as_ref(), ns.as_ref(), vips),
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
@ -58,10 +58,7 @@ async fn load_token(path: &PathBuf) -> io::Result<Vec<u8>> {
|
|||
let t = tokio::fs::read(path).await?;
|
||||
|
||||
if t.is_empty() {
|
||||
return Err(io::Error::new(
|
||||
io::ErrorKind::Other,
|
||||
"token file exists, but was empty",
|
||||
));
|
||||
return Err(io::Error::other("token file exists, but was empty"));
|
||||
}
|
||||
Ok(t)
|
||||
}
|
||||
|
|
|
@ -38,6 +38,9 @@ use keyed_priority_queue::KeyedPriorityQueue;
|
|||
|
||||
const CERT_REFRESH_FAILURE_RETRY_DELAY_MAX_INTERVAL: Duration = Duration::from_secs(150);
|
||||
|
||||
/// Default trust domain to use if not otherwise specified.
|
||||
pub const DEFAULT_TRUST_DOMAIN: &str = "cluster.local";
|
||||
|
||||
#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Hash)]
|
||||
pub enum Identity {
|
||||
Spiffe {
|
||||
|
@ -130,11 +133,10 @@ impl Identity {
|
|||
#[cfg(any(test, feature = "testing"))]
|
||||
impl Default for Identity {
|
||||
fn default() -> Self {
|
||||
const TRUST_DOMAIN: &str = "cluster.local";
|
||||
const SERVICE_ACCOUNT: &str = "ztunnel";
|
||||
const NAMESPACE: &str = "istio-system";
|
||||
Identity::Spiffe {
|
||||
trust_domain: TRUST_DOMAIN.into(),
|
||||
trust_domain: DEFAULT_TRUST_DOMAIN.into(),
|
||||
namespace: NAMESPACE.into(),
|
||||
service_account: SERVICE_ACCOUNT.into(),
|
||||
}
|
||||
|
|
|
@ -12,6 +12,9 @@
|
|||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use once_cell::sync::Lazy;
|
||||
use std::env;
|
||||
|
||||
pub mod admin;
|
||||
pub mod app;
|
||||
pub mod assertions;
|
||||
|
@ -42,3 +45,7 @@ pub mod xds;
|
|||
|
||||
#[cfg(any(test, feature = "testing"))]
|
||||
pub mod test_helpers;
|
||||
|
||||
#[allow(dead_code)]
|
||||
static PQC_ENABLED: Lazy<bool> =
|
||||
Lazy::new(|| env::var("COMPLIANCE_POLICY").unwrap_or_default() == "pqc");
|
||||
|
|
24
src/main.rs
24
src/main.rs
|
@ -14,8 +14,9 @@
|
|||
|
||||
extern crate core;
|
||||
|
||||
use nix::sys::resource::{Resource, getrlimit, setrlimit};
|
||||
use std::sync::Arc;
|
||||
use tracing::info;
|
||||
use tracing::{info, warn};
|
||||
use ztunnel::*;
|
||||
|
||||
#[cfg(feature = "jemalloc")]
|
||||
|
@ -28,6 +29,26 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
|
|||
#[unsafe(export_name = "malloc_conf")]
|
||||
pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0";
|
||||
|
||||
// We use this on Unix systems to increase the number of open file descriptors
|
||||
// if possible. This is useful for high-load scenarios where the default limit
|
||||
// is too low, which can lead to droopped connections and other issues:
|
||||
// see: https://github.com/istio/ztunnel/issues/1585
|
||||
fn increase_open_files_limit() {
|
||||
#[cfg(unix)]
|
||||
if let Ok((soft_limit, hard_limit)) = getrlimit(Resource::RLIMIT_NOFILE) {
|
||||
if let Err(e) = setrlimit(Resource::RLIMIT_NOFILE, hard_limit, hard_limit) {
|
||||
warn!("failed to set file descriptor limits: {e}");
|
||||
} else {
|
||||
info!(
|
||||
"set file descriptor limits from {} to {}",
|
||||
soft_limit, hard_limit
|
||||
);
|
||||
}
|
||||
} else {
|
||||
warn!("failed to get file descriptor limits");
|
||||
}
|
||||
}
|
||||
|
||||
fn main() -> anyhow::Result<()> {
|
||||
let _log_flush = telemetry::setup_logging();
|
||||
|
||||
|
@ -74,6 +95,7 @@ fn version() -> anyhow::Result<()> {
|
|||
|
||||
async fn proxy(cfg: Arc<config::Config>) -> anyhow::Result<()> {
|
||||
info!("version: {}", version::BuildInfo::new());
|
||||
increase_open_files_limit();
|
||||
info!("running with config: {}", serde_yaml::to_string(&cfg)?);
|
||||
app::build(cfg).await?.wait_termination().await
|
||||
}
|
||||
|
|
10
src/proxy.rs
10
src/proxy.rs
|
@ -48,8 +48,9 @@ use crate::state::{DemandProxyState, WorkloadInfo};
|
|||
use crate::{config, identity, socket, tls};
|
||||
|
||||
pub mod connection_manager;
|
||||
pub mod inbound;
|
||||
|
||||
mod h2;
|
||||
mod inbound;
|
||||
mod inbound_passthrough;
|
||||
#[allow(non_camel_case_types)]
|
||||
pub mod metrics;
|
||||
|
@ -259,6 +260,8 @@ pub(super) struct ProxyInputs {
|
|||
socket_factory: Arc<dyn SocketFactory + Send + Sync>,
|
||||
local_workload_information: Arc<LocalWorkloadInformation>,
|
||||
resolver: Option<Arc<dyn Resolver + Send + Sync>>,
|
||||
// If true, inbound connections created with these inputs will not attempt to preserve the original source IP.
|
||||
pub disable_inbound_freebind: bool,
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
|
@ -271,6 +274,7 @@ impl ProxyInputs {
|
|||
socket_factory: Arc<dyn SocketFactory + Send + Sync>,
|
||||
resolver: Option<Arc<dyn Resolver + Send + Sync>>,
|
||||
local_workload_information: Arc<LocalWorkloadInformation>,
|
||||
disable_inbound_freebind: bool,
|
||||
) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
cfg,
|
||||
|
@ -280,6 +284,7 @@ impl ProxyInputs {
|
|||
socket_factory,
|
||||
local_workload_information,
|
||||
resolver,
|
||||
disable_inbound_freebind,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
@ -479,6 +484,9 @@ pub enum Error {
|
|||
#[error("requested service {0} found, but has no IP addresses")]
|
||||
NoIPForService(String),
|
||||
|
||||
#[error("no service for target address: {0}")]
|
||||
NoService(SocketAddr),
|
||||
|
||||
#[error(
|
||||
"ip addresses were resolved for workload {0}, but valid dns response had no A/AAAA records"
|
||||
)]
|
||||
|
|
|
@ -13,7 +13,7 @@
|
|||
// limitations under the License.
|
||||
|
||||
use crate::copy;
|
||||
use bytes::{BufMut, Bytes};
|
||||
use bytes::Bytes;
|
||||
use futures_core::ready;
|
||||
use h2::Reason;
|
||||
use std::io::Error;
|
||||
|
@ -85,7 +85,10 @@ pub struct H2StreamWriteHalf {
|
|||
_dropped: Option<DropCounter>,
|
||||
}
|
||||
|
||||
pub struct TokioH2Stream(H2Stream);
|
||||
pub struct TokioH2Stream {
|
||||
stream: H2Stream,
|
||||
buf: Bytes,
|
||||
}
|
||||
|
||||
struct DropCounter {
|
||||
// Whether the other end of this shared counter has already dropped.
|
||||
|
@ -144,7 +147,10 @@ impl Drop for DropCounter {
|
|||
// then the specific implementation will conflict with the generic one.
|
||||
impl TokioH2Stream {
|
||||
pub fn new(stream: H2Stream) -> Self {
|
||||
Self(stream)
|
||||
Self {
|
||||
stream,
|
||||
buf: Bytes::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -154,24 +160,21 @@ impl tokio::io::AsyncRead for TokioH2Stream {
|
|||
cx: &mut Context<'_>,
|
||||
buf: &mut tokio::io::ReadBuf<'_>,
|
||||
) -> Poll<std::io::Result<()>> {
|
||||
let pinned = std::pin::Pin::new(&mut self.0.read);
|
||||
copy::ResizeBufRead::poll_bytes(pinned, cx).map(|r| match r {
|
||||
Ok(bytes) => {
|
||||
if buf.remaining() < bytes.len() {
|
||||
Err(Error::new(
|
||||
std::io::ErrorKind::Other,
|
||||
format!(
|
||||
"kould overflow buffer of with {} remaining",
|
||||
buf.remaining()
|
||||
),
|
||||
))
|
||||
} else {
|
||||
buf.put(bytes);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
})
|
||||
// Just return the bytes we have left over and don't poll the stream because
|
||||
// its unclear what to do if there are bytes left over from the previous read, and when we
|
||||
// poll, we get an error.
|
||||
if self.buf.is_empty() {
|
||||
// If we have no unread bytes, we can poll the stream
|
||||
// and fill self.buf with the bytes we read.
|
||||
let pinned = std::pin::Pin::new(&mut self.stream.read);
|
||||
let res = ready!(copy::ResizeBufRead::poll_bytes(pinned, cx))?;
|
||||
self.buf = res;
|
||||
}
|
||||
// Copy as many bytes as we can from self.buf.
|
||||
let cnt = Ord::min(buf.remaining(), self.buf.len());
|
||||
buf.put_slice(&self.buf[..cnt]);
|
||||
self.buf = self.buf.split_off(cnt);
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -181,7 +184,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
|
|||
cx: &mut Context<'_>,
|
||||
buf: &[u8],
|
||||
) -> Poll<Result<usize, tokio::io::Error>> {
|
||||
let pinned = std::pin::Pin::new(&mut self.0.write);
|
||||
let pinned = std::pin::Pin::new(&mut self.stream.write);
|
||||
let buf = Bytes::copy_from_slice(buf);
|
||||
copy::AsyncWriteBuf::poll_write_buf(pinned, cx, buf)
|
||||
}
|
||||
|
@ -190,7 +193,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
|
|||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<Result<(), std::io::Error>> {
|
||||
let pinned = std::pin::Pin::new(&mut self.0.write);
|
||||
let pinned = std::pin::Pin::new(&mut self.stream.write);
|
||||
copy::AsyncWriteBuf::poll_flush(pinned, cx)
|
||||
}
|
||||
|
||||
|
@ -198,7 +201,7 @@ impl tokio::io::AsyncWrite for TokioH2Stream {
|
|||
mut self: Pin<&mut Self>,
|
||||
cx: &mut Context<'_>,
|
||||
) -> Poll<Result<(), std::io::Error>> {
|
||||
let pinned = std::pin::Pin::new(&mut self.0.write);
|
||||
let pinned = std::pin::Pin::new(&mut self.stream.write);
|
||||
copy::AsyncWriteBuf::poll_shutdown(pinned, cx)
|
||||
}
|
||||
}
|
||||
|
@ -302,6 +305,6 @@ fn h2_to_io_error(e: h2::Error) -> std::io::Error {
|
|||
if e.is_io() {
|
||||
e.into_io().unwrap()
|
||||
} else {
|
||||
std::io::Error::new(std::io::ErrorKind::Other, e)
|
||||
std::io::Error::other(e)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -45,7 +45,7 @@ use crate::state::{DemandProxyState, ProxyRbacContext};
|
|||
use crate::strng::Strng;
|
||||
use crate::tls::TlsError;
|
||||
|
||||
pub(super) struct Inbound {
|
||||
pub struct Inbound {
|
||||
listener: socket::Listener,
|
||||
drain: DrainWatcher,
|
||||
pi: Arc<ProxyInputs>,
|
||||
|
@ -53,7 +53,7 @@ pub(super) struct Inbound {
|
|||
}
|
||||
|
||||
impl Inbound {
|
||||
pub(super) async fn new(pi: Arc<ProxyInputs>, drain: DrainWatcher) -> Result<Inbound, Error> {
|
||||
pub(crate) async fn new(pi: Arc<ProxyInputs>, drain: DrainWatcher) -> Result<Inbound, Error> {
|
||||
let listener = pi
|
||||
.socket_factory
|
||||
.tcp_bind(pi.cfg.inbound_addr)
|
||||
|
@ -74,11 +74,12 @@ impl Inbound {
|
|||
})
|
||||
}
|
||||
|
||||
pub(super) fn address(&self) -> SocketAddr {
|
||||
/// Returns the socket address this proxy is listening on.
|
||||
pub fn address(&self) -> SocketAddr {
|
||||
self.listener.local_addr()
|
||||
}
|
||||
|
||||
pub(super) async fn run(self) {
|
||||
pub async fn run(self) {
|
||||
let pi = self.pi.clone();
|
||||
let acceptor = InboundCertProvider {
|
||||
local_workload: self.pi.local_workload_information.clone(),
|
||||
|
@ -122,7 +123,7 @@ impl Inbound {
|
|||
let conn = Connection {
|
||||
src_identity,
|
||||
src,
|
||||
dst_network: strng::new(&network), // inbound request must be on our network
|
||||
dst_network: network.clone(), // inbound request must be on our network
|
||||
dst,
|
||||
};
|
||||
debug!(%conn, "accepted connection");
|
||||
|
@ -244,10 +245,18 @@ impl Inbound {
|
|||
SocketAddr::new(loopback, ri.upstream_addr.port()),
|
||||
)
|
||||
} else {
|
||||
(
|
||||
enable_original_source.then_some(ri.rbac_ctx.conn.src.ip()),
|
||||
ri.upstream_addr,
|
||||
)
|
||||
// When ztunnel is proxying to its own internal endpoints (metrics server after HBONE termination),
|
||||
// we must not attempt to use the original external client's IP as the source for this internal connection.
|
||||
// Setting `disable_inbound_freebind` to true for such self-proxy scenarios ensures `upstream_src_ip` is `None`,
|
||||
// causing `freebind_connect` to use a local IP for the connection to ztunnel's own service.
|
||||
// For regular inbound traffic to other workloads, `disable_inbound_freebind` is false, and original source
|
||||
// preservation depends on `enable_original_source`.
|
||||
let upstream_src_ip = if pi.disable_inbound_freebind {
|
||||
None
|
||||
} else {
|
||||
enable_original_source.then_some(ri.rbac_ctx.conn.src.ip())
|
||||
};
|
||||
(upstream_src_ip, ri.upstream_addr)
|
||||
};
|
||||
|
||||
// Establish upstream connection between original source and destination
|
||||
|
@ -536,7 +545,7 @@ impl Inbound {
|
|||
|
||||
/// find_inbound_upstream determines the next hop for an inbound request.
|
||||
#[expect(clippy::type_complexity)]
|
||||
fn find_inbound_upstream(
|
||||
pub(super) fn find_inbound_upstream(
|
||||
cfg: &Config,
|
||||
state: &DemandProxyState,
|
||||
conn: &Connection,
|
||||
|
@ -545,6 +554,7 @@ impl Inbound {
|
|||
) -> Result<(SocketAddr, Option<TunnelRequest>, Vec<Arc<Service>>), Error> {
|
||||
// We always target the local workload IP as the destination. But we need to determine the port to send to.
|
||||
let target_ip = conn.dst.ip();
|
||||
|
||||
// First, fetch the actual target SocketAddr as well as all possible services this could be for.
|
||||
// Given they may request the pod directly, there may be multiple possible services; we will
|
||||
// select a final one (if any) later.
|
||||
|
@ -640,7 +650,7 @@ impl Inbound {
|
|||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
struct TunnelRequest {
|
||||
pub(super) struct TunnelRequest {
|
||||
tunnel_target: SocketAddr,
|
||||
protocol: Protocol,
|
||||
}
|
||||
|
@ -702,37 +712,36 @@ fn build_response(status: StatusCode) -> Response<()> {
|
|||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
mod tests {
|
||||
use super::{Inbound, ProxyInputs};
|
||||
use crate::{config, proxy::ConnectionManager, proxy::inbound::HboneAddress, strng};
|
||||
|
||||
use crate::{
|
||||
config,
|
||||
identity::manager::mock::new_secret_manager,
|
||||
proxy::{
|
||||
ConnectionManager, DefaultSocketFactory, LocalWorkloadInformation,
|
||||
h2::server::RequestParts, inbound::HboneAddress,
|
||||
},
|
||||
rbac::Connection,
|
||||
state::{
|
||||
self, DemandProxyState,
|
||||
self, DemandProxyState, WorkloadInfo,
|
||||
service::{Endpoint, EndpointSet, Service},
|
||||
workload::{
|
||||
ApplicationTunnel, GatewayAddress, InboundProtocol, NetworkAddress, Workload,
|
||||
application_tunnel::Protocol as AppProtocol, gatewayaddress::Destination,
|
||||
ApplicationTunnel, GatewayAddress, HealthStatus, InboundProtocol, NetworkAddress,
|
||||
NetworkMode, Workload, application_tunnel::Protocol as AppProtocol,
|
||||
gatewayaddress::Destination,
|
||||
},
|
||||
},
|
||||
test_helpers,
|
||||
strng, test_helpers,
|
||||
};
|
||||
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
|
||||
use http::{Method, Uri};
|
||||
use prometheus_client::registry::Registry;
|
||||
use std::{
|
||||
net::SocketAddr,
|
||||
sync::{Arc, RwLock},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use crate::identity::manager::mock::new_secret_manager;
|
||||
use crate::proxy::DefaultSocketFactory;
|
||||
use crate::proxy::LocalWorkloadInformation;
|
||||
use crate::proxy::h2::server::RequestParts;
|
||||
use crate::state::WorkloadInfo;
|
||||
use crate::state::workload::HealthStatus;
|
||||
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
|
||||
use http::{Method, Uri};
|
||||
use prometheus_client::registry::Registry;
|
||||
use test_case::test_case;
|
||||
|
||||
const CLIENT_POD_IP: &str = "10.0.0.1";
|
||||
|
@ -904,6 +913,7 @@ mod tests {
|
|||
sf,
|
||||
None,
|
||||
local_workload,
|
||||
false,
|
||||
));
|
||||
let inbound_request = Inbound::build_inbound_request(&pi, conn, &request_parts).await;
|
||||
match want {
|
||||
|
@ -959,7 +969,6 @@ mod tests {
|
|||
"waypoint",
|
||||
WAYPOINT_POD_IP,
|
||||
Waypoint::None,
|
||||
// the waypoint's _workload_ gets the app tunnel field
|
||||
server_waypoint.app_tunnel(),
|
||||
),
|
||||
("client", CLIENT_POD_IP, Waypoint::None, None),
|
||||
|
@ -975,6 +984,7 @@ mod tests {
|
|||
namespace: "default".into(),
|
||||
service_account: strng::format!("service-account-{name}"),
|
||||
application_tunnel: app_tunnel,
|
||||
network_mode: NetworkMode::Standard,
|
||||
..test_helpers::test_default_workload()
|
||||
});
|
||||
|
||||
|
|
|
@ -35,10 +35,10 @@ use crate::proxy::{ConnectionOpen, ConnectionResult, DerivedWorkload, metrics};
|
|||
use crate::drain::DrainWatcher;
|
||||
use crate::drain::run_with_drain;
|
||||
use crate::proxy::h2::{H2Stream, client::WorkloadKey};
|
||||
use crate::state::ServiceResolutionMode;
|
||||
use crate::state::service::ServiceDescription;
|
||||
use crate::state::service::{Service, ServiceDescription};
|
||||
use crate::state::workload::OutboundProtocol;
|
||||
use crate::state::workload::{InboundProtocol, NetworkAddress, Workload, address::Address};
|
||||
use crate::state::{ServiceResolutionMode, Upstream};
|
||||
use crate::{assertions, copy, proxy, socket};
|
||||
|
||||
use super::h2::TokioH2Stream;
|
||||
|
@ -369,6 +369,83 @@ impl OutboundConnection {
|
|||
}
|
||||
}
|
||||
|
||||
// This function is called when the select next hop is on a different network,
|
||||
// so we expect the upstream workload to have a network gatewy configured.
|
||||
//
|
||||
// When we use a gateway to reach to a workload on a remote network we have to
|
||||
// use double HBONE (HBONE incapsulated inside HBONE). The gateway will
|
||||
// terminate the outer HBONE tunnel and forward the inner HBONE to the actual
|
||||
// destination as a opaque stream of bytes and the actual destination will
|
||||
// interpret it as an HBONE connection.
|
||||
//
|
||||
// If the upstream workload does not have an E/W gateway this function returns
|
||||
// an error indicating that it could not find a valid destination.
|
||||
//
|
||||
// A note about double HBONE, in double HBONE both inner and outer HBONE use
|
||||
// destination service name as HBONE target URI.
|
||||
//
|
||||
// Having target URI in the outer HBONE tunnel allows E/W gateway to figure out
|
||||
// where to route the data next witout the need to terminate inner HBONE tunnel.
|
||||
// In other words, it could forward inner HBONE as if it's an opaque stream of
|
||||
// bytes without trying to interpret it.
|
||||
//
|
||||
// NOTE: when connecting through an E/W gateway, regardless of whether there is
|
||||
// a waypoint or not, we always use service hostname and the service port. It's
|
||||
// somewhat different from how regular HBONE works, so I'm calling it out here.
|
||||
async fn build_request_through_gateway(
|
||||
&self,
|
||||
source: Arc<Workload>,
|
||||
// next hop on the remote network that we picked as our destination.
|
||||
// It may be a local view of a Waypoint workload on remote network or
|
||||
// a local view of the service workload (when waypoint is not
|
||||
// configured).
|
||||
upstream: Upstream,
|
||||
// This is a target service we wanted to reach in the first place.
|
||||
//
|
||||
// NOTE: Crossing network boundaries is only supported for services
|
||||
// at the moment, so we should always have a service we could use.
|
||||
service: &Service,
|
||||
target: SocketAddr,
|
||||
) -> Result<Request, Error> {
|
||||
if let Some(gateway) = &upstream.workload.network_gateway {
|
||||
let gateway_upstream = self
|
||||
.pi
|
||||
.state
|
||||
.fetch_network_gateway(gateway, &source, target)
|
||||
.await?;
|
||||
let hbone_target_destination = Some(HboneAddress::SvcHostname(
|
||||
service.hostname.clone(),
|
||||
target.port(),
|
||||
));
|
||||
|
||||
debug!("built request to a destination on another network through an E/W gateway");
|
||||
Ok(Request {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
source,
|
||||
hbone_target_destination,
|
||||
actual_destination_workload: Some(gateway_upstream.workload.clone()),
|
||||
intended_destination_service: Some(ServiceDescription::from(service)),
|
||||
actual_destination: gateway_upstream.workload_socket_addr().ok_or(
|
||||
Error::NoValidDestination(Box::new((*gateway_upstream.workload).clone())),
|
||||
)?,
|
||||
// The outer tunnel of double HBONE is terminated by the E/W
|
||||
// gateway and so for the credentials of the next hop
|
||||
// (upstream_sans) we use gateway credentials.
|
||||
upstream_sans: gateway_upstream.workload_and_services_san(),
|
||||
// The inner HBONE tunnel is terminated by either the server
|
||||
// we want to reach or a Waypoint in front of it, depending on
|
||||
// the configuration. So for the final destination credentials
|
||||
// (final_sans) we use the upstream workload credentials.
|
||||
final_sans: upstream.service_sans(),
|
||||
})
|
||||
} else {
|
||||
// Do not try to send cross-network traffic without network gateway.
|
||||
Err(Error::NoValidDestination(Box::new(
|
||||
(*upstream.workload).clone(),
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
// build_request computes all information about the request we should send
|
||||
// TODO: Do we want a single lock for source and upstream...?
|
||||
async fn build_request(
|
||||
|
@ -381,7 +458,7 @@ impl OutboundConnection {
|
|||
|
||||
// If this is to-service traffic check for a service waypoint
|
||||
// Capture result of whether this is svc addressed
|
||||
let svc_addressed = if let Some(Address::Service(target_service)) = state
|
||||
let service = if let Some(Address::Service(target_service)) = state
|
||||
.fetch_address(&NetworkAddress {
|
||||
network: self.pi.cfg.network.clone(),
|
||||
address: target.ip(),
|
||||
|
@ -393,6 +470,18 @@ impl OutboundConnection {
|
|||
.fetch_service_waypoint(&target_service, &source_workload, target)
|
||||
.await?
|
||||
{
|
||||
if waypoint.workload.network != source_workload.network {
|
||||
debug!("picked a waypoint on remote network");
|
||||
return self
|
||||
.build_request_through_gateway(
|
||||
source_workload.clone(),
|
||||
waypoint,
|
||||
&target_service,
|
||||
target,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
let upstream_sans = waypoint.workload_and_services_san();
|
||||
let actual_destination =
|
||||
waypoint
|
||||
|
@ -413,10 +502,10 @@ impl OutboundConnection {
|
|||
});
|
||||
}
|
||||
// this was service addressed but we did not find a waypoint
|
||||
true
|
||||
Some(target_service)
|
||||
} else {
|
||||
// this wasn't service addressed
|
||||
false
|
||||
None
|
||||
};
|
||||
|
||||
let Some(us) = state
|
||||
|
@ -428,7 +517,7 @@ impl OutboundConnection {
|
|||
)
|
||||
.await?
|
||||
else {
|
||||
if svc_addressed {
|
||||
if service.is_some() {
|
||||
return Err(Error::NoHealthyUpstream(target));
|
||||
}
|
||||
debug!("built request as passthrough; no upstream found");
|
||||
|
@ -446,37 +535,26 @@ impl OutboundConnection {
|
|||
|
||||
// Check whether we are using an E/W gateway and sending cross network traffic
|
||||
if us.workload.network != source_workload.network {
|
||||
if let Some(ew_gtw) = &us.workload.network_gateway {
|
||||
let gtw_us = {
|
||||
self.pi
|
||||
.state
|
||||
.fetch_network_gateway(ew_gtw, &source_workload, target)
|
||||
.await?
|
||||
};
|
||||
|
||||
let svc = us
|
||||
.destination_service
|
||||
.as_ref()
|
||||
.expect("Workloads with network gateways must be service addressed.");
|
||||
let hbone_target_destination =
|
||||
Some(HboneAddress::SvcHostname(svc.hostname.clone(), us.port));
|
||||
|
||||
return Ok(Request {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
source: source_workload,
|
||||
hbone_target_destination,
|
||||
actual_destination_workload: Some(gtw_us.workload.clone()),
|
||||
intended_destination_service: us.destination_service.clone(),
|
||||
actual_destination: gtw_us.workload_socket_addr().ok_or(
|
||||
Error::NoValidDestination(Box::new((*gtw_us.workload).clone())),
|
||||
)?,
|
||||
upstream_sans: gtw_us.workload_and_services_san(),
|
||||
final_sans: us.service_sans(),
|
||||
});
|
||||
} else {
|
||||
// Do not try to send cross-network traffic without network gateway.
|
||||
return Err(Error::NoValidDestination(Box::new((*us.workload).clone())));
|
||||
}
|
||||
// Workloads on remote network must be service addressed, so if we got here
|
||||
// and we don't have a service for the original target address then it's a
|
||||
// bug either in ztunnel itself or in istiod.
|
||||
//
|
||||
// For a double HBONE protocol implementation we have to know the
|
||||
// destination service and if there is no service for the target it's a bug.
|
||||
//
|
||||
// This situation "should never happen" because for workloads fetch_upstream
|
||||
// above only checks the workloads on the same network as this ztunnel
|
||||
// instance and therefore it should not be able to find a workload on a
|
||||
// different network.
|
||||
debug_assert!(
|
||||
service.is_some(),
|
||||
"workload on remote network is not service addressed"
|
||||
);
|
||||
debug!("picked a workload on remote network");
|
||||
let service = service.as_ref().ok_or(Error::NoService(target))?;
|
||||
return self
|
||||
.build_request_through_gateway(source_workload.clone(), us, service, target)
|
||||
.await;
|
||||
}
|
||||
|
||||
// We are not using a network gateway and there is no workload address.
|
||||
|
@ -491,7 +569,7 @@ impl OutboundConnection {
|
|||
// Check if we need to go through a workload addressed waypoint.
|
||||
// Don't traverse waypoint twice if the source is sandwich-outbound.
|
||||
// Don't traverse waypoint if traffic was addressed to a service (handled before)
|
||||
if !from_waypoint && !svc_addressed {
|
||||
if !from_waypoint && service.is_none() {
|
||||
// For case upstream server has enabled waypoint
|
||||
let waypoint = state
|
||||
.fetch_workload_waypoint(&us.workload, &source_workload, target)
|
||||
|
@ -716,6 +794,7 @@ mod tests {
|
|||
local_workload_information: local_workload_information.clone(),
|
||||
connection_manager: ConnectionManager::default(),
|
||||
resolver: None,
|
||||
disable_inbound_freebind: false,
|
||||
}),
|
||||
id: TraceParent::new(),
|
||||
pool: WorkloadHBONEPool::new(
|
||||
|
@ -815,6 +894,8 @@ mod tests {
|
|||
|
||||
#[tokio::test]
|
||||
async fn build_request_double_hbone() {
|
||||
// example.com service has a workload on remote network.
|
||||
// E/W gateway is addressed by an IP.
|
||||
run_build_request_multi(
|
||||
"127.0.0.1",
|
||||
"127.0.0.3:80",
|
||||
|
@ -866,11 +947,13 @@ mod tests {
|
|||
],
|
||||
Some(ExpectedRequest {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
hbone_destination: "example.com:8080",
|
||||
hbone_destination: "example.com:80",
|
||||
destination: "10.22.1.1:15009",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
// example.com service has a workload on remote network.
|
||||
// E/W gateway is addressed by a hostname.
|
||||
run_build_request_multi(
|
||||
"127.0.0.1",
|
||||
"127.0.0.3:80",
|
||||
|
@ -943,11 +1026,218 @@ mod tests {
|
|||
],
|
||||
Some(ExpectedRequest {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
hbone_destination: "example.com:8080",
|
||||
hbone_destination: "example.com:80",
|
||||
destination: "127.0.0.5:15008",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
// example.com service has a waypoint and waypoint workload is on remote network.
|
||||
// E/W gateway is addressed by an IP.
|
||||
run_build_request_multi(
|
||||
"127.0.0.1",
|
||||
"127.0.0.3:80",
|
||||
vec![
|
||||
XdsAddressType::Service(XdsService {
|
||||
hostname: "example.com".to_string(),
|
||||
addresses: vec![XdsNetworkAddress {
|
||||
network: "".to_string(),
|
||||
address: vec![127, 0, 0, 3],
|
||||
}],
|
||||
ports: vec![Port {
|
||||
service_port: 80,
|
||||
target_port: 8080,
|
||||
}],
|
||||
waypoint: Some(xds::istio::workload::GatewayAddress {
|
||||
destination: Some(
|
||||
xds::istio::workload::gateway_address::Destination::Hostname(
|
||||
XdsNamespacedHostname {
|
||||
namespace: Default::default(),
|
||||
hostname: "waypoint.com".into(),
|
||||
},
|
||||
),
|
||||
),
|
||||
hbone_mtls_port: 15008,
|
||||
}),
|
||||
..Default::default()
|
||||
}),
|
||||
XdsAddressType::Service(XdsService {
|
||||
hostname: "waypoint.com".to_string(),
|
||||
addresses: vec![XdsNetworkAddress {
|
||||
network: "".to_string(),
|
||||
address: vec![127, 0, 0, 4],
|
||||
}],
|
||||
ports: vec![Port {
|
||||
service_port: 15008,
|
||||
target_port: 15008,
|
||||
}],
|
||||
..Default::default()
|
||||
}),
|
||||
XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/remote-waypoint-pod".to_string(),
|
||||
addresses: vec![],
|
||||
network: "remote".to_string(),
|
||||
network_gateway: Some(xds::istio::workload::GatewayAddress {
|
||||
destination: Some(
|
||||
xds::istio::workload::gateway_address::Destination::Address(
|
||||
XdsNetworkAddress {
|
||||
network: "remote".to_string(),
|
||||
address: vec![10, 22, 1, 1],
|
||||
},
|
||||
),
|
||||
),
|
||||
hbone_mtls_port: 15009,
|
||||
}),
|
||||
services: std::collections::HashMap::from([(
|
||||
"/waypoint.com".to_string(),
|
||||
PortList {
|
||||
ports: vec![Port {
|
||||
service_port: 15008,
|
||||
target_port: 15008,
|
||||
}],
|
||||
},
|
||||
)]),
|
||||
..Default::default()
|
||||
}),
|
||||
XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/remote-ew-gtw".to_string(),
|
||||
addresses: vec![Bytes::copy_from_slice(&[10, 22, 1, 1])],
|
||||
network: "remote".to_string(),
|
||||
..Default::default()
|
||||
}),
|
||||
],
|
||||
Some(ExpectedRequest {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
hbone_destination: "example.com:80",
|
||||
destination: "10.22.1.1:15009",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_request_failover_to_remote() {
|
||||
// Similar to the double HBONE test that we already have, but it sets up a scenario when
|
||||
// load balancing logic will pick a workload on a remote cluster when local workloads are
|
||||
// unhealthy, thus showing the expected failover behavior.
|
||||
let service = XdsAddressType::Service(XdsService {
|
||||
hostname: "example.com".to_string(),
|
||||
addresses: vec![XdsNetworkAddress {
|
||||
network: "".to_string(),
|
||||
address: vec![127, 0, 0, 3],
|
||||
}],
|
||||
ports: vec![Port {
|
||||
service_port: 80,
|
||||
target_port: 8080,
|
||||
}],
|
||||
// Prefer routing to workloads on the same network, but when nothing is healthy locally
|
||||
// allow failing over to remote networks.
|
||||
load_balancing: Some(xds::istio::workload::LoadBalancing {
|
||||
routing_preference: vec![
|
||||
xds::istio::workload::load_balancing::Scope::Network.into(),
|
||||
],
|
||||
mode: xds::istio::workload::load_balancing::Mode::Failover.into(),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
});
|
||||
let ew_gateway = XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/remote-ew-gtw".to_string(),
|
||||
addresses: vec![Bytes::copy_from_slice(&[10, 22, 1, 1])],
|
||||
network: "remote".to_string(),
|
||||
..Default::default()
|
||||
});
|
||||
let remote_workload = XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/remote-example.com-pod".to_string(),
|
||||
addresses: vec![],
|
||||
network: "remote".to_string(),
|
||||
network_gateway: Some(xds::istio::workload::GatewayAddress {
|
||||
destination: Some(xds::istio::workload::gateway_address::Destination::Address(
|
||||
XdsNetworkAddress {
|
||||
network: "remote".to_string(),
|
||||
address: vec![10, 22, 1, 1],
|
||||
},
|
||||
)),
|
||||
hbone_mtls_port: 15009,
|
||||
}),
|
||||
services: std::collections::HashMap::from([(
|
||||
"/example.com".to_string(),
|
||||
PortList {
|
||||
ports: vec![Port {
|
||||
service_port: 80,
|
||||
target_port: 8080,
|
||||
}],
|
||||
},
|
||||
)]),
|
||||
..Default::default()
|
||||
});
|
||||
let healthy_local_workload = XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/local-example.com-pod".to_string(),
|
||||
addresses: vec![Bytes::copy_from_slice(&[127, 0, 0, 2])],
|
||||
network: "".to_string(),
|
||||
tunnel_protocol: xds::istio::workload::TunnelProtocol::Hbone.into(),
|
||||
services: std::collections::HashMap::from([(
|
||||
"/example.com".to_string(),
|
||||
PortList {
|
||||
ports: vec![Port {
|
||||
service_port: 80,
|
||||
target_port: 8080,
|
||||
}],
|
||||
},
|
||||
)]),
|
||||
status: xds::istio::workload::WorkloadStatus::Healthy.into(),
|
||||
..Default::default()
|
||||
});
|
||||
let unhealthy_local_workload = XdsAddressType::Workload(XdsWorkload {
|
||||
uid: "Kubernetes//Pod/default/local-example.com-pod".to_string(),
|
||||
addresses: vec![Bytes::copy_from_slice(&[127, 0, 0, 2])],
|
||||
network: "".to_string(),
|
||||
tunnel_protocol: xds::istio::workload::TunnelProtocol::Hbone.into(),
|
||||
services: std::collections::HashMap::from([(
|
||||
"/example.com".to_string(),
|
||||
PortList {
|
||||
ports: vec![Port {
|
||||
service_port: 80,
|
||||
target_port: 8080,
|
||||
}],
|
||||
},
|
||||
)]),
|
||||
status: xds::istio::workload::WorkloadStatus::Unhealthy.into(),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
run_build_request_multi(
|
||||
"127.0.0.1",
|
||||
"127.0.0.3:80",
|
||||
vec![
|
||||
service.clone(),
|
||||
ew_gateway.clone(),
|
||||
remote_workload.clone(),
|
||||
healthy_local_workload.clone(),
|
||||
],
|
||||
Some(ExpectedRequest {
|
||||
protocol: OutboundProtocol::HBONE,
|
||||
hbone_destination: "127.0.0.2:8080",
|
||||
destination: "127.0.0.2:15008",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
run_build_request_multi(
|
||||
"127.0.0.1",
|
||||
"127.0.0.3:80",
|
||||
vec![
|
||||
service.clone(),
|
||||
ew_gateway.clone(),
|
||||
remote_workload.clone(),
|
||||
unhealthy_local_workload.clone(),
|
||||
],
|
||||
Some(ExpectedRequest {
|
||||
protocol: OutboundProtocol::DOUBLEHBONE,
|
||||
hbone_destination: "example.com:80",
|
||||
destination: "10.22.1.1:15009",
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
|
@ -594,9 +594,10 @@ mod test {
|
|||
}
|
||||
|
||||
/// This is really a test for TokioH2Stream, but its nicer here because we have access to
|
||||
/// streams
|
||||
/// streams.
|
||||
/// Most important, we make sure there are no panics.
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn small_reads() {
|
||||
async fn read_buffering() {
|
||||
let (mut pool, srv) = setup_test(3).await;
|
||||
|
||||
let key = key(&srv, 2);
|
||||
|
@ -612,13 +613,28 @@ mod test {
|
|||
let c = pool.send_request_pooled(&key.clone(), req()).await.unwrap();
|
||||
let mut c = TokioH2Stream::new(c);
|
||||
c.write_all(b"abcde").await.unwrap();
|
||||
let mut b = [0u8; 0];
|
||||
// Crucially, this should error rather than panic.
|
||||
if let Err(e) = c.read(&mut b).await {
|
||||
assert_eq!(e.kind(), io::ErrorKind::Other);
|
||||
} else {
|
||||
panic!("Should have errored");
|
||||
}
|
||||
let mut b = [0u8; 100];
|
||||
// Properly buffer reads and don't error
|
||||
assert_eq!(c.read(&mut b).await.unwrap(), 8);
|
||||
assert_eq!(&b[..8], b"poolsrv\n"); // this is added by itself
|
||||
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
|
||||
assert_eq!(&b[..1], b"a");
|
||||
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
|
||||
assert_eq!(&b[..1], b"b");
|
||||
assert_eq!(c.read(&mut b[..1]).await.unwrap(), 1);
|
||||
assert_eq!(&b[..1], b"c");
|
||||
assert_eq!(c.read(&mut b).await.unwrap(), 2); // there are only two bytes left
|
||||
assert_eq!(&b[..2], b"de");
|
||||
|
||||
// Once we drop the pool, we should still retained the buffered data,
|
||||
// but then we should error.
|
||||
c.write_all(b"abcde").await.unwrap();
|
||||
assert_eq!(c.read(&mut b[..3]).await.unwrap(), 3);
|
||||
assert_eq!(&b[..3], b"abc");
|
||||
drop(pool);
|
||||
assert_eq!(c.read(&mut b[..2]).await.unwrap(), 2);
|
||||
assert_eq!(&b[..2], b"de");
|
||||
assert!(c.read(&mut b).await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
|
|
|
@ -22,10 +22,9 @@ use crate::dns;
|
|||
use crate::drain::DrainWatcher;
|
||||
|
||||
use crate::proxy::connection_manager::ConnectionManager;
|
||||
use crate::proxy::{DefaultSocketFactory, Proxy, inbound::Inbound};
|
||||
use crate::proxy::{Error, LocalWorkloadInformation, Metrics};
|
||||
|
||||
use crate::proxy::Proxy;
|
||||
|
||||
// Proxy factory creates ztunnel proxies using a socket factory.
|
||||
// this allows us to create our proxies the same way in regular mode and in inpod mode.
|
||||
pub struct ProxyFactory {
|
||||
|
@ -113,6 +112,7 @@ impl ProxyFactory {
|
|||
drain.clone(),
|
||||
socket_factory.as_ref(),
|
||||
local_workload_information.as_fetcher(),
|
||||
self.config.prefered_service_namespace.clone(),
|
||||
)
|
||||
.await?;
|
||||
resolver = Some(server.resolver());
|
||||
|
@ -130,6 +130,7 @@ impl ProxyFactory {
|
|||
socket_factory.clone(),
|
||||
resolver,
|
||||
local_workload_information,
|
||||
false,
|
||||
);
|
||||
result.connection_manager = Some(cm);
|
||||
result.proxy = Some(Proxy::from_inputs(pi, drain).await?);
|
||||
|
@ -137,6 +138,52 @@ impl ProxyFactory {
|
|||
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// Creates an inbound listener specifically for ztunnel's own internal endpoints (metrics).
|
||||
/// This allows ztunnel to act as its own workload, enforcing policies on traffic directed to itself.
|
||||
/// This is distinct from the main inbound listener which handles traffic for other workloads proxied by ztunnel.
|
||||
pub async fn create_ztunnel_self_proxy_listener(
|
||||
&self,
|
||||
) -> Result<Option<crate::proxy::inbound::Inbound>, Error> {
|
||||
if self.config.proxy_mode != config::ProxyMode::Shared {
|
||||
return Ok(None);
|
||||
}
|
||||
|
||||
if let (Some(ztunnel_identity), Some(ztunnel_workload)) =
|
||||
(&self.config.ztunnel_identity, &self.config.ztunnel_workload)
|
||||
{
|
||||
tracing::info!(
|
||||
"creating ztunnel self-proxy listener with identity: {:?}",
|
||||
ztunnel_identity
|
||||
);
|
||||
|
||||
let local_workload_information = Arc::new(LocalWorkloadInformation::new(
|
||||
Arc::new(ztunnel_workload.clone()),
|
||||
self.state.clone(),
|
||||
self.cert_manager.clone(),
|
||||
));
|
||||
|
||||
let socket_factory = Arc::new(DefaultSocketFactory(self.config.socket_config));
|
||||
|
||||
let cm = ConnectionManager::default();
|
||||
|
||||
let pi = crate::proxy::ProxyInputs::new(
|
||||
self.config.clone(),
|
||||
cm.clone(),
|
||||
self.state.clone(),
|
||||
self.proxy_metrics.clone(),
|
||||
socket_factory,
|
||||
None,
|
||||
local_workload_information,
|
||||
true,
|
||||
);
|
||||
|
||||
let inbound = Inbound::new(pi, self.drain.clone()).await?;
|
||||
Ok(Some(inbound))
|
||||
} else {
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
|
|
51
src/state.rs
51
src/state.rs
|
@ -366,6 +366,19 @@ impl ProxyState {
|
|||
debug!("failed to fetch workload for {}", ep.workload_uid);
|
||||
return None;
|
||||
};
|
||||
|
||||
let in_network = wl.network == src.network;
|
||||
let has_network_gateway = wl.network_gateway.is_some();
|
||||
let has_address = !wl.workload_ips.is_empty() || !wl.hostname.is_empty();
|
||||
if !has_address {
|
||||
// Workload has no IP. We can only reach it via a network gateway
|
||||
// WDS is client-agnostic, so we will get a network gateway for a workload
|
||||
// even if it's in the same network; we should never use it.
|
||||
if in_network || !has_network_gateway {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
|
||||
match resolution_mode {
|
||||
ServiceResolutionMode::Standard => {
|
||||
if target_port.unwrap_or_default() == 0 && !ep.port.contains_key(&svc_port) {
|
||||
|
@ -1571,6 +1584,22 @@ mod tests {
|
|||
},
|
||||
..test_helpers::test_default_workload()
|
||||
};
|
||||
let wl_empty_ip = Workload {
|
||||
uid: "cluster1//v1/Pod/default/wl_empty_ip".into(),
|
||||
name: "wl_empty_ip".into(),
|
||||
namespace: "default".into(),
|
||||
trust_domain: "cluster.local".into(),
|
||||
service_account: "default".into(),
|
||||
workload_ips: vec![], // none!
|
||||
network: "network".into(),
|
||||
locality: Locality {
|
||||
region: "reg".into(),
|
||||
zone: "zone".into(),
|
||||
subzone: "".into(),
|
||||
},
|
||||
..test_helpers::test_default_workload()
|
||||
};
|
||||
|
||||
let _ep_almost = Workload {
|
||||
uid: "cluster1//v1/Pod/default/ep_almost".into(),
|
||||
name: "wl_almost".into(),
|
||||
|
@ -1617,6 +1646,11 @@ mod tests {
|
|||
port: HashMap::from([(80u16, 80u16)]),
|
||||
status: HealthStatus::Healthy,
|
||||
},
|
||||
Endpoint {
|
||||
workload_uid: "cluster1//v1/Pod/default/wl_empty_ip".into(),
|
||||
port: HashMap::from([(80u16, 80u16)]),
|
||||
status: HealthStatus::Healthy,
|
||||
},
|
||||
]);
|
||||
let strict_svc = Service {
|
||||
endpoints: endpoints.clone(),
|
||||
|
@ -1649,6 +1683,7 @@ mod tests {
|
|||
state.workloads.insert(Arc::new(wl_no_locality.clone()));
|
||||
state.workloads.insert(Arc::new(wl_match.clone()));
|
||||
state.workloads.insert(Arc::new(wl_almost.clone()));
|
||||
state.workloads.insert(Arc::new(wl_empty_ip.clone()));
|
||||
state.services.insert(strict_svc.clone());
|
||||
state.services.insert(failover_svc.clone());
|
||||
|
||||
|
@ -1663,6 +1698,15 @@ mod tests {
|
|||
assert!(want.contains(&got.unwrap()), "{}", desc);
|
||||
}
|
||||
};
|
||||
let assert_not_endpoint =
|
||||
|src: &Workload, svc: &Service, uid: &str, tries: usize, desc: &str| {
|
||||
for _ in 0..tries {
|
||||
let got = state
|
||||
.load_balance(src, svc, 80, ServiceResolutionMode::Standard)
|
||||
.map(|(ep, _)| ep.workload_uid.as_str());
|
||||
assert!(got != Some(uid), "{}", desc);
|
||||
}
|
||||
};
|
||||
|
||||
assert_endpoint(
|
||||
&wl_no_locality,
|
||||
|
@ -1708,5 +1752,12 @@ mod tests {
|
|||
vec!["cluster1//v1/Pod/default/wl_match"],
|
||||
"failover full match selects closest match",
|
||||
);
|
||||
assert_not_endpoint(
|
||||
&wl_no_locality,
|
||||
&failover_svc,
|
||||
"cluster1//v1/Pod/default/wl_empty_ip",
|
||||
10,
|
||||
"failover no match can select any endpoint",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -326,9 +326,7 @@ impl io::Write for WriteAdaptor<'_> {
|
|||
let s =
|
||||
std::str::from_utf8(buf).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
|
||||
|
||||
self.fmt_write
|
||||
.write_str(s)
|
||||
.map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
|
||||
self.fmt_write.write_str(s).map_err(io::Error::other)?;
|
||||
|
||||
Ok(s.len())
|
||||
}
|
||||
|
|
|
@ -52,6 +52,7 @@ pub struct TestApp {
|
|||
|
||||
pub namespace: Option<super::netns::Namespace>,
|
||||
pub shutdown: ShutdownTrigger,
|
||||
pub ztunnel_identity: Option<identity::Identity>,
|
||||
}
|
||||
|
||||
impl From<(&Bound, Arc<SecretManager>)> for TestApp {
|
||||
|
@ -66,6 +67,7 @@ impl From<(&Bound, Arc<SecretManager>)> for TestApp {
|
|||
cert_manager,
|
||||
namespace: None,
|
||||
shutdown: app.shutdown.trigger(),
|
||||
ztunnel_identity: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -298,6 +298,7 @@ pub async fn run_dns(responses: HashMap<Name, Vec<IpAddr>>) -> anyhow::Result<Te
|
|||
}),
|
||||
state.clone(),
|
||||
),
|
||||
Some("prefered-namespace".to_string()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
|
|
|
@ -124,38 +124,104 @@ impl WorkloadManager {
|
|||
wli: Option<state::WorkloadInfo>,
|
||||
) -> anyhow::Result<TestApp> {
|
||||
let mut inpod_uds: PathBuf = "/dev/null".into();
|
||||
let ztunnel_server = if self.mode == Shared {
|
||||
inpod_uds = self.tmp_dir.join(node);
|
||||
Some(start_ztunnel_server(inpod_uds.clone()).await)
|
||||
let current_mode = self.mode;
|
||||
let proxy_mode = match current_mode {
|
||||
Shared => ProxyMode::Shared,
|
||||
Dedicated => ProxyMode::Dedicated,
|
||||
};
|
||||
let ztunnel_name = format!("ztunnel-{node}");
|
||||
|
||||
// Define ztunnel's own identity and workload info if it's a Shared proxy.
|
||||
// These are used for registering ztunnel as a workload and for cfg.ztunnel_identity/workload.
|
||||
let ztunnel_shared_identity: Option<identity::Identity> = if proxy_mode == ProxyMode::Shared
|
||||
{
|
||||
Some(identity::Identity::Spiffe {
|
||||
trust_domain: "cluster.local".into(),
|
||||
namespace: "default".into(),
|
||||
service_account: ztunnel_name.clone().into(),
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let ns = TestWorkloadBuilder::new(&format!("ztunnel-{node}"), self)
|
||||
.on_node(node)
|
||||
.uncaptured()
|
||||
.register()
|
||||
.await?;
|
||||
|
||||
let ztunnel_shared_workload_info: Option<state::WorkloadInfo> =
|
||||
if proxy_mode == ProxyMode::Shared {
|
||||
Some(state::WorkloadInfo::new(
|
||||
ztunnel_name.clone(),
|
||||
"default".to_string(),
|
||||
ztunnel_name.clone(),
|
||||
))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let ztunnel_server = match current_mode {
|
||||
Shared => {
|
||||
inpod_uds = self.tmp_dir.join(node);
|
||||
Some(start_ztunnel_server(inpod_uds.clone()).await)
|
||||
}
|
||||
Dedicated => None,
|
||||
};
|
||||
|
||||
let ns = match current_mode {
|
||||
Shared => {
|
||||
// Shared mode: Ztunnel has its own identity, registered as HBONE
|
||||
TestWorkloadBuilder::new(&ztunnel_name, self)
|
||||
.on_node(node)
|
||||
.identity(
|
||||
ztunnel_shared_identity
|
||||
.clone()
|
||||
.expect("Shared mode must have an identity for ztunnel registration"),
|
||||
)
|
||||
.hbone() // Shared ztunnel uses HBONE protocol
|
||||
.register()
|
||||
.await?
|
||||
}
|
||||
Dedicated => {
|
||||
TestWorkloadBuilder::new(&ztunnel_name, self)
|
||||
.on_node(node)
|
||||
.uncaptured() // Dedicated ztunnel is treated as uncaptured TCP
|
||||
.register()
|
||||
.await?
|
||||
}
|
||||
};
|
||||
let _ztunnel_local_workload = self
|
||||
.workloads
|
||||
.last()
|
||||
.cloned()
|
||||
.expect("ztunnel workload should be registered");
|
||||
|
||||
let ip = ns.ip();
|
||||
let initial_config = LocalConfig {
|
||||
workloads: self.workloads.clone(),
|
||||
policies: self.policies.clone(),
|
||||
services: self.services.values().cloned().collect_vec(),
|
||||
};
|
||||
let proxy_mode = if ztunnel_server.is_some() {
|
||||
ProxyMode::Shared
|
||||
} else {
|
||||
ProxyMode::Dedicated
|
||||
};
|
||||
let (mut tx_cfg, rx_cfg) = mpsc_ack(1);
|
||||
tx_cfg.send(initial_config).await?;
|
||||
let local_xds_config = Some(ConfigSource::Dynamic(Arc::new(Mutex::new(rx_cfg))));
|
||||
|
||||
// Config for ztunnel's own identity and workload, primarily for when it acts as a server (metrics endpoint).
|
||||
let cfg_ztunnel_identity = ztunnel_shared_identity.clone();
|
||||
let cfg_ztunnel_workload_info = ztunnel_shared_workload_info.clone();
|
||||
|
||||
// Config for the workload this ztunnel instance is proxying for :
|
||||
// If Shared, ztunnel is effectively proxying for itself
|
||||
// If Dedicated, it's for the application workload `wli`
|
||||
let cfg_proxy_workload_information = match proxy_mode {
|
||||
// Ztunnel's own info for shared mode proxy
|
||||
ProxyMode::Shared => ztunnel_shared_workload_info.clone(),
|
||||
// Application's workload info for dedicated mode
|
||||
ProxyMode::Dedicated => wli,
|
||||
};
|
||||
|
||||
let cfg = config::Config {
|
||||
xds_address: None,
|
||||
dns_proxy: true,
|
||||
fake_ca: true,
|
||||
local_xds_config,
|
||||
local_node: Some(node.to_string()),
|
||||
proxy_workload_information: wli,
|
||||
proxy_workload_information: cfg_proxy_workload_information,
|
||||
inpod_uds,
|
||||
proxy_mode,
|
||||
// We use packet mark even in dedicated to distinguish proxy from application
|
||||
|
@ -166,12 +232,16 @@ impl WorkloadManager {
|
|||
Some(true)
|
||||
},
|
||||
localhost_app_tunnel: true,
|
||||
ztunnel_identity: cfg_ztunnel_identity,
|
||||
ztunnel_workload: cfg_ztunnel_workload_info,
|
||||
..config::parse_config().unwrap()
|
||||
};
|
||||
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel(0);
|
||||
// Setup the ztunnel...
|
||||
let cloned_ns = ns.clone();
|
||||
let cloned_ns2 = ns.clone();
|
||||
let ztunnel_identity = ztunnel_shared_identity.clone();
|
||||
// run_ready will spawn a thread and block on it. Run with spawn_blocking so it doesn't block the runtime.
|
||||
tokio::task::spawn_blocking(move || {
|
||||
ns.run_ready(move |ready| async move {
|
||||
|
@ -210,9 +280,9 @@ impl WorkloadManager {
|
|||
ip,
|
||||
)),
|
||||
cert_manager,
|
||||
|
||||
namespace: Some(cloned_ns),
|
||||
shutdown,
|
||||
ztunnel_identity: ztunnel_identity.clone(),
|
||||
};
|
||||
ta.ready().await;
|
||||
info!("ready");
|
||||
|
@ -512,16 +582,18 @@ impl<'a> TestWorkloadBuilder<'a> {
|
|||
pub async fn register(mut self) -> anyhow::Result<Namespace> {
|
||||
let zt = self.manager.ztunnels.get(self.w.workload.node.as_str());
|
||||
let node = self.w.workload.node.clone();
|
||||
let network_namespace = if self.manager.mode == Dedicated && zt.is_some() {
|
||||
// This is a bit of hack. For dedicated mode, we run the app and ztunnel in the same namespace
|
||||
// We probably should express this more natively in the framework, but for now we just detect it
|
||||
// and re-use the namespace.
|
||||
tracing::info!("node already has ztunnel and dedicate mode, sharing");
|
||||
zt.as_ref().unwrap().namespace.clone()
|
||||
} else {
|
||||
self.manager
|
||||
let network_namespace = match (self.manager.mode, zt.is_some()) {
|
||||
(Dedicated, true) => {
|
||||
// This is a bit of hack. For dedicated mode, we run the app and ztunnel in the same namespace
|
||||
// We probably should express this more natively in the framework, but for now we just detect it
|
||||
// and re-use the namespace.
|
||||
tracing::info!("node already has ztunnel and dedicate mode, sharing");
|
||||
zt.as_ref().unwrap().namespace.clone()
|
||||
}
|
||||
_ => self
|
||||
.manager
|
||||
.namespaces
|
||||
.child(&self.w.workload.node, &self.w.workload.name)?
|
||||
.child(&self.w.workload.node, &self.w.workload.name)?,
|
||||
};
|
||||
if self.w.workload.network_gateway.is_some() {
|
||||
// This is a little inefficient, because we create the
|
||||
|
@ -569,7 +641,7 @@ impl<'a> TestWorkloadBuilder<'a> {
|
|||
let fd = network_namespace.netns().file().as_raw_fd();
|
||||
let msg = inpod::Message::Start(inpod::StartZtunnelMessage {
|
||||
uid: uid.to_string(),
|
||||
workload_info: Some(wli),
|
||||
workload_info: Some(wli.clone()),
|
||||
fd,
|
||||
});
|
||||
zt_info
|
||||
|
|
|
@ -124,28 +124,44 @@ impl CsrOptions {
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::tls;
|
||||
use itertools::Itertools;
|
||||
|
||||
#[test]
|
||||
fn test_csr() {
|
||||
use x509_parser::prelude::FromDer;
|
||||
use x509_parser::prelude::*;
|
||||
let csr = tls::csr::CsrOptions {
|
||||
san: "spiffe://td/ns/ns1/sa/sa1".to_string(),
|
||||
}
|
||||
.generate()
|
||||
.unwrap();
|
||||
|
||||
let (_, der) = x509_parser::pem::parse_x509_pem(csr.csr.as_bytes()).unwrap();
|
||||
|
||||
let (_, cert) =
|
||||
x509_parser::certification_request::X509CertificationRequest::from_der(&der.contents)
|
||||
.unwrap();
|
||||
cert.verify_signature().unwrap();
|
||||
let subject = cert.certification_request_info.subject.iter().collect_vec();
|
||||
assert_eq!(subject.len(), 0);
|
||||
let attr = cert
|
||||
.certification_request_info
|
||||
.iter_attributes()
|
||||
.next()
|
||||
.unwrap();
|
||||
// SAN is encoded in some format I don't understand how to parse; this could be improved.
|
||||
// but make sure it's there in a hacky manner
|
||||
assert!(attr.value.ends_with(b"spiffe://td/ns/ns1/sa/sa1"));
|
||||
|
||||
let ParsedCriAttribute::ExtensionRequest(parsed) = attr.parsed_attribute() else {
|
||||
panic!("not a ExtensionRequest")
|
||||
};
|
||||
let ext = parsed.clone().extensions;
|
||||
assert_eq!(ext.len(), 1);
|
||||
let ext = ext.into_iter().next().unwrap();
|
||||
assert!(ext.critical);
|
||||
let ParsedExtension::SubjectAlternativeName(san) = ext.parsed_extension() else {
|
||||
panic!("not a SubjectAlternativeName")
|
||||
};
|
||||
assert_eq!(
|
||||
&format!("{san:?}"),
|
||||
"SubjectAlternativeName { general_names: [URI(\"spiffe://td/ns/ns1/sa/sa1\")] }"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -14,6 +14,8 @@
|
|||
|
||||
use super::Error;
|
||||
|
||||
#[allow(unused_imports)]
|
||||
use crate::PQC_ENABLED;
|
||||
use crate::identity::{self, Identity};
|
||||
|
||||
use std::fmt::Debug;
|
||||
|
@ -40,6 +42,15 @@ pub trait ServerCertProvider: Send + Sync + Clone {
|
|||
|
||||
pub(super) static TLS_VERSIONS: &[&rustls::SupportedProtocolVersion] = &[&rustls::version::TLS13];
|
||||
|
||||
#[cfg(feature = "tls-aws-lc")]
|
||||
pub static CRYPTO_PROVIDER: &str = "tls-aws-lc";
|
||||
#[cfg(feature = "tls-ring")]
|
||||
pub static CRYPTO_PROVIDER: &str = "tls-ring";
|
||||
#[cfg(feature = "tls-boring")]
|
||||
pub static CRYPTO_PROVIDER: &str = "tls-boring";
|
||||
#[cfg(feature = "tls-openssl")]
|
||||
pub static CRYPTO_PROVIDER: &str = "tls-openssl";
|
||||
|
||||
// Ztunnel use `rustls` with pluggable crypto modules.
|
||||
// All crypto MUST be done via the below providers.
|
||||
//
|
||||
|
@ -68,14 +79,20 @@ pub(super) fn provider() -> Arc<CryptoProvider> {
|
|||
|
||||
#[cfg(feature = "tls-aws-lc")]
|
||||
pub(super) fn provider() -> Arc<CryptoProvider> {
|
||||
Arc::new(CryptoProvider {
|
||||
let mut provider = CryptoProvider {
|
||||
// Limit to only the subset of ciphers that are FIPS compatible
|
||||
cipher_suites: vec![
|
||||
rustls::crypto::aws_lc_rs::cipher_suite::TLS13_AES_256_GCM_SHA384,
|
||||
rustls::crypto::aws_lc_rs::cipher_suite::TLS13_AES_128_GCM_SHA256,
|
||||
],
|
||||
..rustls::crypto::aws_lc_rs::default_provider()
|
||||
})
|
||||
};
|
||||
|
||||
if *PQC_ENABLED {
|
||||
provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519MLKEM768]
|
||||
}
|
||||
|
||||
Arc::new(provider)
|
||||
}
|
||||
|
||||
#[cfg(feature = "tls-openssl")]
|
||||
|
|
|
@ -17,10 +17,11 @@ use std::fmt;
|
|||
use std::fmt::{Display, Formatter};
|
||||
use std::string::String;
|
||||
|
||||
use crate::tls::CRYPTO_PROVIDER;
|
||||
|
||||
const BUILD_VERSION: &str = env!("ZTUNNEL_BUILD_buildVersion");
|
||||
const BUILD_GIT_REVISION: &str = env!("ZTUNNEL_BUILD_buildGitRevision");
|
||||
const BUILD_STATUS: &str = env!("ZTUNNEL_BUILD_buildStatus");
|
||||
const BUILD_TAG: &str = env!("ZTUNNEL_BUILD_buildTag");
|
||||
const BUILD_RUST_VERSION: &str = env!("ZTUNNEL_BUILD_RUSTC_VERSION");
|
||||
const BUILD_RUST_PROFILE: &str = env!("ZTUNNEL_BUILD_PROFILE_NAME");
|
||||
|
||||
|
@ -32,8 +33,8 @@ pub struct BuildInfo {
|
|||
rust_version: String,
|
||||
build_profile: String,
|
||||
build_status: String,
|
||||
git_tag: String,
|
||||
pub istio_version: String,
|
||||
crypto_provider: String,
|
||||
}
|
||||
|
||||
impl BuildInfo {
|
||||
|
@ -44,8 +45,9 @@ impl BuildInfo {
|
|||
rust_version: BUILD_RUST_VERSION.to_string(),
|
||||
build_profile: BUILD_RUST_PROFILE.to_string(),
|
||||
build_status: BUILD_STATUS.to_string(),
|
||||
git_tag: BUILD_TAG.to_string(),
|
||||
istio_version: env::var("ISTIO_VERSION").unwrap_or_else(|_| "unknown".to_string()),
|
||||
istio_version: env::var("ISTIO_META_ISTIO_VERSION")
|
||||
.unwrap_or_else(|_| "unknown".to_string()),
|
||||
crypto_provider: CRYPTO_PROVIDER.to_string(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -54,14 +56,14 @@ impl Display for BuildInfo {
|
|||
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
|
||||
write!(
|
||||
f,
|
||||
"version.BuildInfo{{Version:\"{}\", GitRevision:\"{}\", RustVersion:\"{}\", BuildProfile:\"{}\", BuildStatus:\"{}\", GitTag:\"{}\", IstioVersion:\"{}\"}}",
|
||||
"version.BuildInfo{{Version:\"{}\", GitRevision:\"{}\", RustVersion:\"{}\", BuildProfile:\"{}\", BuildStatus:\"{}\", IstioVersion:\"{}\", CryptoProvider:\"{}\"}}",
|
||||
self.version,
|
||||
self.git_revision,
|
||||
self.rust_version,
|
||||
self.build_profile,
|
||||
self.build_status,
|
||||
self.git_tag,
|
||||
self.istio_version
|
||||
self.istio_version,
|
||||
self.crypto_provider,
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -18,14 +18,16 @@ mod namespaced {
|
|||
use futures::future::poll_fn;
|
||||
use http_body_util::Empty;
|
||||
use std::collections::HashMap;
|
||||
use ztunnel::state::workload::ApplicationTunnel;
|
||||
use ztunnel::state::workload::application_tunnel::Protocol;
|
||||
use ztunnel::state::workload::gatewayaddress::Destination;
|
||||
use ztunnel::state::workload::{GatewayAddress, NamespacedHostname};
|
||||
use ztunnel::test_helpers::linux::TestMode;
|
||||
|
||||
use std::net::{IpAddr, SocketAddr};
|
||||
|
||||
use anyhow::Context;
|
||||
use std::str::FromStr;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread::JoinHandle;
|
||||
use std::time::Duration;
|
||||
use ztunnel::rbac::{Authorization, RbacMatch, StringMatch};
|
||||
|
@ -39,18 +41,17 @@ mod namespaced {
|
|||
use tokio::time::timeout;
|
||||
use tracing::{error, info};
|
||||
|
||||
use ztunnel::state::workload::{ApplicationTunnel, NetworkAddress};
|
||||
use ztunnel::test_helpers::app::ParsedMetrics;
|
||||
use ztunnel::test_helpers::app::TestApp;
|
||||
use ztunnel::state::workload::NetworkAddress;
|
||||
use ztunnel::test_helpers::app::{ParsedMetrics, TestApp};
|
||||
use ztunnel::test_helpers::linux::TestMode::{Dedicated, Shared};
|
||||
use ztunnel::test_helpers::linux::WorkloadManager;
|
||||
use ztunnel::test_helpers::netns::{Namespace, Resolver};
|
||||
use ztunnel::test_helpers::*;
|
||||
|
||||
use ztunnel::{identity, strng, telemetry};
|
||||
|
||||
use crate::namespaced::WorkloadMode::Captured;
|
||||
use ztunnel::setup_netns_test;
|
||||
use ztunnel::test_helpers::linux::TestMode::{Dedicated, Shared};
|
||||
use ztunnel::test_helpers::linux::WorkloadManager;
|
||||
use ztunnel::test_helpers::netns::{Namespace, Resolver};
|
||||
use ztunnel::test_helpers::*;
|
||||
|
||||
const WAYPOINT_MESSAGE: &[u8] = b"waypoint\n";
|
||||
|
||||
|
@ -926,27 +927,12 @@ mod namespaced {
|
|||
// Now shutdown the server. In real world, the server app would shutdown, then ztunnel would remove itself.
|
||||
// In this test, we will leave the server app running, but shutdown ztunnel.
|
||||
manager.delete_workload("server").await.unwrap();
|
||||
// Request should fail now
|
||||
let tx = Arc::new(Mutex::new(tx));
|
||||
#[allow(clippy::await_holding_lock)]
|
||||
assert_eventually(
|
||||
Duration::from_secs(2),
|
||||
|| async { tx.lock().unwrap().send_and_wait(()).await.is_err() },
|
||||
true,
|
||||
)
|
||||
.await;
|
||||
// Close the connection
|
||||
drop(tx);
|
||||
|
||||
// Should fail as the last request fails
|
||||
assert!(cjh.join().unwrap().is_err());
|
||||
|
||||
// Now try to connect and make sure it fails
|
||||
// In shared mode, verify that new connections succeed but data transfer fails
|
||||
client
|
||||
.run_and_wait(move || async move {
|
||||
let mut stream = TcpStream::connect(srv).await.unwrap();
|
||||
// We should be able to connect (since client is running), but not send a request
|
||||
|
||||
const BODY: &[u8] = b"hello world";
|
||||
stream.write_all(BODY).await.unwrap();
|
||||
let mut buf = [0; BODY.len() * 2];
|
||||
|
@ -955,6 +941,16 @@ mod namespaced {
|
|||
Ok(())
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
// The long running connection should also fail on next attempt
|
||||
let tx_send_result = tx.send_and_wait(()).await;
|
||||
assert!(
|
||||
tx_send_result.is_err(),
|
||||
"long running connection should fail after workload deletion"
|
||||
);
|
||||
|
||||
drop(tx);
|
||||
assert!(cjh.join().unwrap().is_err());
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
@ -1220,7 +1216,7 @@ mod namespaced {
|
|||
vec![
|
||||
(zt, 15001, Request), // Outbound: should be blocked due to recursive call
|
||||
(zt, 15006, Request), // Inbound: should be blocked due to recursive call
|
||||
(zt, 15008, Request), // HBONE: expected TLS, reject
|
||||
(zt, 15008, Request), // HBONE: Connection succeeds (ztunnel listens) but request fails due to TLS
|
||||
// Localhost still get connection established, as ztunnel accepts anything. But they are dropped immediately.
|
||||
(zt, 15080, Request), // socks5: localhost
|
||||
(zt, 15000, Request), // admin: localhost
|
||||
|
@ -1252,7 +1248,7 @@ mod namespaced {
|
|||
// Ztunnel doesn't listen on these ports...
|
||||
(zt, 15001, Connection), // Outbound: should be blocked due to recursive call
|
||||
(zt, 15006, Connection), // Inbound: should be blocked due to recursive call
|
||||
(zt, 15008, Connection), // HBONE: expected TLS, reject
|
||||
(zt, 15008, Request), // HBONE: Connection succeeds (ztunnel listens) but request fails due to TLS
|
||||
// Localhost is not accessible
|
||||
(zt, 15080, Connection), // socks5: localhost
|
||||
(zt, 15000, Connection), // admin: localhost
|
||||
|
@ -1329,66 +1325,191 @@ mod namespaced {
|
|||
let id1s = id1.to_string();
|
||||
|
||||
let ta = manager.deploy_ztunnel(DEFAULT_NODE).await?;
|
||||
let ztunnel_identity_obj = ta.ztunnel_identity.as_ref().unwrap().clone();
|
||||
ta.cert_manager
|
||||
.fetch_certificate(&ztunnel_identity_obj)
|
||||
.await?;
|
||||
let ztunnel_identity_str = ztunnel_identity_obj.to_string();
|
||||
|
||||
let check = |want: Vec<String>, help: &str| {
|
||||
let cm = ta.cert_manager.clone();
|
||||
let help = help.to_string();
|
||||
let mut sorted_want = want.clone();
|
||||
sorted_want.sort();
|
||||
async move {
|
||||
// Cert manager is async, so we need to wait
|
||||
let res = check_eventually(
|
||||
Duration::from_secs(2),
|
||||
|| cm.collect_certs(|a, _b| a.to_string()),
|
||||
want,
|
||||
|| async {
|
||||
let mut certs = cm.collect_certs(|a, _b| a.to_string()).await;
|
||||
certs.sort();
|
||||
certs
|
||||
},
|
||||
sorted_want,
|
||||
)
|
||||
.await;
|
||||
assert!(res.is_ok(), "{}: got {:?}", help, res.err().unwrap());
|
||||
}
|
||||
};
|
||||
check(vec![], "initially empty").await;
|
||||
check(
|
||||
vec![ztunnel_identity_str.clone()],
|
||||
"initially only ztunnel cert",
|
||||
)
|
||||
.await;
|
||||
|
||||
manager
|
||||
.workload_builder("id1-a-remote-node", REMOTE_NODE)
|
||||
.identity(id1.clone())
|
||||
.register()
|
||||
.await?;
|
||||
check(vec![], "we should not prefetch remote nodes").await;
|
||||
check(
|
||||
vec![ztunnel_identity_str.clone()],
|
||||
"we should not prefetch remote nodes",
|
||||
)
|
||||
.await;
|
||||
|
||||
manager
|
||||
.workload_builder("id1-a-same-node", DEFAULT_NODE)
|
||||
.identity(id1.clone())
|
||||
.register()
|
||||
.await?;
|
||||
check(vec![id1s.clone()], "we should prefetch our nodes").await;
|
||||
check(
|
||||
vec![ztunnel_identity_str.clone(), id1s.clone()],
|
||||
"we should prefetch our nodes",
|
||||
)
|
||||
.await;
|
||||
|
||||
manager
|
||||
.workload_builder("id1-b-same-node", DEFAULT_NODE)
|
||||
.identity(id1.clone())
|
||||
.register()
|
||||
.await?;
|
||||
check(
|
||||
vec![id1s.clone()],
|
||||
vec![ztunnel_identity_str.clone(), id1s.clone()],
|
||||
"multiple of same identity shouldn't do anything",
|
||||
)
|
||||
.await;
|
||||
manager.delete_workload("id1-a-remote-node").await?;
|
||||
// Deleting remote node should not affect local certs if local workloads still exist
|
||||
check(
|
||||
vec![id1s.clone()],
|
||||
vec![ztunnel_identity_str.clone(), id1s.clone()],
|
||||
"removing remote node shouldn't impact anything",
|
||||
)
|
||||
.await;
|
||||
manager.delete_workload("id1-b-same-node").await?;
|
||||
// Deleting one local node shouldn't impact certs if another local workload still exists
|
||||
check(
|
||||
vec![id1s.clone()],
|
||||
vec![ztunnel_identity_str.clone(), id1s.clone()],
|
||||
"removing local node shouldn't impact anything if I still have some running",
|
||||
)
|
||||
.await;
|
||||
manager.delete_workload("id1-a-same-node").await?;
|
||||
// TODO: this should be vec![], but our testing setup doesn't exercise the real codepath
|
||||
// After deleting all workloads using sa1, give cert manager time to clean up
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// In shared mode, certificates may be kept alive by the inbound listener
|
||||
// for handling inbound connections, even after workload deletion
|
||||
let expected_certs = match manager.mode() {
|
||||
TestMode::Shared => vec![ztunnel_identity_str.clone(), id1s.clone()],
|
||||
TestMode::Dedicated => vec![ztunnel_identity_str.clone()],
|
||||
};
|
||||
check(
|
||||
vec![id1s.clone()],
|
||||
"removing final workload should clear things out",
|
||||
expected_certs,
|
||||
"removing final workload should clear certs except those needed by inbound listener",
|
||||
)
|
||||
.await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hbone_metrics_access() -> Result<(), anyhow::Error> {
|
||||
let mut manager = setup_netns_test!(Shared);
|
||||
|
||||
// Deploy ztunnel for the node
|
||||
let zt = manager.deploy_ztunnel(DEFAULT_NODE).await?;
|
||||
let ztunnel_node_ip = manager.resolve("ztunnel-node")?;
|
||||
// Use the actual metrics address ztunnel is listening on (e.g., [::]:15020)
|
||||
// but combine it with the node IP for the client to target.
|
||||
let target_metrics_addr = SocketAddr::new(ztunnel_node_ip, zt.metrics_address.port());
|
||||
let target_metrics_url = format!("http://{}/metrics", target_metrics_addr);
|
||||
|
||||
// Deploy a client workload (simulating Prometheus)
|
||||
let client = manager
|
||||
.workload_builder("client", DEFAULT_NODE)
|
||||
.register()
|
||||
.await?;
|
||||
|
||||
let zt_identity_str = zt.ztunnel_identity.as_ref().unwrap().to_string();
|
||||
|
||||
// Client makes a standard HTTP GET request to ztunnel's metrics endpoint
|
||||
// Ztunnel's outbound capture should intercept this, initiate HBONE to its own inbound,
|
||||
// which then proxies to the internal metrics server.
|
||||
client
|
||||
.run(move || async move {
|
||||
info!(target=%target_metrics_url, "Client attempting standard HTTP GET to metrics endpoint");
|
||||
|
||||
let client = hyper_util::client::legacy::Client::builder(
|
||||
ztunnel::hyper_util::TokioExecutor,
|
||||
)
|
||||
.build_http();
|
||||
|
||||
let req = hyper::Request::builder()
|
||||
.method(Method::GET)
|
||||
.uri(&target_metrics_url)
|
||||
.body(Empty::<Bytes>::new())?;
|
||||
|
||||
let response = client.request(req).await?;
|
||||
|
||||
info!("Received response status: {:?}", response.status());
|
||||
assert_eq!(response.status(), StatusCode::OK, "GET request failed");
|
||||
|
||||
let body_bytes = http_body_util::BodyExt::collect(response.into_body())
|
||||
.await?
|
||||
.to_bytes();
|
||||
let response_str = String::from_utf8_lossy(&body_bytes);
|
||||
|
||||
assert!(
|
||||
response_str.contains("# TYPE"),
|
||||
"Expected Prometheus metrics (# TYPE) in response, got:\n{}",
|
||||
response_str
|
||||
);
|
||||
info!("Successfully verified metrics response body");
|
||||
|
||||
Ok(())
|
||||
})?
|
||||
.join()
|
||||
.unwrap()?;
|
||||
|
||||
// Verify metrics from the DESTINATION perspective (ztunnel handling its own inbound)
|
||||
let metrics = [
|
||||
(CONNECTIONS_OPENED, 1), // One connection opened (client -> zt inbound via HBONE)
|
||||
(CONNECTIONS_CLOSED, 1), // One connection closed
|
||||
];
|
||||
verify_metrics(&zt, &metrics, &destination_labels()).await;
|
||||
|
||||
// Verify INBOUND telemetry log for the metrics connection
|
||||
let dst_addr_log = format!("{}:15008", ztunnel_node_ip);
|
||||
let dst_hbone_addr_log = format!("{}", target_metrics_addr);
|
||||
|
||||
// We don't know exact byte counts, so omit them from the check for now
|
||||
let want = HashMap::from([
|
||||
("scope", "access"),
|
||||
("src.workload", "client"),
|
||||
("dst.workload", "ztunnel-node"), // ztunnel's workload name
|
||||
("dst.addr", dst_addr_log.as_str()), // Connected to HBONE port
|
||||
("dst.hbone_addr", dst_hbone_addr_log.as_str()), // Original target
|
||||
("direction", "inbound"),
|
||||
("message", "connection complete"), // Assuming success
|
||||
(
|
||||
"src.identity",
|
||||
"spiffe://cluster.local/ns/default/sa/client",
|
||||
), // Client identity
|
||||
("dst.identity", zt_identity_str.as_str()), // Ztunnel identity
|
||||
]);
|
||||
telemetry::testing::assert_contains(want);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
const TEST_VIP: &str = "10.10.0.1";
|
||||
const TEST_VIP2: &str = "10.10.0.2";
|
||||
const TEST_VIP3: &str = "10.10.0.3";
|
||||
|
@ -1759,7 +1880,6 @@ mod namespaced {
|
|||
}
|
||||
use Failure::*;
|
||||
use ztunnel::state::WorkloadInfo;
|
||||
use ztunnel::state::workload::application_tunnel::Protocol;
|
||||
|
||||
async fn malicious_calls_test(
|
||||
client: Namespace,
|
||||
|
@ -1786,14 +1906,14 @@ mod namespaced {
|
|||
let stream = timeout(Duration::from_secs(1), TcpStream::connect(tgt)).await?;
|
||||
error!("stream {stream:?}");
|
||||
if failure == Connection {
|
||||
assert!(stream.is_err());
|
||||
assert!(stream.is_err(), "expected connection to fail for {tgt}");
|
||||
continue;
|
||||
}
|
||||
let mut stream = stream.unwrap();
|
||||
|
||||
let res = timeout(Duration::from_secs(1), send_traffic(&mut stream)).await?;
|
||||
if failure == Request {
|
||||
assert!(res.is_err());
|
||||
assert!(res.is_err(), "expected request to fail for {tgt}");
|
||||
continue;
|
||||
}
|
||||
res.unwrap();
|
||||
|
|
Loading…
Reference in New Issue