diff --git a/charts/grid-operator/values.yaml b/charts/grid-operator/values.yaml index 11d4e0b63..4a7123b73 100644 --- a/charts/grid-operator/values.yaml +++ b/charts/grid-operator/values.yaml @@ -303,13 +303,13 @@ signals: scrapeIntervalSeconds: "" # -- Advertised site gateway address configuration. gateway: - # -- Discover and advertise a LoadBalancer address for the site gateway. + # -- Discover a LoadBalancer address for SWIM and the own GridSite. discoveryEnabled: true # -- GridNetwork consumerGateways[].praxisConfig for the site gateway, as the CRD # takes it. Empty renders no block. praxisConfig: {} # -- Reachable site gateway host:port override. Set this for ClusterIP or - # NodePort Services; otherwise the operator discovers a LoadBalancer address. + # NodePort Services; the own GridSite uses it when gatewayEndpoint is absent. address: "" # -- Site gateway Kubernetes Service name for the operator to discover. Defaults # to grid-gateway with enrollment enabled or grid.id set; it also names the diff --git a/crates/operator/src/controller/grid_site.rs b/crates/operator/src/controller/grid_site.rs index 6bb9540c5..a4303849f 100644 --- a/crates/operator/src/controller/grid_site.rs +++ b/crates/operator/src/controller/grid_site.rs @@ -6,7 +6,7 @@ //! //! [`GridSite`]: crate::crd::grid_site::GridSite -use std::sync::Arc; +use std::{borrow::Cow, sync::Arc}; use k8s_openapi::api::core::v1::ObjectReference; use kube::{ @@ -24,7 +24,7 @@ use crate::{ controller::grid_network, crd::{ grid_network::{GridNetwork, PeerTrustMode}, - grid_site::{EgressTlsMode, GridSite, GridSitePhase, GridSiteStatus}, + grid_site::{EgressConfig, EgressTls, EgressTlsMode, GridSite, GridSitePhase, GridSiteStatus}, inference_provider::ConditionStatus, }, error::OperatorError, @@ -73,6 +73,10 @@ pub struct SiteContext { pub client: Arc, /// How this site learns its members; declared sites leave `Pending` without discovery only under Local. pub membership: crate::cli::Membership, + /// This operator's site name, when it has an identity. + pub site_name: Option, + /// The configured or discovered address for this operator's gateway. + pub gateway_address: tokio::sync::watch::Receiver>, } /// Reconcile a [`GridSite`] resource. @@ -98,6 +102,9 @@ pub async fn reconcile(site: Arc, ctx: Arc) -> Result network, // A declared site can precede its GridNetwork, as this cluster's own does while it enrolls. @@ -111,8 +118,8 @@ pub async fn reconcile(site: Arc, ctx: Arc) -> Result, ctx: Arc) -> Result bool { // Phase Determination // --------------------------------------------------------------------------- +/// How the own site's gateway endpoint was supplied for reconciliation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum GatewayAddressSource { + /// The `GridSite` declares a usable gateway endpoint or operator settings supplied one. + Available, + /// This is a peer `GridSite`; the local gateway setting does not apply. + Peer, + /// The own `GridSite` has no declared or resolved gateway address. + Undeclared, +} + +/// Add this operator's resolved address to a reconciliation copy of its own `GridSite`. +fn apply_local_gateway_address<'site>( + site: &'site GridSite, + local_site: Option<&str>, + address: Option<&str>, +) -> (Cow<'site, GridSite>, GatewayAddressSource) { + let Some(name) = site.metadata.name.as_deref() else { + return (Cow::Borrowed(site), GatewayAddressSource::Peer); + }; + if local_site != Some(name) { + return (Cow::Borrowed(site), GatewayAddressSource::Peer); + } + if site + .spec + .gateway_endpoint + .as_ref() + .is_some_and(|endpoint| !endpoint.address.trim().is_empty()) + { + return (Cow::Borrowed(site), GatewayAddressSource::Available); + } + if let Some(address) = address.map(str::trim).filter(|address| !address.is_empty()) { + let mut reconciled = site.clone(); + reconciled.spec.gateway_endpoint = Some(EgressConfig { + address: address.to_owned(), + tls: EgressTls { + server_name: Some(crate::signals::gateway_server_name(name)), + ..EgressTls::default() + }, + }); + return (Cow::Owned(reconciled), GatewayAddressSource::Available); + } + (Cow::Borrowed(site), GatewayAddressSource::Undeclared) +} + +/// Advance a site's phase, giving the own site an actionable reason when its endpoint is absent. +fn site_phase_for_gateway( + current: &GridSitePhase, + site: &GridSite, + outcome: Option<&GatewayProbeOutcome>, + membership: crate::cli::Membership, + gateway_source: GatewayAddressSource, +) -> (GridSitePhase, String, String) { + let (phase, reason, message) = + site_phase_under(current, site, outcome, membership == crate::cli::Membership::Local); + if gateway_source == GatewayAddressSource::Undeclared && phase != GridSitePhase::Left { + return ( + phase, + "GatewayAddressMissing".to_owned(), + "this site's gateway address is not declared or discoverable; set gateway.address in the operator configuration" + .to_owned(), + ); + } + (phase, reason, message) +} + /// The next lifecycle phase under Swim membership, as [`site_phase_under`] with `declared` false. #[cfg(test)] pub(crate) fn site_phase_next( @@ -949,37 +1027,23 @@ mod tests { ) } + /// Checks that Connected status preserves each lifecycle reason and its verdict. #[test] fn connected_names_the_cause() { - assert_eq!(connected_for("TlsVerified"), ("True", CONNECTED.to_owned(), None)); - assert_eq!( - connected_for("AdvertisedCertMismatch"), - ("True", CONNECTED.to_owned(), None) - ); - assert_eq!( - connected_for("ConnectTimeout"), - ("False", "Unreachable".to_owned(), None) - ); - assert_eq!( - connected_for("PlaintextUnreachable"), - ("False", "Unreachable".to_owned(), None) - ); - assert_eq!( - connected_for("HandshakeTimeout"), - ("False", "TLSProtocolError".to_owned(), None) - ); - assert_eq!( - connected_for("GatewayAddressMissing"), - ("False", "AddressMissing".to_owned(), None) - ); - assert_eq!( - connected_for("GossipedAddressRefused"), - ("False", "AddressRefused".to_owned(), None) - ); - assert_eq!( - connected_for("UntrustedIssuer"), - ("False", "UntrustedIssuer".to_owned(), None) - ); + let cases = [ + ("TlsVerified", ("True", CONNECTED, None)), + ("AdvertisedCertMismatch", ("True", CONNECTED, None)), + ("ConnectTimeout", ("False", "Unreachable", None)), + ("PlaintextUnreachable", ("False", "Unreachable", None)), + ("HandshakeTimeout", ("False", "TLSProtocolError", None)), + ("GatewayAddressMissing", ("False", "AddressMissing", None)), + ("GossipedAddressRefused", ("False", "AddressRefused", None)), + ("UntrustedIssuer", ("False", "UntrustedIssuer", None)), + ]; + for (reason, expected) in cases { + let (status, connected, moving) = connected_for(reason); + assert_eq!((status, connected.as_str(), moving), expected); + } } #[test] @@ -1017,7 +1081,7 @@ mod tests { "patch should carry null resourceVersion when not set" ); } - use crate::crd::grid_site::{EgressConfig, EgressTls, GridSiteSpec}; + use crate::crd::grid_site::GridSiteSpec; fn site_with_egress(phase: Option, egress: &str) -> GridSite { GridSite { @@ -1106,13 +1170,80 @@ mod tests { assert_eq!(reason, "EndpointsDeclared"); } + /// Checks the missing-address reason for an own site with no declared endpoint. #[test] - fn pending_declared_without_a_gateway_stays_pending() { + fn pending_declared_without_a_gateway_reports_missing() { let mut site = declared_at("peer.example.com:443"); site.spec.gateway_endpoint = None; - let (next, reason, _msg) = site_phase_under(&GridSitePhase::Pending, &site, None, true); + let (site, source) = apply_local_gateway_address(&site, Some("test-site"), None); + let (next, reason, message) = site_phase_for_gateway( + &GridSitePhase::Pending, + &site, + None, + crate::cli::Membership::Local, + source, + ); assert_eq!(next, GridSitePhase::Pending, "no gateway, no endpoints"); - assert_eq!(reason, "AwaitingDiscovery"); + assert_eq!(reason, "GatewayAddressMissing"); + assert!( + message.contains("gateway.address"), + "message names the configuration setting" + ); + } + + /// Checks that a resolved operator address advances its own site without changing the API spec. + #[test] + fn local_gateway_override_advances_the_own_site_without_mutating_its_spec() { + let site = site_no_egress(Some(GridSitePhase::Pending)); + let (reconciled, source) = apply_local_gateway_address(&site, Some("test-site"), Some("gateway.test:8443")); + let endpoint = reconciled + .spec + .gateway_endpoint + .as_ref() + .unwrap_or_else(|| std::process::abort()); + assert_eq!(source, GatewayAddressSource::Available); + assert_eq!(endpoint.address, "gateway.test:8443"); + assert_eq!(endpoint.tls.server_name.as_deref(), Some("test-site.grid.internal")); + assert!(site.spec.gateway_endpoint.is_none(), "the API object remains unchanged"); + let (next, reason, _) = site_phase_for_gateway( + &GridSitePhase::Pending, + &reconciled, + None, + crate::cli::Membership::Local, + source, + ); + assert_eq!( + next, + GridSitePhase::Connecting, + "the override supplies the own gateway endpoint" + ); + assert_eq!(reason, "EndpointsDeclared"); + } + + /// Checks declared-endpoint precedence and keeps the local override off peer sites. + #[test] + fn declared_own_endpoint_wins_and_the_local_override_does_not_apply_to_peers() { + let own_site = site_with_egress(Some(GridSitePhase::Pending), "declared.test:8443"); + let (own_site, own_source) = + apply_local_gateway_address(&own_site, Some("test-site"), Some("gateway.test:9443")); + assert_eq!(own_source, GatewayAddressSource::Available); + assert_eq!( + own_site + .spec + .gateway_endpoint + .as_ref() + .map(|endpoint| endpoint.address.as_str()), + Some("declared.test:8443"), + "the declared endpoint takes precedence over the local setting" + ); + + let peer = site_no_egress(Some(GridSitePhase::Pending)); + let (peer, peer_source) = apply_local_gateway_address(&peer, Some("another-site"), Some("gateway.test:9443")); + assert_eq!(peer_source, GatewayAddressSource::Peer); + assert!( + peer.spec.gateway_endpoint.is_none(), + "the local override stays on this site's record" + ); } #[test] diff --git a/crates/operator/src/enroll/local/conformance.rs b/crates/operator/src/enroll/local/conformance.rs index 6c2fde46d..1879ffda5 100644 --- a/crates/operator/src/enroll/local/conformance.rs +++ b/crates/operator/src/enroll/local/conformance.rs @@ -513,6 +513,7 @@ async fn invariant_8_no_signer_reads_rotation_on_schedule_no_signer_and_never_bl assert_eq!(degraded["status"], "False"); } +/// Checks that `GridSite` reconciliation leaves the own-site peer rollup to its designated writer. #[tokio::test] async fn invariant_5_the_own_grid_site_carries_only_the_peer_rollup_and_the_network_one_writer_each() { let api = seeded(); @@ -526,11 +527,14 @@ async fn invariant_5_the_own_grid_site_carries_only_the_peer_rollup_and_the_netw .expect("networks list"); observed.push(rolled(&api, &client).await); let site: GridSite = serde_json::from_value(api.get(SITE_PATH).expect("own site stored")).expect("a GridSite"); + let (_gateway_address_tx, gateway_address) = tokio::sync::watch::channel(None); Box::pin(crate::controller::grid_site::reconcile( Arc::new(site), Arc::new(crate::controller::grid_site::SiteContext { client: Arc::new(client.clone()), membership: crate::cli::Membership::default(), + site_name: Some(SITE.to_owned()), + gateway_address, }), )) .await diff --git a/crates/operator/src/gateway.rs b/crates/operator/src/gateway.rs index 8690ff2ff..f081780c3 100644 --- a/crates/operator/src/gateway.rs +++ b/crates/operator/src/gateway.rs @@ -1,8 +1,8 @@ //! Site gateway address self-discovery. //! //! Resolves the data-plane gateway address this operator advertises to SWIM -//! peers (populates `GridSite.spec.gatewayEndpoint.address`): an explicit override wins, -//! else a background poller discovers the Service `LoadBalancer` address. +//! peers and uses for its own `GridSite`: an explicit override wins, else a +//! background poller discovers the Service `LoadBalancer` address. use std::{sync::Arc, time::Duration}; @@ -45,7 +45,7 @@ pub struct Config { )] pub discovery_enabled: bool, - /// Explicit gateway address (host:port); skips discovery when set. + /// Explicit gateway address (host:port); skips discovery and supplies the own site's endpoint. /// /// Blank means unset here, unlike the discovery fields. #[arg(long = "gateway-address", env = "GRID_GATEWAY_ADDRESS")] @@ -104,6 +104,12 @@ impl Config { } } +/// The configured override, trimmed and empty values treated as unset. +#[must_use] +pub fn configured_address(config: &Config) -> Option { + config.address_override().map(str::to_owned) +} + /// Resolve the gateway address: explicit override, else Service discovery. /// /// `Ok(None)` means no address yet. @@ -125,10 +131,15 @@ pub async fn resolve(client: &Client, config: &Config) -> Result, Ok(discovery.into_address()) } -/// Poll for the gateway Service address and re-announce it via SWIM. +/// Poll for the gateway Service address and publish it to controllers and SWIM. /// /// No-op when discovery is disabled or an explicit address override is set. -pub async fn run_discovery_poller(client: Client, swim: Arc, config: Config) { +pub async fn run_discovery_poller( + client: Client, + config: Config, + resolved_address: tokio::sync::watch::Sender>, + swim: tokio::sync::watch::Receiver>>, +) { if !config.discovery_enabled { tracing::info!("gateway address discovery disabled; skipping discovery poller"); return; @@ -140,29 +151,30 @@ pub async fn run_discovery_poller(client: Client, swim: Arc, config: let interval = config.discovery_interval(); let interval_ms = u64::try_from(interval.as_millis()).unwrap_or(u64::MAX); tracing::info!(interval_ms, "starting gateway address discovery poller"); - poll_loop(&client, &swim, interval, &config).await; + poll_loop(&client, &resolved_address, &swim, interval, &config).await; } /// Inner polling loop; separated to satisfy clippy complexity lints. -async fn poll_loop(client: &Client, swim: &SwimHandle, interval: Duration, config: &Config) -> ! { +async fn poll_loop( + client: &Client, + resolved_address: &tokio::sync::watch::Sender>, + swim: &tokio::sync::watch::Receiver>>, + interval: Duration, + config: &Config, +) -> ! { let mut last: Option = None; loop { - tokio::time::sleep(interval).await; match discover_from_service(client, config).await { Ok(discovery) => { log_discovery(&discovery, config, is_new(last.as_ref(), &discovery)); - // Re-announce even if unchanged: a peer may have joined since. - if let Discovery::Found(addr) = &discovery - && let Err(e) = swim.set_gateway_address(Some(addr.clone())) - { - tracing::warn!(error = %e, "failed to update gateway address on SWIM handle"); - } + publish_address(resolved_address, swim, discovery.address()); last = Some(discovery); }, Err(e) => { tracing::warn!(error = %e, "gateway discovery poll failed; will retry"); }, } + tokio::time::sleep(interval).await; } } @@ -180,6 +192,14 @@ enum Discovery { } impl Discovery { + /// The discovered address, if any, without consuming the outcome. + fn address(&self) -> Option<&str> { + match self { + Self::Found(addr) => Some(addr), + Self::NoAddress | Self::NotLoadBalancer(_) | Self::NoService => None, + } + } + /// The discovered address, if any. fn into_address(self) -> Option { match self { @@ -189,6 +209,34 @@ impl Discovery { } } +/// Publish a discovery result to controllers and any active SWIM runtime. +/// +/// Missing discovery results preserve the last published address. This avoids +/// flapping during Service recreation and keeps the wire format compatible +/// with peers that treat an empty address as a value. +fn publish_address( + resolved_address: &tokio::sync::watch::Sender>, + swim: &tokio::sync::watch::Receiver>>, + address: Option<&str>, +) { + let Some(address) = address else { + return; + }; + resolved_address.send_if_modified(|current| { + if current.as_deref() == Some(address) { + false + } else { + *current = Some(address.to_owned()); + true + } + }); + if let Some(handle) = swim.borrow().as_ref() + && let Err(error) = handle.set_gateway_address(Some(address.to_owned())) + { + tracing::warn!(%error, "failed to update gateway address on SWIM handle"); + } +} + /// Look up the gateway Service and extract its `LoadBalancer` address. async fn discover_from_service(client: &Client, config: &Config) -> Result { let api: Api = Api::namespaced(client.clone(), &config.namespace); @@ -334,6 +382,70 @@ mod tests { Service::default() } + /// Checks that repeating an address does not notify watch receivers. + #[test] + fn publishing_an_unchanged_address_does_not_notify_watchers() { + let (resolved_address, mut address_rx) = tokio::sync::watch::channel(None); + let (_swim_tx, swim_rx) = tokio::sync::watch::channel(None); + + publish_address(&resolved_address, &swim_rx, Some("gateway.test:8080")); + assert!(address_rx.has_changed().is_ok_and(|changed| changed)); + drop(address_rx.borrow_and_update()); + + publish_address(&resolved_address, &swim_rx, Some("gateway.test:8080")); + + assert!(address_rx.has_changed().is_ok_and(|changed| !changed)); + } + + /// Checks that a later poll applies an address after the SWIM handle appears. + #[test] + fn publishing_updates_a_swim_handle_that_appears_afterward() { + let (resolved_address, _address_rx) = tokio::sync::watch::channel(None); + let (swim_tx, swim_rx) = tokio::sync::watch::channel(None); + let (handle, _gateway_rx) = crate::swim_runtime::test_handle(); + let handle = Arc::new(handle); + + publish_address(&resolved_address, &swim_rx, Some("gateway.test:8080")); + assert_eq!( + *resolved_address.borrow(), + Some("gateway.test:8080".to_owned()), + "the first publication reaches controller watchers" + ); + + assert!(swim_tx.send(Some(Arc::clone(&handle))).is_ok()); + publish_address(&resolved_address, &swim_rx, Some("gateway.test:8080")); + + assert_eq!( + handle.gateway_address().as_deref(), + Some("gateway.test:8080"), + "a later poll applies the address to the newly available SWIM handle" + ); + } + + /// Checks that a missing lookup preserves the last address held by SWIM and controllers. + #[test] + fn publishing_a_missing_lookup_preserves_the_last_address() { + let (resolved_address, _address_rx) = tokio::sync::watch::channel(None); + let (handle, _gateway_rx) = crate::swim_runtime::test_handle(); + let handle = Arc::new(handle); + assert!(handle.set_gateway_address(Some("gateway.test:8080".to_owned())).is_ok()); + let (_swim_tx, swim_rx) = tokio::sync::watch::channel(Some(Arc::clone(&handle))); + + publish_address(&resolved_address, &swim_rx, Some("gateway.test:8080")); + publish_address(&resolved_address, &swim_rx, None); + + assert_eq!( + handle.gateway_address().as_deref(), + Some("gateway.test:8080"), + "an absent Service lookup must not withdraw the address from SWIM" + ); + assert_eq!( + resolved_address.borrow().as_deref(), + Some("gateway.test:8080"), + "an absent Service lookup must not withdraw the address from controllers" + ); + } + #[test] fn a_service_without_a_load_balancer_type_is_its_own_outcome() { let typed = |kind: &str, svc: Service| Service { diff --git a/crates/operator/src/main.rs b/crates/operator/src/main.rs index 8c02b0701..d85c8d81c 100644 --- a/crates/operator/src/main.rs +++ b/crates/operator/src/main.rs @@ -87,6 +87,7 @@ use operator::{ // Main // --------------------------------------------------------------------------- +/// Starts the operator and its controllers. #[tokio::main] #[expect( clippy::large_stack_frames, @@ -318,7 +319,7 @@ async fn main() { // This site's providers first, then what its peers publish, on /metrics. operator::metrics::register_provider_signals(vec![ctx.signals(), ctx.peers()]); - // Controllers run now. Only SWIM and what dials peers wait on the advertise address. + // Controllers run now; the GridSite controller follows gateway address updates as they arrive. let (swim_tx, swim_rx) = tokio::sync::watch::channel(SwimStage::Starting); route_to_holder(&routes, signals_enabled.then(|| ctx.signals()), &leadership.holding()); @@ -333,10 +334,26 @@ async fn main() { let writer_site_name = ctx.site_name().map(str::to_owned); // Every mode scrapes for readiness; only Poll mode publishes what it scrapes. let writer_scrape_interval = config.signals.scrape_interval(); + let (gateway_address_tx, gateway_address_rx) = + tokio::sync::watch::channel(gateway::configured_address(&config.gateway)); + let (swim_handle_tx, swim_handle_rx) = tokio::sync::watch::channel(None); + if writer_site_name.is_some() { + tokio::spawn(gateway::run_discovery_poller( + client.clone(), + config.gateway.clone(), + gateway_address_tx, + swim_handle_rx, + )); + } let writer_group = async move { tokio::try_join!( run_network_controller(writer_client.clone(), Arc::clone(&writer_ctx), writer_swim), - run_site_controller(writer_client.clone(), writer_ctx.membership()), + run_site_controller( + writer_client.clone(), + writer_ctx.membership(), + writer_site_name.clone(), + gateway_address_rx.clone(), + ), run_provider_controller(writer_client.clone(), writer_site_name), run_agent_tool_provider_controller(writer_client.clone()), run_local_scraper(writer_scrape_interval, Arc::clone(&writer_ctx), writer_client.clone()), @@ -368,7 +385,7 @@ async fn main() { ready: Arc::clone(&ready), routes: routes.clone(), }, - swim_tx, + (swim_tx, swim_handle_tx), membership_declared, ), writers, @@ -582,6 +599,12 @@ enum SwimStage { /// SWIM startup progress, watched by what waits on the advertise address. type SwimStartup = tokio::sync::watch::Receiver; +/// Senders that publish SWIM lifecycle progress and the running handle. +type SwimStartupSenders = ( + tokio::sync::watch::Sender, + tokio::sync::watch::Sender>>, +); + /// Start SWIM off the controllers' path, marking ready once it runs or is not configured. /// /// # Errors @@ -591,9 +614,10 @@ async fn start_swim( cli: Cli, ctx: Arc, restart: StandDown, - started: tokio::sync::watch::Sender, + startup: SwimStartupSenders, membership_declared: bool, ) -> Result<(), Box> { + let (started, swim_handle) = startup; let client = ctx.client.clone(); let retry = restart.clone(); let ready = Arc::clone(&restart.ready); @@ -608,11 +632,7 @@ async fn start_swim( let swim = report_swim(&client, cli.grid.membership, started_swim, retry)?; if let Some(handle) = &swim { ctx.set_swim(Arc::clone(handle)); - tokio::spawn(gateway::run_discovery_poller( - client.clone(), - Arc::clone(handle), - cli.gateway.clone(), - )); + swim_handle.send_replace(Some(Arc::clone(handle))); tokio::spawn(release_membership_once_settled(Arc::clone(handle), Arc::clone(&ctx))); } else { settle_without_swim(&client, cli.grid.membership, &ctx).await?; @@ -1879,11 +1899,18 @@ fn watch_expired(error: &kube::runtime::controller::Error) async fn run_site_controller( client: Client, membership: Membership, + site_name: Option, + gateway_address: tokio::sync::watch::Receiver>, ) -> Result<(), Box> { let api = Api::::all(client.clone()); + let address_changes = futures::stream::unfold(gateway_address.clone(), |mut address| async move { + address.changed().await.ok()?; + Some(((), address)) + }); // Unfiltered: a status write from the peer poller reconciles, so Degraded follows it. Controller::new(api, watcher::Config::default()) + .reconcile_all_on(address_changes) .with_config(kube::runtime::controller::Config::default().concurrency(16)) .run( grid_site::reconcile, @@ -1891,6 +1918,8 @@ async fn run_site_controller( Arc::new(grid_site::SiteContext { client: Arc::new(client), membership, + site_name, + gateway_address, }), ) .for_each(|result| async { diff --git a/crates/operator/src/swim_runtime.rs b/crates/operator/src/swim_runtime.rs index 30a840d74..984d71213 100644 --- a/crates/operator/src/swim_runtime.rs +++ b/crates/operator/src/swim_runtime.rs @@ -1424,6 +1424,37 @@ impl SwimHandle { } } +/// Create a test handle with a live gateway-address receiver. +#[cfg(test)] +pub(crate) fn test_handle() -> (SwimHandle, watch::Receiver>) { + let (snapshot_tx, snapshot_rx) = watch::channel(MembershipSnapshot::default()); + let (state_tx, state_rx) = watch::channel(GridStateSnapshot::new("test".to_owned())); + let (broadcast_tx, _broadcast_rx) = mpsc::channel(1); + let (seed_tx, _seed_rx) = mpsc::channel(16); + let (key_tx, _key_rx) = watch::channel(KeyState::Plain); + let (gateway_tx, gateway_rx) = watch::channel(None); + let handle = SwimHandle { + publisher_reservation: (1, 2), + site_name: "test".to_owned(), + advertise_addr: "127.0.0.1:7946".parse().unwrap_or_else(|_| std::process::abort()), + signals_address: None, + snapshot_rx, + state_rx, + model_inventories_rx: watch::channel(Vec::new()).1, + broadcast_tx, + seed_tx, + key_tx, + trust_update_tx: mpsc::unbounded_channel().0, + inventory_broadcast_tx: mpsc::channel(1).0, + gateway_tx, + runtime_tx: watch::channel(true).0, + withdraw_tx: mpsc::channel(16).0, + leave_tx: mpsc::channel(1).0, + }; + drop((snapshot_tx, state_tx)); + (handle, gateway_rx) +} + /// Stable SWIM data that can change rendered Grid resources. #[derive(Debug, PartialEq)] struct ReconciliationView { @@ -2064,8 +2095,8 @@ async fn run_loop( ) .await; publish_state(&state_tx, &node); - // Discovery re-announces an unchanged address for late joiners; only a change is news. if previous.as_deref() == Some(addr) { + // Discovery re-announces an unchanged address for late joiners; only a change is news. tracing::debug!(addr = %addr, "gateway address re-announced"); } else { tracing::info!(addr = %addr, "gateway address updated at runtime"); diff --git a/crates/operator/tests/status_conformance/peer_site.rs b/crates/operator/tests/status_conformance/peer_site.rs index 8a546ebaf..d65fd7869 100644 --- a/crates/operator/tests/status_conformance/peer_site.rs +++ b/crates/operator/tests/status_conformance/peer_site.rs @@ -29,9 +29,12 @@ fn seeded() -> (FakeApi, String) { /// One reconcile of the site at `path` under `membership`. async fn reconcile_under(api: &FakeApi, path: &str, membership: Membership) { let site: GridSite = fixtures::stored(api, path); + let (_gateway_address_tx, gateway_address) = tokio::sync::watch::channel(None); let ctx = Arc::new(grid_site::SiteContext { client: Arc::new(api.client("default")), membership, + site_name: None, + gateway_address, }); Box::pin(grid_site::reconcile(Arc::new(site), ctx)) .await diff --git a/crates/swim/src/state_broadcast.rs b/crates/swim/src/state_broadcast.rs index e4374976f..7e6a4d615 100644 --- a/crates/swim/src/state_broadcast.rs +++ b/crates/swim/src/state_broadcast.rs @@ -119,7 +119,7 @@ pub struct StateBroadcast { /// Data-plane gateway address advertised by this site. /// - /// Carried as optional trailing extension data. `None` when the originating + /// Carried as optional trailing extension data. `None` when the originating /// operator has no configured gateway address. #[serde(skip_serializing_if = "Option::is_none")] pub gateway_address: Option, @@ -4562,6 +4562,7 @@ mod tests { ); } + /// Checks that an address broadcast is retained for a remote site. #[test] fn handler_stores_gateway_address_from_extended_broadcast() { let mut handler = StateBroadcastHandler::new("site-local".to_owned()); diff --git a/docs/architecture/crds.md b/docs/architecture/crds.md index 2c6262786..8dc8eed33 100644 --- a/docs/architecture/crds.md +++ b/docs/architecture/crds.md @@ -790,7 +790,7 @@ A discovered SWIM peer is not automatically authorized for routing. |---|---|---| | `Pending` | Resource created (manually or by auto-discovery) | Initial default | | `Discovered` | SWIM peer observed as Alive | `GridNetwork` controller writes on first observation | -| `Connecting` | Gateway address known (`spec.gatewayEndpoint.address` non-empty) | `GridSite` controller advances from Discovered, or from Pending for a declared site with both endpoints; performs identity-aware probe | +| `Connecting` | Gateway address known (`spec.gatewayEndpoint.address` non-empty, or the own site's resolved operator address) | `GridSite` controller advances from Discovered, or from Pending for a declared site with both endpoints; performs identity-aware probe | | `Active` | `TlsVerified` | `GridSite` controller promotes from Connecting only after identity-verified TLS succeeds | | `Unreachable` | Connectivity failure while Active | `GridSite` controller moves Active → Unreachable when the endpoint cannot be reached | | `Left` | Set on graceful site departure | Preserved by operator once set | @@ -803,7 +803,7 @@ A discovered SWIM peer is not automatically authorized for routing. | `SWIMDiscovered` | Discovered | Peer observed as Alive in SWIM membership; gateway address propagating | | `GatewayAddressKnown` | Connecting | Gateway address received; advancing to Connecting | | `EndpointsDeclared` | Connecting | Declared site names a dialable gateway endpoint, which also gives its signals endpoint; advancing without discovery | -| `GatewayAddressMissing` | Discovered, Connecting, or Unreachable | No gateway address known (see `GRID_GATEWAY_ADDRESS`), or a previously probed site lost it | +| `GatewayAddressMissing` | Pending, Discovered, Connecting, or Unreachable | No usable gateway address is known: the own site's endpoint and configured address are absent, or a peer has not announced an address | | `TlsVerified` | Active | TLS handshake succeeded; certificate chain, identity, and configured pin verified | | `IdentityVerificationRequired` | Connecting | TCP endpoint is reachable, but plaintext cannot establish the gateway identity | | `PlaintextUnreachable` | Connecting or Unreachable | TCP probe failed (explicit Plaintext mode) | @@ -824,9 +824,12 @@ A discovered SWIM peer is not automatically authorized for routing. label on the `GridNetwork`). - Pending → Connecting: the `GridSite` controller advances a declared site (one without the `grid.praxis.fast/auto-discovered` label) that sets a dialable - `spec.gatewayEndpoint.address`, so it does not wait on SWIM. + `spec.gatewayEndpoint.address`, so it does not wait on SWIM. For this operator's own + site, the resolved `gateway.address` or LoadBalancer address supplies the endpoint when + the spec omits it; the operator does not write that address to the spec. - Discovered → Connecting: the `GridSite` controller advances automatically when - `spec.gatewayEndpoint.address` is non-empty. For auto-discovered sites, the egress address comes from + `spec.gatewayEndpoint.address` is non-empty. The own site can also use its resolved + operator address without changing the spec. For auto-discovered sites, the egress address comes from the remote operator's `GRID_GATEWAY_ADDRESS` env var, propagated via SWIM state broadcast. If the remote operator has not configured `GRID_GATEWAY_ADDRESS`, the egress address is empty and the site stays Discovered with reason `GatewayAddressMissing`. @@ -853,7 +856,9 @@ still counts the transitions by phase and reason. the remote operator's `GRID_GATEWAY_ADDRESS` environment variable, propagated through the SWIM state broadcast. If the remote operator has not configured `GRID_GATEWAY_ADDRESS`, the field is empty and the site stays Discovered. For manually-applied `GridSite` resources, set -`spec.gatewayEndpoint.address` explicitly to the data-plane gateway endpoint. +`spec.gatewayEndpoint.address` explicitly to the data-plane gateway endpoint. The operator's +own `GridSite` uses its configured or discovered local gateway address at reconcile time when +the spec omits it, without patching the spec. **`status.publicCertPem`:** The public site certificate PEM received from the remote site via SWIM state broadcast. Before storage, the operator performs a structural check: @@ -930,9 +935,13 @@ time by provider gateway filters. **Single-site / combined-site deployments:** Site discovery is driven by *remote* SWIM peers - the `Pending -> Discovered` transition above requires a remote peer observed Alive. A single or combined cluster has no peers, so its own `GridSite` -stays `Pending` with reason `AwaitingDiscovery`. **This is expected and does not -block local serving:** local `InferenceProvider`s are eligible regardless of -`GridSite.status.phase`; only *remote* CRDT provider records are phase-gated (see +stays `Pending` with reason `AwaitingDiscovery` when a usable gateway address is +available. If its endpoint is undeclared and gateway discovery has no address, +reconciliation uses `GatewayAddressMissing` under both Swim and Local +membership; this reason is emitted as a `Warning` event. **Either reason is +expected and does not block local serving:** local `InferenceProvider`s are +eligible regardless of `GridSite.status.phase`; only *remote* CRDT provider +records are phase-gated (see "Routing eligibility" above). Do not add SWIM seeds or extra operator replicas to try to force the site `Active` - there is no second site to discover, and a lone operator legitimately runs a single-node mesh with zero peers. The `Active` phase diff --git a/docs/architecture/operations.md b/docs/architecture/operations.md index 03302f245..7c3e0281c 100644 --- a/docs/architecture/operations.md +++ b/docs/architecture/operations.md @@ -1077,7 +1077,9 @@ auto-discovered sites. poller retries every 5 seconds (configurable via `GRID_GATEWAY_DISCOVERY_INTERVAL_MS`) until the address appears, then continues watching for changes. Discovered addresses are pushed to the SWIM -runtime via a watch channel. +runtime and the `GridSite` controller via watch channels. The own `GridSite` +uses this address when its spec has no `gatewayEndpoint`; the operator leaves +the spec unchanged. **Explicit override:** Set `GRID_GATEWAY_ADDRESS` to skip the self-discovery poller entirely. @@ -1090,8 +1092,9 @@ still wins and is advertised. Discovery uses the first LoadBalancer ingress hostname or IP. The advertised port is `GRID_GATEWAY_PORT` when set, otherwise the first Service `spec.ports` entry, with `8080` as the fallback if no usable port exists. A ClusterIP or -NodePort Service supplies no discovered address; use an explicit reachable -address for those Service types. +NodePort Service supplies no discovered address; set `GRID_GATEWAY_ADDRESS` +to a reachable `host:port`. This override is also used by the own `GridSite` +when the operator runs without SWIM. ```bash # Self-discovery (default): operator discovers from provider-gateway Service @@ -1184,6 +1187,12 @@ separate steps. will propagate the address through SWIM once discovered. - Reason will be `GatewayAddressMissing`. +**The own GridSite has no gateway address** + +- Reason will be `GatewayAddressMissing`. Set `gateway.address` in the + operator configuration, or declare `spec.gatewayEndpoint.address` on the + own `GridSite`. + **Phase stays Connecting** - Check `status.reason`: diff --git a/scripts/e2e-hub-site.sh b/scripts/e2e-hub-site.sh index 0a107196c..4d1272f53 100755 --- a/scripts/e2e-hub-site.sh +++ b/scripts/e2e-hub-site.sh @@ -152,10 +152,13 @@ cmd_up() { fi resolve_forge CREATED=1 + # `up` creates the cross-cluster network as well as the configured Kind clusters. + # Stacks remain separate so the amd64 VCR image can be loaded before the site stack applies. # Word splitting is the point: the prefix is a command and its flags. # shellcheck disable=SC2086 ${KIND_CREATE_PREFIX:-} "$FORGE_BIN" --config "$TOPO/forge.yaml" --state-dir "$FORGE_STATE" \ - --non-interactive up + --non-interactive up || die "create the environment network and clusters" + preload_model for c in hub site; do for img in operator gateway enrollment; do forge cluster load-image "$c" "$(image "$img")" @@ -165,6 +168,18 @@ cmd_up() { feedback_contract_test } +# preload_model: the upstream VCR image is amd64-only. Pull and load that image on non-amd64 +# hosts before Forge applies the model workload; all other images keep their native platform. +preload_model() { + local img + [[ $(uname -m) != x86_64 ]] || return 0 + img=$(sed -n 's/^ *vcrImage: "\(.*\)"$/\1/p' "$TOPO/forge.yaml") + [[ -n $img ]] || die "no vcrImage in $TOPO/forge.yaml" + log "pulling amd64 model backend image $img" + "$CONTAINER_TOOL" pull --platform=linux/amd64 "$img" || die "pull amd64 model backend image" + forge cluster load-image site "$img" || die "load amd64 model backend image into site cluster" +} + # The CI-only test owns exactly one object and the CRD it installs on the empty hub. feedback_contract_test() ( [[ ${GRID_FEEDBACK_CONTRACT_TEST:-0} == 1 ]] || return 0 @@ -280,6 +295,9 @@ renewal_args() { fi ! local_source || RA+=(--set grid.membership.source=Local) ! local_source || [[ $1 != grid-operator ]] || RA+=(--set signals.service.type=LoadBalancer) + ! local_source || [[ $1 != grid-operator ]] || RA+=( + --set gateway.serviceName=grid-gateway-internal + --set "gateway.address=grid-gateway-internal.$NS.svc:8080") polls || return 0 case $1 in # grid.signals starts the operator in poll before its GridNetwork exists; without grid.id @@ -381,6 +399,34 @@ network_condition() { # other writer left cannot pass as enrolled. site_reports_enrolled() { [[ $(network_condition "$1" Enrolled) == True/Enrolled ]]; } +own_site_connected() { [[ $(k "$1" get gridsite "$2" \ + -o jsonpath='{.status.conditions[?(@.type=="Connected")].status}/{.status.conditions[?(@.type=="Connected")].reason}') == True/Connected ]]; } + +own_site_gateway_not_in_spec() { + local address + address=$(k "$1" get gridsite "$2" -o jsonpath='{.spec.gatewayEndpoint.address}') || return 1 + [[ -z $address ]] +} + +cluster_ip_gateway() { + kubectl --context "$1" -n "$NS" apply -f - >/dev/null <