Automator: update common-files@master in istio/ztunnel@master (#1595)

* Buffer h2 streams

* Tests

* naming

* Review

simplify code

.devcontainer/devcontainer.json

@@ -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",

Cargo.lock

@@ -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",
 ]

Cargo.toml

@@ -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"

common/.commonfiles.sha

@@ -1 +1 @@
-a1d5c4198ab79a14c09c034f2d95245efa3e2bcb
+d46067e1a8ba3db4abe2635af5807f00ba1981e6

common/scripts/kind_provisioner.sh

@@ -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}"

common/scripts/setup_env.sh

@@ -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

fuzz/Cargo.lock

@@ -3843,6 +3843,7 @@ dependencies = [
  "log",
  "netns-rs",
  "nix 0.29.0",
+ "num_cpus",
  "once_cell",
  "pin-project-lite",
  "pingora-pool",

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()

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);
+        }
+    }
 }

src/dns/server.rs

@@ -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),
         )
     }
 

src/identity/auth.rs

@@ -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)
 }

src/identity/manager.rs

@@ -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(),
         }

src/lib.rs

@@ -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");

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
 }

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"
     )]

src/proxy/h2.rs

@@ -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)
     }
 }

src/proxy/inbound.rs

@@ -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()
         });
 

src/proxy/outbound.rs

@@ -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]

src/proxy/pool.rs

@@ -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)]

src/proxyfactory.rs

@@ -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)]

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",
+        );
     }
 }

src/telemetry.rs

@@ -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())
     }

src/test_helpers/app.rs

@@ -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,
         }
     }
 }

src/test_helpers/dns.rs

@@ -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?;
 

src/test_helpers/linux.rs

@@ -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

src/tls/csr.rs

@@ -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\")] }"
+        )
     }
 }

src/tls/lib.rs

@@ -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")]

src/version.rs

@@ -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,
         )
     }
 }

tests/namespaced.rs

@@ -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();