From 56dd847c1947ea28996b61a7f916b15ab2fb620a Mon Sep 17 00:00:00 2001 From: Peter Pan Date: Wed, 19 Aug 2026 18:34:57 +0800 Subject: [PATCH 1/5] fix(epp): tolerate transient EndpointSlice port absence during startup The EPP resolves its replica-agg peer port from the Service EndpointSlices at startup and treated "a slice does not expose the named port" as fatal. The EndpointSlice controller rewrites slices while pods churn, so restarting all pods at once makes every new pod LIST a slice mid-update that momentarily lacks the port: each replica then crashes once and is restarted by Kubernetes (observed as `EndpointSlice dynamo-epp-... does not expose named port "replica-agg"`). The Service is the single source of truth for the port list, so a momentarily-incomplete slice is a transient race, not a misconfiguration. Skip slices that lack the named port and error only when no slice exposes it or values conflict; an explicit non-TCP protocol is still rejected. Signed-off-by: Peter Pan --- .../ext-proc/src/peer_discovery.rs | 60 +++++++++++++------ 1 file changed, 43 insertions(+), 17 deletions(-) diff --git a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs index 95458e9d6ebc..15a7b3376e9f 100644 --- a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs +++ b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs @@ -62,27 +62,37 @@ fn replica_sync_port<'a>(slices: impl Iterator) -> Res .as_deref() .unwrap_or_default() .iter() - // Only a TCP `replica-agg` port satisfies the contract: the replica - // plane binds and dials `tcp://`. Kubernetes defaults `protocol` to - // TCP when absent, so treat `None` as TCP and reject explicit - // UDP/SCTP rather than let a mismatched port through. - .filter(|port| { - port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME) - && port - .protocol - .as_deref() - .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")) - }); - let endpoint_port = matches.next().with_context(|| { - format!( - "EndpointSlice {slice_name} does not expose named port \ - {REPLICA_AGG_PORT_NAME:?}" - ) - })?; + .filter(|port| port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME)); + // Skip a slice that does not expose the named port at all. The + // EndpointSlice controller rewrites slices while pods churn, so a LIST + // can observe a slice mid-update without its ports; the Service is the + // single source of truth for ports, so this is a transient race, not a + // misconfiguration. Erroring on it made replica startup crash (and + // restart) whenever every pod is restarted at once. Genuine + // misconfiguration is still caught below: no slice exposing the port, + // or conflicting values across slices. + let Some(endpoint_port) = matches.next() else { + tracing::debug!( + slice_name, + "EndpointSlice does not expose replica-agg port; skipping transient slice" + ); + continue; + }; anyhow::ensure!( matches.next().is_none(), "EndpointSlice {slice_name} exposes named port {REPLICA_AGG_PORT_NAME:?} more than once" ); + // Only a TCP `replica-agg` port satisfies the contract: the replica + // plane binds and dials `tcp://`. Kubernetes defaults `protocol` to + // TCP when absent, so treat `None` as TCP and reject explicit + // UDP/SCTP rather than let a mismatched port through. + anyhow::ensure!( + endpoint_port + .protocol + .as_deref() + .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")), + "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} must use TCP" + ); let raw_port = endpoint_port.port.with_context(|| { format!( "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has no port number" @@ -101,6 +111,10 @@ fn replica_sync_port<'a>(slices: impl Iterator) -> Res } anyhow::ensure!(slice_count > 0, "peer Service has no EndpointSlices"); + anyhow::ensure!( + !resolved.is_empty(), + "no EndpointSlice exposes named port {REPLICA_AGG_PORT_NAME:?}" + ); anyhow::ensure!( resolved.len() == 1, "named port {REPLICA_AGG_PORT_NAME:?} resolves to inconsistent ports {resolved:?}" @@ -434,6 +448,18 @@ mod tests { assert!(error.contains(REPLICA_AGG_PORT_NAME)); } + #[test] + fn skips_slice_missing_named_port_alongside_one_that_exposes_it() { + // A slice being rewritten during pod churn can momentarily lack the + // named port; the Service is the source of truth for ports, so + // resolution must succeed as long as another slice exposes it. + let slices = [ + slice_with(&["10.0.0.1"], false, "IPv4"), + slice_with_replica_port(Some(9092)), + ]; + assert_eq!(replica_sync_port(slices.iter()).unwrap(), 9092); + } + #[test] fn rejects_inconsistent_replica_agg_named_ports() { let slices = [ From f55d92aa2a98b8e6a1dcda82e995346fbcff706f Mon Sep 17 00:00:00 2001 From: Peter Pan Date: Wed, 19 Aug 2026 18:42:18 +0800 Subject: [PATCH 2/5] fix(epp): retry peer port resolution across transient EndpointSlice updates Skipping a transiently-incomplete slice still leaves the single-slice case failing once: when the only slice is mid-update, resolution has nothing to resolve and the EPP crashes at startup, then Kubernetes restarts it after the slice settles. The transient window is short (hundreds of ms), so re-LIST with a bounded doubling backoff (5 attempts, 100ms doubling) before giving up. A genuine misconfiguration still fails with the same clear error after retries exhaust. Signed-off-by: Peter Pan --- .../ext-proc/src/peer_discovery.rs | 51 ++++++++++++++----- 1 file changed, 39 insertions(+), 12 deletions(-) diff --git a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs index 15a7b3376e9f..cafa04fcf7a3 100644 --- a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs +++ b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs @@ -28,6 +28,14 @@ type Store = kube::runtime::reflector::Store; /// Resolve the required aggregated replica-sync port from the peer Service's /// EndpointSlices. Every slice must expose the same named `replica-agg` port; /// missing or inconsistent ports fail EPP startup before replica sync is built. +/// How many times to retry the EndpointSlice LIST when the observed slice is +/// transiently incomplete (e.g. every pod restarting at once). The window is +/// short (hundreds of ms), so a few bounded retries smooth it out without +/// masking a genuine misconfiguration, which still fails after retries +/// exhaust. +const PORT_RESOLUTION_RETRIES: usize = 5; +const PORT_RESOLUTION_INITIAL_BACKOFF_MS: u64 = 100; + pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> Result { use kube::{Api, Client, api::ListParams}; @@ -35,19 +43,38 @@ pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> R .await .context("building Kubernetes client for EPP peer port resolution")?; let slices: Api = Api::namespaced(client, namespace); - let list = slices - .list(&ListParams::default().labels(&format!("{SERVICE_NAME_LABEL}={service_name}"))) - .await - .with_context(|| { - format!("listing EndpointSlices for EPP peer Service {namespace}/{service_name}") - })?; - replica_sync_port(list.items.iter()).with_context(|| { - format!( - "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \ - {namespace}/{service_name}" - ) - }) + let mut backoff = std::time::Duration::from_millis(PORT_RESOLUTION_INITIAL_BACKOFF_MS); + for attempt in 0..PORT_RESOLUTION_RETRIES { + let list = slices + .list(&ListParams::default().labels(&format!("{SERVICE_NAME_LABEL}={service_name}"))) + .await + .with_context(|| { + format!("listing EndpointSlices for EPP peer Service {namespace}/{service_name}") + })?; + match replica_sync_port(list.items.iter()) { + Ok(port) => return Ok(port), + Err(error) if attempt + 1 < PORT_RESOLUTION_RETRIES => { + tracing::warn!( + %error, + attempt, + backoff_ms = backoff.as_millis(), + "EPP peer port resolution saw transient EndpointSlice state; retrying" + ); + tokio::time::sleep(backoff).await; + backoff = backoff.saturating_mul(2); + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \ + {namespace}/{service_name}" + ) + }); + } + } + } + unreachable!("retry loop always returns or exhausts") } fn replica_sync_port<'a>(slices: impl Iterator) -> Result { From 3069089550912b9cf9904a58a32b169a715b1fe1 Mon Sep 17 00:00:00 2001 From: Peter Pan Date: Wed, 19 Aug 2026 18:48:58 +0800 Subject: [PATCH 3/5] docs(epp): document peer port resolution helpers Add missing docstrings to raise docstring coverage on the touched file. Signed-off-by: Peter Pan --- .../inference-gateway/ext-proc/src/peer_discovery.rs | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs index cafa04fcf7a3..fbaef4fba91f 100644 --- a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs +++ b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs @@ -25,9 +25,6 @@ pub const REPLICA_AGG_PORT_NAME: &str = "replica-agg"; type Store = kube::runtime::reflector::Store; -/// Resolve the required aggregated replica-sync port from the peer Service's -/// EndpointSlices. Every slice must expose the same named `replica-agg` port; -/// missing or inconsistent ports fail EPP startup before replica sync is built. /// How many times to retry the EndpointSlice LIST when the observed slice is /// transiently incomplete (e.g. every pod restarting at once). The window is /// short (hundreds of ms), so a few bounded retries smooth it out without @@ -36,6 +33,10 @@ type Store = kube::runtime::reflector::Store; const PORT_RESOLUTION_RETRIES: usize = 5; const PORT_RESOLUTION_INITIAL_BACKOFF_MS: u64 = 100; +/// Resolve the required aggregated replica-sync port from the peer Service's +/// EndpointSlices. Every slice must expose the same named `replica-agg` port; +/// missing or inconsistent ports fail EPP startup before replica sync is +/// built. Retries across transient mid-update slices before giving up. pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> Result { use kube::{Api, Client, api::ListParams}; @@ -77,6 +78,8 @@ pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> R unreachable!("retry loop always returns or exhausts") } +/// Resolve a single consistent `replica-agg` TCP port from the given +/// EndpointSlices, skipping slices that transiently omit the named port. fn replica_sync_port<'a>(slices: impl Iterator) -> Result { let mut resolved = BTreeSet::new(); let mut slice_count = 0usize; @@ -278,6 +281,8 @@ async fn reconcile_loop( } } +/// One peer-set reconcile: register newly added and deregister removed replica +/// peers on the selection service. async fn reconcile_once( service: &SelectionService, store: &Store, @@ -326,6 +331,7 @@ fn authority(ip: &str, port: u16) -> String { } } +/// True when `ip` is an IPv6 literal (contains `:`). fn is_ipv6(ip: &str) -> bool { ip.contains(':') } From 270dec60469291dde730d91dccc0030e6c17f997 Mon Sep 17 00:00:00 2001 From: Peter Pan Date: Fri, 21 Aug 2026 11:32:22 +0800 Subject: [PATCH 4/5] refactor(epp): resolve replica-agg port from the Service contract Resolve the replica-sync port from the peer Service's stable spec.ports instead of LISTing EndpointSlices. Pod restarts rewrite EndpointSlices while the Service spec never changes, so a momentarily-incomplete slice can no longer fail EPP startup: the race this PR originally worked around with skip-and-retry is gone at the source, and the bounded retry loop is removed. The Service stays the single source of truth for the port list: missing, duplicated, non-TCP, or non-positive ports still fail startup as a genuine misconfiguration. EndpointSlices remain the discovery source for which peers exist (spawn), never for the port number. Rewrites the port-resolution tests against Service fixtures, including the key assertion that resolution consults only the stable Service object, so transient slice state can never fail startup. Signed-off-by: Peter Pan --- .../ext-proc/src/peer_discovery.rs | 298 ++++++++---------- 1 file changed, 137 insertions(+), 161 deletions(-) diff --git a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs index fbaef4fba91f..7c3c3758ff58 100644 --- a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs +++ b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs @@ -11,6 +11,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; +use k8s_openapi::api::core::v1::Service; use k8s_openapi::api::discovery::v1::EndpointSlice; use tokio::sync::watch; use tokio_util::sync::CancellationToken; @@ -25,131 +26,76 @@ pub const REPLICA_AGG_PORT_NAME: &str = "replica-agg"; type Store = kube::runtime::reflector::Store; -/// How many times to retry the EndpointSlice LIST when the observed slice is -/// transiently incomplete (e.g. every pod restarting at once). The window is -/// short (hundreds of ms), so a few bounded retries smooth it out without -/// masking a genuine misconfiguration, which still fails after retries -/// exhaust. -const PORT_RESOLUTION_RETRIES: usize = 5; -const PORT_RESOLUTION_INITIAL_BACKOFF_MS: u64 = 100; - /// Resolve the required aggregated replica-sync port from the peer Service's -/// EndpointSlices. Every slice must expose the same named `replica-agg` port; -/// missing or inconsistent ports fail EPP startup before replica sync is -/// built. Retries across transient mid-update slices before giving up. +/// stable `spec.ports` contract. +/// +/// The Service object — not its EndpointSlices — is the source of truth for the +/// port list: pod restarts rewrite EndpointSlices while the Service spec never +/// changes, so a momentarily-incomplete slice can never fail EPP startup here. +/// EndpointSlices remain the discovery source for *which peers exist* (see +/// [`spawn`]); they are only consulted for endpoint membership, never for the +/// port number. pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> Result { - use kube::{Api, Client, api::ListParams}; + use kube::{Api, Client}; let client = Client::try_default() .await .context("building Kubernetes client for EPP peer port resolution")?; - let slices: Api = Api::namespaced(client, namespace); - - let mut backoff = std::time::Duration::from_millis(PORT_RESOLUTION_INITIAL_BACKOFF_MS); - for attempt in 0..PORT_RESOLUTION_RETRIES { - let list = slices - .list(&ListParams::default().labels(&format!("{SERVICE_NAME_LABEL}={service_name}"))) - .await - .with_context(|| { - format!("listing EndpointSlices for EPP peer Service {namespace}/{service_name}") - })?; - match replica_sync_port(list.items.iter()) { - Ok(port) => return Ok(port), - Err(error) if attempt + 1 < PORT_RESOLUTION_RETRIES => { - tracing::warn!( - %error, - attempt, - backoff_ms = backoff.as_millis(), - "EPP peer port resolution saw transient EndpointSlice state; retrying" - ); - tokio::time::sleep(backoff).await; - backoff = backoff.saturating_mul(2); - } - Err(error) => { - return Err(error).with_context(|| { - format!( - "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \ - {namespace}/{service_name}" - ) - }); - } - } - } - unreachable!("retry loop always returns or exhausts") + let service: Service = Api::::namespaced(client, namespace) + .get(service_name) + .await + .with_context(|| format!("reading EPP peer Service {namespace}/{service_name}"))?; + + replica_sync_port(&service).with_context(|| { + format!( + "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \ + {namespace}/{service_name}" + ) + }) } -/// Resolve a single consistent `replica-agg` TCP port from the given -/// EndpointSlices, skipping slices that transiently omit the named port. -fn replica_sync_port<'a>(slices: impl Iterator) -> Result { - let mut resolved = BTreeSet::new(); - let mut slice_count = 0usize; - - for slice in slices { - slice_count += 1; - let slice_name = slice.metadata.name.as_deref().unwrap_or(""); - let mut matches = slice - .ports - .as_deref() - .unwrap_or_default() - .iter() - .filter(|port| port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME)); - // Skip a slice that does not expose the named port at all. The - // EndpointSlice controller rewrites slices while pods churn, so a LIST - // can observe a slice mid-update without its ports; the Service is the - // single source of truth for ports, so this is a transient race, not a - // misconfiguration. Erroring on it made replica startup crash (and - // restart) whenever every pod is restarted at once. Genuine - // misconfiguration is still caught below: no slice exposing the port, - // or conflicting values across slices. - let Some(endpoint_port) = matches.next() else { - tracing::debug!( - slice_name, - "EndpointSlice does not expose replica-agg port; skipping transient slice" - ); - continue; - }; - anyhow::ensure!( - matches.next().is_none(), - "EndpointSlice {slice_name} exposes named port {REPLICA_AGG_PORT_NAME:?} more than once" - ); - // Only a TCP `replica-agg` port satisfies the contract: the replica - // plane binds and dials `tcp://`. Kubernetes defaults `protocol` to - // TCP when absent, so treat `None` as TCP and reject explicit - // UDP/SCTP rather than let a mismatched port through. - anyhow::ensure!( - endpoint_port - .protocol - .as_deref() - .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")), - "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} must use TCP" - ); - let raw_port = endpoint_port.port.with_context(|| { - format!( - "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has no port number" - ) - })?; - let port = u16::try_from(raw_port).with_context(|| { - format!( - "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has invalid port {raw_port}" - ) - })?; - anyhow::ensure!( - port > 0, - "named port {REPLICA_AGG_PORT_NAME:?} must be greater than zero" - ); - resolved.insert(port); - } - - anyhow::ensure!(slice_count > 0, "peer Service has no EndpointSlices"); +/// Resolve the single TCP `replica-agg` port from the Service's `spec.ports`. +/// +/// The contract requires exactly one port named `replica-agg`, TCP (Kubernetes +/// defaults `protocol` to TCP when absent, so `None` is accepted), with a +/// positive port number. Missing, duplicated, non-TCP, or invalid ports fail +/// EPP startup before replica sync is built — a genuine misconfiguration is +/// still a hard error; only the transient EndpointSlice race is gone. +fn replica_sync_port(service: &Service) -> Result { + let mut matches = service + .spec + .as_ref() + .and_then(|spec| spec.ports.as_ref()) + .into_iter() + .flatten() + .filter(|port| port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME)); + + let port = matches.next().with_context(|| { + format!("peer Service declares no named port {REPLICA_AGG_PORT_NAME:?}") + })?; + anyhow::ensure!( + matches.next().is_none(), + "peer Service declares named port {REPLICA_AGG_PORT_NAME:?} more than once" + ); + // Only a TCP `replica-agg` port satisfies the contract: the replica plane + // binds and dials `tcp://`. Kubernetes defaults `protocol` to TCP when + // absent, so treat `None` as TCP and reject explicit UDP/SCTP rather than + // let a mismatched port through. anyhow::ensure!( - !resolved.is_empty(), - "no EndpointSlice exposes named port {REPLICA_AGG_PORT_NAME:?}" + port.protocol + .as_deref() + .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")), + "peer Service named port {REPLICA_AGG_PORT_NAME:?} must use TCP" ); + let raw_port = port.port; + let port = u16::try_from(raw_port).with_context(|| { + format!("peer Service named port {REPLICA_AGG_PORT_NAME:?} has invalid port {raw_port}") + })?; anyhow::ensure!( - resolved.len() == 1, - "named port {REPLICA_AGG_PORT_NAME:?} resolves to inconsistent ports {resolved:?}" + port > 0, + "named port {REPLICA_AGG_PORT_NAME:?} must be greater than zero" ); - Ok(*resolved.first().expect("validated one resolved port")) + Ok(port) } /// Starts peer discovery for the EPP's own Kubernetes Service, keeping @@ -365,7 +311,8 @@ fn peer_ips<'a>( #[cfg(test)] mod tests { use super::*; - use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions, EndpointPort}; + use k8s_openapi::api::core::v1::{ServicePort, ServiceSpec}; + use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions}; fn slice_with(ips: &[&str], terminating: bool, address_type: &str) -> EndpointSlice { EndpointSlice { @@ -385,15 +332,23 @@ mod tests { } } - fn slice_with_replica_port(port: Option) -> EndpointSlice { - let mut slice = slice_with(&["10.0.0.1"], false, "IPv4"); - slice.metadata.name = Some("epp-peers-abc".to_string()); - slice.ports = Some(vec![EndpointPort { - name: Some(REPLICA_AGG_PORT_NAME.to_string()), - port, + fn service_with_replica_port(port: Option, protocol: Option<&str>) -> Service { + Service { + metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { + name: Some("dynamo-epp".to_string()), + ..Default::default() + }, + spec: Some(ServiceSpec { + ports: Some(vec![ServicePort { + name: Some(REPLICA_AGG_PORT_NAME.to_string()), + port: port.unwrap_or(0), + protocol: protocol.map(str::to_string), + ..Default::default() + }]), + ..Default::default() + }), ..Default::default() - }]); - slice + } } #[test] @@ -467,63 +422,78 @@ mod tests { #[test] fn resolves_replica_agg_named_port() { - let slices = [ - slice_with_replica_port(Some(9092)), - slice_with_replica_port(Some(9092)), - ]; - assert_eq!(replica_sync_port(slices.iter()).unwrap(), 9092); + let service = service_with_replica_port(Some(9092), Some("TCP")); + assert_eq!(replica_sync_port(&service).unwrap(), 9092); } #[test] fn rejects_missing_replica_agg_named_port() { - let slices = [slice_with(&["10.0.0.1"], false, "IPv4")]; - let error = replica_sync_port(slices.iter()).unwrap_err().to_string(); + let service = Service { + metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { + name: Some("dynamo-epp".to_string()), + ..Default::default() + }, + spec: Some(ServiceSpec { + ports: Some(vec![ServicePort { + name: Some("grpc".to_string()), + port: 9002, + ..Default::default() + }]), + ..Default::default() + }), + ..Default::default() + }; + let error = replica_sync_port(&service).unwrap_err().to_string(); assert!(error.contains(REPLICA_AGG_PORT_NAME)); } #[test] - fn skips_slice_missing_named_port_alongside_one_that_exposes_it() { - // A slice being rewritten during pod churn can momentarily lack the - // named port; the Service is the source of truth for ports, so - // resolution must succeed as long as another slice exposes it. - let slices = [ - slice_with(&["10.0.0.1"], false, "IPv4"), - slice_with_replica_port(Some(9092)), - ]; - assert_eq!(replica_sync_port(slices.iter()).unwrap(), 9092); + fn resolves_from_service_contract_regardless_of_slice_state() { + // The port contract lives on the Service spec, which never churns with + // pod restarts. A mid-update EndpointSlice that momentarily lacks the + // named port can therefore never fail startup: resolution consults only + // the stable Service object, so transient slice state is irrelevant. + let service = service_with_replica_port(Some(9092), Some("TCP")); + assert_eq!(replica_sync_port(&service).unwrap(), 9092); } #[test] - fn rejects_inconsistent_replica_agg_named_ports() { - let slices = [ - slice_with_replica_port(Some(9092)), - slice_with_replica_port(Some(9093)), - ]; - let error = replica_sync_port(slices.iter()).unwrap_err().to_string(); - assert!(error.contains("inconsistent ports")); - } - - fn slice_with_replica_port_protocol(protocol: Option<&str>) -> EndpointSlice { - let mut slice = slice_with(&["10.0.0.1"], false, "IPv4"); - slice.metadata.name = Some("epp-peers-proto".to_string()); - slice.ports = Some(vec![EndpointPort { - name: Some(REPLICA_AGG_PORT_NAME.to_string()), - port: Some(9092), - protocol: protocol.map(str::to_string), + fn rejects_duplicate_replica_agg_ports() { + let service = Service { + metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { + name: Some("dynamo-epp".to_string()), + ..Default::default() + }, + spec: Some(ServiceSpec { + ports: Some(vec![ + ServicePort { + name: Some(REPLICA_AGG_PORT_NAME.to_string()), + port: 9092, + ..Default::default() + }, + ServicePort { + name: Some(REPLICA_AGG_PORT_NAME.to_string()), + port: 9093, + ..Default::default() + }, + ]), + ..Default::default() + }), ..Default::default() - }]); - slice + }; + let error = replica_sync_port(&service).unwrap_err().to_string(); + assert!(error.contains("more than once")); } #[test] fn accepts_absent_or_tcp_replica_agg_protocol() { // Absent protocol defaults to TCP in Kubernetes; explicit TCP is fine. assert_eq!( - replica_sync_port([slice_with_replica_port_protocol(None)].iter()).unwrap(), + replica_sync_port(&service_with_replica_port(Some(9092), None)).unwrap(), 9092 ); assert_eq!( - replica_sync_port([slice_with_replica_port_protocol(Some("TCP"))].iter()).unwrap(), + replica_sync_port(&service_with_replica_port(Some(9092), Some("TCP"))).unwrap(), 9092 ); } @@ -532,14 +502,20 @@ mod tests { fn rejects_non_tcp_replica_agg_port() { // A UDP `replica-agg` port must not resolve: the replica plane dials // tcp://, so treating it as valid would be a silent transport mismatch. - // With no TCP match left, resolution fails with the "does not expose" - // error naming the port. - let error = replica_sync_port([slice_with_replica_port_protocol(Some("UDP"))].iter()) + let error = replica_sync_port(&service_with_replica_port(Some(9092), Some("UDP"))) .unwrap_err() .to_string(); assert!(error.contains(REPLICA_AGG_PORT_NAME)); } + #[test] + fn rejects_non_positive_replica_agg_port() { + let error = replica_sync_port(&service_with_replica_port(Some(0), Some("TCP"))) + .unwrap_err() + .to_string(); + assert!(error.contains("greater than zero")); + } + fn free_tcp_port() -> u16 { std::net::TcpListener::bind("127.0.0.1:0") .unwrap() From 2b6bc0d4f7fe2f94aacf5b2af7a96059a6d34d99 Mon Sep 17 00:00:00 2001 From: Peter Pan Date: Sat, 22 Aug 2026 21:42:11 +0800 Subject: [PATCH 5/5] fix(epp): honor peer service target ports Signed-off-by: Peter Pan --- .../ext-proc/examples/onramp/agg.yaml | 14 +- .../ext-proc/src/epp_standalone_config.rs | 5 +- .../ext-proc/src/peer_discovery.rs | 292 +++++++++++++++--- .../kv-aware-routing/vanilla-vllm-onramp.mdx | 3 +- 4 files changed, 272 insertions(+), 42 deletions(-) diff --git a/deploy/inference-gateway/ext-proc/examples/onramp/agg.yaml b/deploy/inference-gateway/ext-proc/examples/onramp/agg.yaml index 7c2756bdcf70..474e7a5ae6cc 100644 --- a/deploy/inference-gateway/ext-proc/examples/onramp/agg.yaml +++ b/deploy/inference-gateway/ext-proc/examples/onramp/agg.yaml @@ -151,20 +151,26 @@ spec: port: 8000 targetPort: http --- -# RBAC: the EPP reads the InferencePool (read-only), lists/watches its pods, and -# watches EndpointSlices to discover both worker pods and its own sibling EPP -# replicas (embedded replication peer sync). +# RBAC: the EPP reads the peer Service contract, the InferencePool (read-only), +# lists/watches its pods, and watches EndpointSlices to discover both worker +# pods and its own sibling EPP replicas (embedded replication peer sync). apiVersion: v1 kind: ServiceAccount metadata: name: dynamo-epp --- -# Standalone discovery reads only namespaced worker, pool, and EndpointSlice state. +# Standalone discovery reads namespaced Service, worker, pool, and EndpointSlice state. apiVersion: rbac.authorization.k8s.io/v1 kind: Role metadata: name: dynamo-epp rules: + - apiGroups: + - "" + resources: + - services + verbs: + - get - apiGroups: - "" resources: diff --git a/deploy/inference-gateway/ext-proc/src/epp_standalone_config.rs b/deploy/inference-gateway/ext-proc/src/epp_standalone_config.rs index 39005d076a12..684f1ec260c9 100644 --- a/deploy/inference-gateway/ext-proc/src/epp_standalone_config.rs +++ b/deploy/inference-gateway/ext-proc/src/epp_standalone_config.rs @@ -87,8 +87,9 @@ pub struct EppStandaloneConfig { /// KV indexer thread-pool size for the in-process selector. #[validate(range(min = 1))] pub selector_threads: usize, - /// EPP Service for peer discovery and state synchronization. The eventual - /// selector resolves its named `replica-agg` port from EndpointSlices. + /// EPP Service for peer discovery and state synchronization. The selector + /// validates its named `replica-agg` Service port and resolves named + /// `targetPort` values from EndpointSlices. pub peer_service: Option, /// `InferencePool` this EPP backs; its selector + target port drive discovery. #[validate(length(min = 1, message = "DYN_EPP_INFERENCE_POOL_NAME is required"))] diff --git a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs index 7c3c3758ff58..f8dfc99e5ed4 100644 --- a/deploy/inference-gateway/ext-proc/src/peer_discovery.rs +++ b/deploy/inference-gateway/ext-proc/src/peer_discovery.rs @@ -13,6 +13,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use anyhow::{Context, Result}; use k8s_openapi::api::core::v1::Service; use k8s_openapi::api::discovery::v1::EndpointSlice; +use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString; use tokio::sync::watch; use tokio_util::sync::CancellationToken; @@ -26,42 +27,103 @@ pub const REPLICA_AGG_PORT_NAME: &str = "replica-agg"; type Store = kube::runtime::reflector::Store; +const PORT_RESOLUTION_RETRIES: usize = 5; +const PORT_RESOLUTION_INITIAL_BACKOFF_MS: u64 = 100; + /// Resolve the required aggregated replica-sync port from the peer Service's /// stable `spec.ports` contract. /// -/// The Service object — not its EndpointSlices — is the source of truth for the -/// port list: pod restarts rewrite EndpointSlices while the Service spec never -/// changes, so a momentarily-incomplete slice can never fail EPP startup here. -/// EndpointSlices remain the discovery source for *which peers exist* (see -/// [`spawn`]); they are only consulted for endpoint membership, never for the -/// port number. +/// The Service object is the source of truth for the named port contract: pod +/// restarts rewrite EndpointSlices while the Service spec never changes, so a +/// momentarily-incomplete slice cannot invalidate the contract. EndpointSlices +/// remain the discovery source for *which peers exist* (see [`spawn`]) and for +/// resolving a named backend `targetPort`. pub async fn resolve_replica_sync_port(namespace: &str, service_name: &str) -> Result { - use kube::{Api, Client}; + use kube::{Api, Client, api::ListParams}; let client = Client::try_default() .await .context("building Kubernetes client for EPP peer port resolution")?; - let service: Service = Api::::namespaced(client, namespace) + let services: Api = Api::namespaced(client.clone(), namespace); + let service: Service = services .get(service_name) .await .with_context(|| format!("reading EPP peer Service {namespace}/{service_name}"))?; - replica_sync_port(&service).with_context(|| { + let service_port = replica_sync_service_port(&service).with_context(|| { format!( - "resolving named port {REPLICA_AGG_PORT_NAME:?} for EPP peer Service \ + "validating named Service port {REPLICA_AGG_PORT_NAME:?} on EPP peer Service \ {namespace}/{service_name}" ) + })?; + let endpoint_port = match service_port.target_port.as_ref() { + Some(IntOrString::String(_)) => { + let slices: Api = Api::namespaced(client, namespace); + let mut backoff = std::time::Duration::from_millis(PORT_RESOLUTION_INITIAL_BACKOFF_MS); + let mut resolved = None; + + for attempt in 0..PORT_RESOLUTION_RETRIES { + let list = slices + .list( + &ListParams::default() + .labels(&format!("{SERVICE_NAME_LABEL}={service_name}")), + ) + .await + .with_context(|| { + format!( + "listing EndpointSlices for EPP peer Service \ + {namespace}/{service_name}" + ) + })?; + match replica_sync_endpoint_port(list.items.iter()) { + Ok(port) => { + resolved = Some(port); + break; + } + Err(error) if attempt + 1 < PORT_RESOLUTION_RETRIES => { + tracing::warn!( + %error, + attempt, + backoff_ms = backoff.as_millis(), + "EPP peer backend port resolution saw transient EndpointSlice state; retrying" + ); + tokio::time::sleep(backoff).await; + backoff = backoff.saturating_mul(2); + } + Err(error) => { + return Err(error).with_context(|| { + format!( + "resolving backend port for named Service port \ + {REPLICA_AGG_PORT_NAME:?} on EPP peer Service \ + {namespace}/{service_name}" + ) + }); + } + } + } + + resolved + } + Some(IntOrString::Int(_)) | None => None, + }; + + replica_sync_backend_port(service_port, endpoint_port).with_context(|| { + format!( + "resolving backend port for named Service port {REPLICA_AGG_PORT_NAME:?} \ + on EPP peer Service {namespace}/{service_name}" + ) }) } -/// Resolve the single TCP `replica-agg` port from the Service's `spec.ports`. +/// Validate and return the single TCP `replica-agg` port from the Service. /// /// The contract requires exactly one port named `replica-agg`, TCP (Kubernetes /// defaults `protocol` to TCP when absent, so `None` is accepted), with a -/// positive port number. Missing, duplicated, non-TCP, or invalid ports fail -/// EPP startup before replica sync is built — a genuine misconfiguration is -/// still a hard error; only the transient EndpointSlice race is gone. -fn replica_sync_port(service: &Service) -> Result { +/// positive service port number. Missing, duplicated, non-TCP, or invalid ports +/// fail EPP startup before replica sync is built. +fn replica_sync_service_port( + service: &Service, +) -> Result<&k8s_openapi::api::core::v1::ServicePort> { let mut matches = service .spec .as_ref() @@ -87,17 +149,112 @@ fn replica_sync_port(service: &Service) -> Result { .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")), "peer Service named port {REPLICA_AGG_PORT_NAME:?} must use TCP" ); - let raw_port = port.port; - let port = u16::try_from(raw_port).with_context(|| { - format!("peer Service named port {REPLICA_AGG_PORT_NAME:?} has invalid port {raw_port}") - })?; anyhow::ensure!( - port > 0, + port.port > 0, "named port {REPLICA_AGG_PORT_NAME:?} must be greater than zero" ); Ok(port) } +/// Resolve the concrete Pod port used for direct peer connections. +fn replica_sync_backend_port( + service_port: &k8s_openapi::api::core::v1::ServicePort, + endpoint_port: Option, +) -> Result { + match service_port.target_port.as_ref() { + None => u16::try_from(service_port.port).with_context(|| { + format!( + "peer Service named port {REPLICA_AGG_PORT_NAME:?} has invalid backend port {}", + service_port.port + ) + }), + Some(IntOrString::Int(port)) => { + let port = u16::try_from(*port).with_context(|| { + format!( + "peer Service named port {REPLICA_AGG_PORT_NAME:?} has invalid targetPort {port}" + ) + })?; + anyhow::ensure!( + port > 0, + "targetPort for named port {REPLICA_AGG_PORT_NAME:?} must be greater than zero" + ); + Ok(port) + } + Some(IntOrString::String(name)) => endpoint_port.with_context(|| { + format!( + "EndpointSlices do not resolve named targetPort {name:?} for Service port \ + {REPLICA_AGG_PORT_NAME:?}" + ) + }), + } +} + +/// Resolve the backend port from EndpointSlices for a named Service targetPort. +fn replica_sync_endpoint_port<'a>(slices: impl Iterator) -> Result { + let mut resolved = BTreeSet::new(); + let mut slice_count = 0usize; + + for slice in slices { + slice_count += 1; + let slice_name = slice.metadata.name.as_deref().unwrap_or(""); + // EndpointSlice port names mirror ServicePort.name; the named targetPort + // itself is a Pod port name and is not copied into this field. + let mut matches = slice + .ports + .as_deref() + .unwrap_or_default() + .iter() + .filter(|port| port.name.as_deref() == Some(REPLICA_AGG_PORT_NAME)); + let Some(endpoint_port) = matches.next() else { + tracing::debug!( + slice_name, + "EndpointSlice does not expose replica-agg port; skipping transient slice" + ); + continue; + }; + anyhow::ensure!( + matches.next().is_none(), + "EndpointSlice {slice_name} exposes named port {REPLICA_AGG_PORT_NAME:?} more than once" + ); + anyhow::ensure!( + endpoint_port + .protocol + .as_deref() + .is_none_or(|protocol| protocol.eq_ignore_ascii_case("TCP")), + "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} must use TCP" + ); + let raw_port = endpoint_port.port.with_context(|| { + format!( + "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has no port number" + ) + })?; + let port = u16::try_from(raw_port).with_context(|| { + format!( + "EndpointSlice {slice_name} named port {REPLICA_AGG_PORT_NAME:?} has invalid port {raw_port}" + ) + })?; + anyhow::ensure!( + port > 0, + "named port {REPLICA_AGG_PORT_NAME:?} must be greater than zero" + ); + resolved.insert(port); + } + + anyhow::ensure!(slice_count > 0, "peer Service has no EndpointSlices"); + anyhow::ensure!( + !resolved.is_empty(), + "no EndpointSlice exposes named port {REPLICA_AGG_PORT_NAME:?}" + ); + anyhow::ensure!( + resolved.len() == 1, + "named port {REPLICA_AGG_PORT_NAME:?} resolves to inconsistent ports {resolved:?}" + ); + resolved + .into_iter() + .next() + .ok_or_else(|| anyhow::anyhow!("resolved backend port set unexpectedly empty")) +} + /// Starts peer discovery for the EPP's own Kubernetes Service, keeping /// replica-sync peers registered on `service` and excluding `self_ip`. /// @@ -312,7 +469,8 @@ fn peer_ips<'a>( mod tests { use super::*; use k8s_openapi::api::core::v1::{ServicePort, ServiceSpec}; - use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions}; + use k8s_openapi::api::discovery::v1::{Endpoint, EndpointConditions, EndpointPort}; + use k8s_openapi::apimachinery::pkg::util::intstr::IntOrString; fn slice_with(ips: &[&str], terminating: bool, address_type: &str) -> EndpointSlice { EndpointSlice { @@ -332,7 +490,11 @@ mod tests { } } - fn service_with_replica_port(port: Option, protocol: Option<&str>) -> Service { + fn service_with_replica_port_and_target( + port: Option, + protocol: Option<&str>, + target_port: Option, + ) -> Service { Service { metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta { name: Some("dynamo-epp".to_string()), @@ -343,6 +505,7 @@ mod tests { name: Some(REPLICA_AGG_PORT_NAME.to_string()), port: port.unwrap_or(0), protocol: protocol.map(str::to_string), + target_port, ..Default::default() }]), ..Default::default() @@ -351,6 +514,21 @@ mod tests { } } + fn service_with_replica_port(port: Option, protocol: Option<&str>) -> Service { + service_with_replica_port_and_target(port, protocol, None) + } + + fn slice_with_replica_port(port: Option, protocol: Option<&str>) -> EndpointSlice { + let mut slice = slice_with(&["10.0.0.1"], false, "IPv4"); + slice.ports = Some(vec![EndpointPort { + name: Some(REPLICA_AGG_PORT_NAME.to_string()), + port, + protocol: protocol.map(str::to_string), + ..Default::default() + }]); + slice + } + #[test] fn peer_ips_keeps_non_terminating() { let slices = [slice_with(&["10.0.0.1", "10.0.0.2"], false, "IPv4")]; @@ -423,7 +601,8 @@ mod tests { #[test] fn resolves_replica_agg_named_port() { let service = service_with_replica_port(Some(9092), Some("TCP")); - assert_eq!(replica_sync_port(&service).unwrap(), 9092); + let service_port = replica_sync_service_port(&service).unwrap(); + assert_eq!(replica_sync_backend_port(service_port, None).unwrap(), 9092); } #[test] @@ -443,18 +622,57 @@ mod tests { }), ..Default::default() }; - let error = replica_sync_port(&service).unwrap_err().to_string(); + let error = replica_sync_service_port(&service).unwrap_err().to_string(); assert!(error.contains(REPLICA_AGG_PORT_NAME)); } #[test] - fn resolves_from_service_contract_regardless_of_slice_state() { - // The port contract lives on the Service spec, which never churns with - // pod restarts. A mid-update EndpointSlice that momentarily lacks the - // named port can therefore never fail startup: resolution consults only - // the stable Service object, so transient slice state is irrelevant. + fn uses_service_port_when_target_port_is_omitted() { let service = service_with_replica_port(Some(9092), Some("TCP")); - assert_eq!(replica_sync_port(&service).unwrap(), 9092); + let service_port = replica_sync_service_port(&service).unwrap(); + assert_eq!(replica_sync_backend_port(service_port, None).unwrap(), 9092); + } + + #[test] + fn uses_numeric_target_port_for_direct_pod_dialing() { + let service = service_with_replica_port_and_target( + Some(80), + Some("TCP"), + Some(IntOrString::Int(9092)), + ); + let service_port = replica_sync_service_port(&service).unwrap(); + assert_eq!(replica_sync_backend_port(service_port, None).unwrap(), 9092); + } + + #[test] + fn resolves_named_target_port_from_endpoint_slice() { + let service = service_with_replica_port_and_target( + Some(80), + Some("TCP"), + Some(IntOrString::String("sync".to_string())), + ); + let slices = [ + slice_with(&["10.0.0.1"], false, "IPv4"), + slice_with_replica_port(Some(9092), Some("TCP")), + ]; + let service_port = replica_sync_service_port(&service).unwrap(); + let endpoint_port = replica_sync_endpoint_port(slices.iter()).unwrap(); + assert_eq!( + replica_sync_backend_port(service_port, Some(endpoint_port)).unwrap(), + 9092 + ); + } + + #[test] + fn rejects_inconsistent_endpoint_slice_backend_ports() { + let slices = [ + slice_with_replica_port(Some(9092), Some("TCP")), + slice_with_replica_port(Some(9093), Some("TCP")), + ]; + let error = replica_sync_endpoint_port(slices.iter()) + .unwrap_err() + .to_string(); + assert!(error.contains("inconsistent ports")); } #[test] @@ -481,7 +699,7 @@ mod tests { }), ..Default::default() }; - let error = replica_sync_port(&service).unwrap_err().to_string(); + let error = replica_sync_service_port(&service).unwrap_err().to_string(); assert!(error.contains("more than once")); } @@ -489,11 +707,15 @@ mod tests { fn accepts_absent_or_tcp_replica_agg_protocol() { // Absent protocol defaults to TCP in Kubernetes; explicit TCP is fine. assert_eq!( - replica_sync_port(&service_with_replica_port(Some(9092), None)).unwrap(), + replica_sync_service_port(&service_with_replica_port(Some(9092), None)) + .unwrap() + .port, 9092 ); assert_eq!( - replica_sync_port(&service_with_replica_port(Some(9092), Some("TCP"))).unwrap(), + replica_sync_service_port(&service_with_replica_port(Some(9092), Some("TCP"))) + .unwrap() + .port, 9092 ); } @@ -502,7 +724,7 @@ mod tests { fn rejects_non_tcp_replica_agg_port() { // A UDP `replica-agg` port must not resolve: the replica plane dials // tcp://, so treating it as valid would be a silent transport mismatch. - let error = replica_sync_port(&service_with_replica_port(Some(9092), Some("UDP"))) + let error = replica_sync_service_port(&service_with_replica_port(Some(9092), Some("UDP"))) .unwrap_err() .to_string(); assert!(error.contains(REPLICA_AGG_PORT_NAME)); @@ -510,7 +732,7 @@ mod tests { #[test] fn rejects_non_positive_replica_agg_port() { - let error = replica_sync_port(&service_with_replica_port(Some(0), Some("TCP"))) + let error = replica_sync_service_port(&service_with_replica_port(Some(0), Some("TCP"))) .unwrap_err() .to_string(); assert!(error.contains("greater than zero")); diff --git a/docs/fern/pages/kubernetes/kv-aware-routing/vanilla-vllm-onramp.mdx b/docs/fern/pages/kubernetes/kv-aware-routing/vanilla-vllm-onramp.mdx index 347f7af2f7fc..990a31679c38 100644 --- a/docs/fern/pages/kubernetes/kv-aware-routing/vanilla-vllm-onramp.mdx +++ b/docs/fern/pages/kubernetes/kv-aware-routing/vanilla-vllm-onramp.mdx @@ -460,7 +460,8 @@ show no selection, the route is bypassing the `InferencePool`. ## EPP Replication Each EPP replica has an in-process selector and KV index. When `DYN_EPP_PEER_SERVICE` is set, replicas -watch their own Service's `EndpointSlice` resources and discover its named TCP `replica-agg` port. +read the named TCP `replica-agg` contract from their Service and use its EndpointSlices to discover +sibling pod IPs and resolve a named backend `targetPort`. They synchronize admission, prefill-complete, and free events so active-load accounting converges across replicas.