From 831df2b53b27e60ca501857671f17fc49939dcd2 Mon Sep 17 00:00:00 2001 From: ejc3 Date: Sun, 25 Oct 2026 12:03:28 +0000 Subject: [PATCH 1/4] fleet: serve the EC2 half's aggregator to browsers under agg-ec2 names The EC2 aggregator runs beside the coordinator in its Fargate task and was reachable only at --ec2-aggregate-url, through the coordinator's Cloudflare tunnel (Access-gated, and Cloudflare in the media path). `fcvm fleet coord --ec2-aggregate-tls` (with --ec2-aggregate-listen, --tls-secret and --dns-token-secret) gives it the task's own addresses: - src/fleet/coord/ecs.rs reads them at start: the global IPv6 address from the ECS task metadata, and the public IPv4 address of the task's network interface from EC2 DescribeNetworkInterfaces (filtered by its private address; the metadata does not carry it). A task with neither is an error. - The coordinator writes the certificate from --tls-secret as tls.json into --ec2-aggregate-tls-dir (the agent's tls::refresh/run, hourly) and runs the aggregator with --tls-listen on 0.0.0.0:443 and []:443. - Config carries the two addresses; dns::facts adds `agg-ec2` (generation 0, both on 443), owned() recognises agg-ec2-[-v4], and the published names are kept in Fleet::ec2_aggregate_published (cleared when a coordinator resumes, since they were for the previous task's addresses). - view: with the DNS writer the agg:ec2 entry's frame_url/frame_url_v4 are its published names, else null; without it, --ec2-aggregate-url as before. - --ec2-aggregate-listen now needs --ec2-aggregate-url or --ec2-aggregate-tls. The task must admit 443 in its security group and lower net.ipv4.ip_unprivileged_port_start (the image runs as 65534); that is the Terraform side. Staging is unchanged until it passes the flag. Tests: the_ec2_aggregator_is_published_at_the_coordinators_addresses (tests/test_fleet_coord.rs) waits for the AAAA and A records, sees no URL before confirmation, then wss://agg-ec2-./frames and the -v4 name with no port. Without the agg-ec2 fact it fails: no record is ever written. the_task_addresses_are_read_from_metadata_and_ec2 covers dual stack, IPv4 only, a ULA address and the EC2 answer. --- fleet/PROTOCOL.md | 15 +++- fleet/core/src/apply.rs | 11 +++ fleet/core/src/state.rs | 17 +++++ fleet/core/src/view.rs | 34 +++++++-- src/cli/args.rs | 17 ++++- src/fleet/coord/dns.rs | 29 ++++++-- src/fleet/coord/ecs.rs | 141 ++++++++++++++++++++++++++++++++++++++ src/fleet/coord/mod.rs | 49 +++++++++++-- tests/test_fleet_coord.rs | 70 ++++++++++++++++++- 9 files changed, 362 insertions(+), 21 deletions(-) create mode 100644 src/fleet/coord/ecs.rs diff --git a/fleet/PROTOCOL.md b/fleet/PROTOCOL.md index ea6c1cb72..a82d06a9b 100644 --- a/fleet/PROTOCOL.md +++ b/fleet/PROTOCOL.md @@ -414,7 +414,8 @@ relaunching (its region would be black; claimed GPUs and dark tiles do not count fcvm fleet coord --listen [--agent-cmd '']... [--state-file ] [--text-check bedrock --text-check-model --text-check-region | --text-check none] [--max-tiles 100] [--ec2 none|fake] [--allowed-origin ]... - [--ec2-aggregate-listen --ec2-aggregate-url ] [--aggregate-arg ]... + [--ec2-aggregate-listen (--ec2-aggregate-url | --ec2-aggregate-tls + [--ec2-aggregate-tls-dir /tmp/fcvm-fleet/tls])] [--aggregate-arg ]... [--race-every ] [--restart-cooldown 60] [--camera-seed 1] [--max-borrowed ] [--trust-cf-connecting-ip] [--turnstile-secret --turnstile-site-key ] [--admin-listen ] [--ffmpeg ffmpeg] @@ -467,6 +468,18 @@ fcvm fleet coord --listen [--agent-cmd '']... [--state-file IPv6 address, A the host's IPv4 address, with its port in `frame_url_v4`. Its names last while it reports the same addresses. - `--tls-secret` keeps the wildcard certificate renewed (see "Certificate"). +- `--ec2-aggregate-tls` (with `--ec2-aggregate-listen`, `--tls-secret` and + `--dns-token-secret`) serves the EC2 half's aggregator to browsers with TLS on port 443 + of the coordinator task's own addresses, read at start on ECS: the global IPv6 address + from the task metadata (`ECS_CONTAINER_METADATA_URI_V4`) and the public IPv4 address of + the task's network interface (EC2 `DescribeNetworkInterfaces` by its private address, in + `AWS_REGION`, with the task role). It writes the certificate (from `--tls-secret`) as + `tls.json` into `--ec2-aggregate-tls-dir` for the aggregator and reads the secret again + hourly, and names the addresses `agg-ec2-` (AAAA and `-v4` A, both on 443) like a + host aggregator. A non-root process binds 443 only where the task lowers + `net.ipv4.ip_unprivileged_port_start` (it covers IPv6 too). With published names the + EC2 aggregate's `frame_url`/`frame_url_v4` are those names, else null; + `--ec2-aggregate-url` is used only without the DNS writer. ### Certificate diff --git a/fleet/core/src/apply.rs b/fleet/core/src/apply.rs index 4efeb5c3c..9330037d7 100644 --- a/fleet/core/src/apply.rs +++ b/fleet/core/src/apply.rs @@ -196,6 +196,17 @@ fn apply_inner(state: &mut State, event: &Event) -> Option { None } Event::Published { slot, names, .. } => { + let c = &state.config; + if slot == crate::state::EC2_AGGREGATE_ID { + if names.names( + c.ec2_aggregate_ipv6.as_deref(), + c.ec2_aggregate_ipv4.as_deref(), + None, + ) { + state.fleet.ec2_aggregate_published = Some(names.clone()); + } + return None; + } let f = &mut state.fleet; // A host aggregator's names, for the addresses it reports now. if let Some(h) = f diff --git a/fleet/core/src/state.rs b/fleet/core/src/state.rs index c02f25cf2..171f2013d 100644 --- a/fleet/core/src/state.rs +++ b/fleet/core/src/state.rs @@ -27,6 +27,12 @@ pub struct Config { /// The EC2 half's aggregator frame WebSocket, when the coordinator runs one. #[serde(default)] pub ec2_aggregate_url: Option, + /// The EC2 half's aggregator's public addresses when it serves browsers with TLS (on + /// 443 in both families): the coordinator task's own. Named `agg-ec2-`. + #[serde(default)] + pub ec2_aggregate_ipv6: Option, + #[serde(default)] + pub ec2_aggregate_ipv4: Option, /// The owner's race cadence: a relaunch of every tile this often. `None`: no schedule. pub race_interval_secs: Option, /// After any race, `restart_now` is refused for this long, for everyone. @@ -131,6 +137,8 @@ impl Default for Config { light_interval_ms: 1_000, relaunch_interval_ms: 2_000, ec2_aggregate_url: None, + ec2_aggregate_ipv6: None, + ec2_aggregate_ipv4: None, race_interval_secs: None, restart_cooldown_secs: 60, camera_seed: 1, @@ -515,6 +523,9 @@ pub struct Fleet { pub budget: crate::budget::Budget, /// Coordinator components that are degraded: reason and since when. pub degraded: BTreeMap, + /// The EC2 half's aggregator's published DNS names (`EC2_AGGREGATE_ID`). + #[serde(default)] + pub ec2_aggregate_published: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -563,6 +574,7 @@ impl State { warming: Default::default(), power: Power::default(), budget: Default::default(), + ec2_aggregate_published: None, }, config, receipts: BTreeMap::new(), @@ -586,6 +598,8 @@ impl State { f.viewers.clear(); f.hidden.clear(); f.degraded.clear(); + // The EC2 aggregator's names were for the previous coordinator's addresses. + f.ec2_aggregate_published = None; // A race interrupted by the restart cannot finish (its slots come back at // generation 1), on either half. let finished = |h: &HalfRace| h.done_at.is_some() || h.tiles.is_empty(); @@ -654,6 +668,9 @@ pub fn aggregate_id(host: &str) -> Option { ok.then(|| format!("agg-{host}")) } +/// The id the EC2 half's aggregator is named under in DNS. +pub const EC2_AGGREGATE_ID: &str = "agg-ec2"; + /// An address in its canonical text (`2001:db8::2` for `2001:0db8::2`); text that is not an /// address is kept as it is. pub fn canonical_ip(a: &str) -> String { diff --git a/fleet/core/src/view.rs b/fleet/core/src/view.rs index b859f6f6d..9e30b8df7 100644 --- a/fleet/core/src/view.rs +++ b/fleet/core/src/view.rs @@ -276,7 +276,7 @@ pub struct AggregateView { /// With published names (the fleet's DNS), those of the addresses it serves now, once /// published: `wss://agg--./frames` and /// `wss://agg---v4.:/frames`. Otherwise the URL it was reported - /// at, with no IPv4 one. + /// at, with no IPv4 one. The EC2 half's are `agg-ec2-`, both on 443. pub frame_url: Option, pub frame_url_v4: Option, /// The tiles inside the region, by tile id. @@ -499,14 +499,18 @@ pub fn view(state: &State) -> View { }) .collect(); aggregates.sort_by(|a, b| a.rect.y.total_cmp(&b.rect.y)); - if let (Some(url), Some(rect)) = (&c.ec2_aggregate_url, f.layout.ec2) { + let ec2_aggregator = c.ec2_aggregate_url.is_some() + || c.ec2_aggregate_ipv6.is_some() + || c.ec2_aggregate_ipv4.is_some(); + if let (true, Some(rect)) = (ec2_aggregator, f.layout.ec2) { + let (frame_url, frame_url_v4) = ec2_aggregate_urls(state); aggregates.push(AggregateView { id: "agg:ec2".into(), half: Half::Ec2, host: None, rect, - frame_url: Some(url.clone()), - frame_url_v4: None, + frame_url, + frame_url_v4, tiles: (1..=f.ec2_n).map(|k| format!("ec2:{k}")).collect(), }); } @@ -858,6 +862,28 @@ pub fn host_aggregate_urls(state: &State, h: &Host) -> (Option, Option (Option, Option) { + let c = &state.config; + if !c.names_published { + return (c.ec2_aggregate_url.clone(), None); + } + match &state.fleet.ec2_aggregate_published { + Some(p) + if p.names( + c.ec2_aggregate_ipv6.as_deref(), + c.ec2_aggregate_ipv4.as_deref(), + None, + ) => + { + (p.frame_url.clone(), p.frame_url_v4.clone()) + } + _ => (None, None), + } +} + /// A tile's frame URLs (IPv6, IPv4). With published names, those of the launch the slot is /// on, once published; otherwise the URL its agent reported, with no IPv4 one. pub fn frame_urls(state: &State, s: &Slot) -> (Option, Option) { diff --git a/src/cli/args.rs b/src/cli/args.rs index 18dd3c24a..b32fb3d39 100644 --- a/src/cli/args.rs +++ b/src/cli/args.rs @@ -194,14 +194,27 @@ pub struct FleetCoordArgs { #[arg(long = "allowed-origin")] pub allowed_origins: Vec, - /// Run the EC2 half's aggregator on this address - #[arg(long, requires = "ec2_aggregate_url")] + /// Run the EC2 half's aggregator on this address (its layout, stats and a stream for + /// the coordinator); browsers reach it at --ec2-aggregate-url, or with + /// --ec2-aggregate-tls at its published names + #[arg(long)] pub ec2_aggregate_listen: Option, /// The EC2 half's aggregator frame WebSocket as browsers reach it #[arg(long, requires = "ec2_aggregate_listen")] pub ec2_aggregate_url: Option, + /// Serve the EC2 half's aggregator to browsers with TLS (the --tls-secret certificate) + /// on port 443 of the coordinator task's own addresses, read on ECS (the task metadata + /// for IPv6, EC2 DescribeNetworkInterfaces for the public IPv4 address), and name them + /// `agg-ec2-` (needs --dns-token-secret) + #[arg(long, requires_all = ["ec2_aggregate_listen", "tls_secret", "dns_token_secret"])] + pub ec2_aggregate_tls: bool, + + /// Where the coordinator keeps the certificate file of --ec2-aggregate-tls + #[arg(long, default_value = "/tmp/fcvm-fleet/tls")] + pub ec2_aggregate_tls_dir: std::path::PathBuf, + /// Extra argument for the EC2 half's aggregator. Repeatable. #[arg(long = "aggregate-arg", allow_hyphen_values = true)] pub aggregate_args: Vec, diff --git a/src/fleet/coord/dns.rs b/src/fleet/coord/dns.rs index 8ff1f09f0..5af49b77c 100644 --- a/src/fleet/coord/dns.rs +++ b/src/fleet/coord/dns.rs @@ -10,6 +10,8 @@ //! A host's aggregator is named the same way under `agg--` //! (`fleet_core::state::aggregate_id`): its own IPv6 address (port 443) and the host's IPv4 //! address on the aggregator's port. Its label lasts while it reports the same addresses. +//! The EC2 half's aggregator, which runs beside the coordinator, is `agg-ec2-`: the +//! coordinator task's own addresses, both on 443. //! //! The order is: write the records, confirm them at the zone's authoritative servers, and //! only then tell the coordinator (`Event::Published`), which puts the frame URLs in the @@ -54,8 +56,8 @@ pub struct Fact { pub published: Option, } -/// The facts of every slot, tile and host aggregator in the fleet. An aggregator has one -/// launch (generation 0) for as long as it reports the same addresses. +/// The facts of every slot, tile and aggregator in the fleet. An aggregator has one launch +/// (generation 0) for as long as it has the same addresses. pub fn facts(state: &State) -> Vec { let f = &state.fleet; let units = f.slots.values().chain(f.ec2.values()).map(|s| Fact { @@ -76,7 +78,19 @@ pub fn facts(state: &State) -> Vec { published: h.aggregate_published.clone(), }) }); - units.chain(aggregators).collect() + let c = &state.config; + let ec2_aggregator = Fact { + id: fleet_core::state::EC2_AGGREGATE_ID.into(), + generation: 0, + ipv6: c.ec2_aggregate_ipv6.clone(), + ipv4: c.ec2_aggregate_ipv4.clone(), + v4_port: None, + published: f.ec2_aggregate_published.clone(), + }; + units + .chain(aggregators) + .chain(std::iter::once(ec2_aggregator)) + .collect() } // Addresses compare in their canonical text, so the record, the published names and an @@ -162,7 +176,7 @@ fn is_token(t: &str) -> bool { } /// Whether a name in the zone is one this module owns: `slot-` or `ec2-`, optionally -/// with a launch token, or `agg--` (an aggregator), each optionally with +/// with a launch token, or `agg--` / `agg-ec2-` (an aggregator), each optionally with /// `-v4`. Anything else in the zone is never touched. fn owned(name: &str, zone: &str) -> bool { let Some(label) = name.strip_suffix(&format!(".{zone}")) else { @@ -171,7 +185,10 @@ fn owned(name: &str, zone: &str) -> bool { let label = label.strip_suffix("-v4").unwrap_or(label); if let Some(rest) = label.strip_prefix("agg-") { return rest.rsplit_once('-').is_some_and(|(host, token)| { - fleet_core::state::aggregate_id(host).is_some() && is_token(token) + let id = format!("agg-{host}"); + (fleet_core::state::aggregate_id(host).as_ref() == Some(&id) + || id == fleet_core::state::EC2_AGGREGATE_ID) + && is_token(token) }); } let Some(rest) = label @@ -676,6 +693,8 @@ mod tests { "agg-i-0123456789abcdef0-abcdefgh.fcvm.dev", "agg-i-0123456789abcdef0-abcdefgh-v4.fcvm.dev", "agg-gpu-a-zzzz2345.fcvm.dev", + "agg-ec2-abcdefgh.fcvm.dev", + "agg-ec2-abcdefgh-v4.fcvm.dev", ] { assert!(owned(name, z), "{name}"); } diff --git a/src/fleet/coord/ecs.rs b/src/fleet/coord/ecs.rs new file mode 100644 index 000000000..4cb22b7d8 --- /dev/null +++ b/src/fleet/coord/ecs.rs @@ -0,0 +1,141 @@ +//! The coordinator task's own public addresses on ECS (Fargate, awsvpc), for the EC2 half's +//! aggregator that runs beside it: the IPv6 address from the task metadata endpoint +//! (`ECS_CONTAINER_METADATA_URI_V4`, which lists the task's addresses, IPv6 when its subnet +//! and the account's `dualStackIPv6` setting give one), and the public IPv4 address that +//! EC2 associated with the task's network interface (`DescribeNetworkInterfaces` by its +//! private address; the metadata does not carry it). + +use super::aws::{signed, xml_tag, CredentialChain}; +use anyhow::{bail, Context, Result}; +use std::time::Duration; + +/// The public addresses: (IPv6, IPv4). +#[derive(Debug, Clone, PartialEq)] +pub struct TaskAddresses { + pub ipv6: Option, + pub ipv4: Option, +} + +/// The task's private IPv4 and global IPv6 addresses from a container metadata answer +/// (`GET ${ECS_CONTAINER_METADATA_URI_V4}`). +pub fn parse_metadata(body: &str) -> Result<(Option, Option)> { + let v: serde_json::Value = + serde_json::from_str(body).context("the task metadata is not JSON")?; + let net = v["Networks"] + .as_array() + .and_then(|n| n.first()) + .context("the task metadata lists no network")?; + let first = |k: &str| { + net[k] + .as_array() + .and_then(|a| a.first()) + .and_then(|a| a.as_str()) + .map(str::to_string) + }; + let ipv6 = first("IPv6Addresses").filter(|a| { + a.parse::() + .is_ok_and(|ip| (ip.segments()[0] & 0xe000) == 0x2000) + }); + Ok((first("IPv4Addresses"), ipv6)) +} + +/// The public IPv4 address in a `DescribeNetworkInterfaces` answer, if the interface has one. +pub fn parse_public_ip(xml: &str) -> Option { + xml_tag(xml, "publicIp").filter(|a| a.parse::().is_ok()) +} + +/// Read both, with the task role's credentials, in `region` (the task's own). +pub async fn task_addresses(region: &str) -> Result { + let uri = std::env::var("ECS_CONTAINER_METADATA_URI_V4") + .context("ECS_CONTAINER_METADATA_URI_V4 is not set: not an ECS task")?; + let http = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build()?; + let body = http + .get(&uri) + .send() + .await + .context("reading the task metadata")? + .error_for_status()? + .text() + .await?; + let (private_v4, ipv6) = parse_metadata(&body)?; + let ipv4 = match private_v4 { + Some(private) => { + let chain = CredentialChain::new()?; + let ec2 = reqwest::Client::builder() + .timeout(Duration::from_secs(20)) + .build()?; + let body = format!( + "Action=DescribeNetworkInterfaces&Version=2016-11-15&Filter.1.Name=addresses.private-ip-address&Filter.1.Value.1={}", + super::sigv4::uri_encode(&private) + ); + let (status, bytes) = signed( + &chain, + &ec2, + &format!("https://ec2.{region}.amazonaws.com"), + reqwest::Method::POST, + "/", + "", + region, + "ec2", + &[( + "content-type", + "application/x-www-form-urlencoded; charset=utf-8".into(), + )], + body.into_bytes(), + ) + .await?; + let xml = String::from_utf8_lossy(&bytes); + if !status.is_success() { + bail!( + "EC2 DescribeNetworkInterfaces in {region} answered {status}: {}: {}", + xml_tag(&xml, "Code").unwrap_or_default(), + xml_tag(&xml, "Message").unwrap_or_default() + ); + } + parse_public_ip(&xml) + } + None => None, + }; + if ipv6.is_none() && ipv4.is_none() { + bail!("the task has neither a global IPv6 address nor a public IPv4 address"); + } + Ok(TaskAddresses { ipv6, ipv4 }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_task_addresses_are_read_from_metadata_and_ec2() { + let dual = r#"{"DockerId":"x","Networks":[{"NetworkMode":"awsvpc", + "IPv4Addresses":["10.0.1.5"],"IPv6Addresses":["2001:db8:20::5"]}]}"#; + assert_eq!( + parse_metadata(dual).unwrap(), + (Some("10.0.1.5".into()), Some("2001:db8:20::5".into())) + ); + let v4_only = r#"{"Networks":[{"IPv4Addresses":["10.0.1.5"]}]}"#; + assert_eq!( + parse_metadata(v4_only).unwrap(), + (Some("10.0.1.5".into()), None) + ); + // A link-local or unique-local address is not one a browser reaches. + let ula = r#"{"Networks":[{"IPv4Addresses":["10.0.1.5"],"IPv6Addresses":["fd00::5"]}]}"#; + assert_eq!(parse_metadata(ula).unwrap().1, None); + assert!(parse_metadata(r#"{"Networks":[]}"#).is_err()); + + let xml = "\ + eni-0\ + 192.0.2.44amazon\ + 10.0.1.5\ + "; + assert_eq!(parse_public_ip(xml).as_deref(), Some("192.0.2.44")); + assert_eq!( + parse_public_ip(""), + None + ); + } +} diff --git a/src/fleet/coord/mod.rs b/src/fleet/coord/mod.rs index 61bf26811..15ac49a2e 100644 --- a/src/fleet/coord/mod.rs +++ b/src/fleet/coord/mod.rs @@ -13,6 +13,7 @@ pub mod cert; pub mod dns; pub mod ec2; pub mod ec2_aws; +pub mod ecs; pub mod live; pub mod poster; pub mod scale; @@ -740,6 +741,28 @@ pub fn power_managed(args: &FleetCoordArgs) -> bool { } pub async fn run(args: FleetCoordArgs) -> Result<()> { + if args.ec2_aggregate_listen.is_some() + && args.ec2_aggregate_url.is_none() + && !args.ec2_aggregate_tls + { + bail!("--ec2-aggregate-listen needs --ec2-aggregate-url or --ec2-aggregate-tls (how browsers reach it)"); + } + let secrets_region = || { + args.secrets_region + .clone() + .or_else(|| std::env::var("AWS_REGION").ok()) + .context("reading a secret needs --secrets-region or AWS_REGION") + }; + // The EC2 aggregator serving browsers: the task's own addresses, fixed for its life. + let ec2_public = if args.ec2_aggregate_tls { + let region = std::env::var("AWS_REGION") + .context("--ec2-aggregate-tls reads the task's addresses in AWS_REGION")?; + let a = ecs::task_addresses(®ion).await?; + tracing::info!(ipv6 = ?a.ipv6, ipv4 = ?a.ipv4, "the EC2 aggregator's public addresses"); + Some(a) + } else { + None + }; let config = Config { max_tiles: args.max_tiles, ec2_max: match args.ec2 { @@ -747,6 +770,8 @@ pub async fn run(args: FleetCoordArgs) -> Result<()> { FleetEc2::Fake | FleetEc2::Aws => args.ec2_tiles.min(args.max_tiles), }, ec2_aggregate_url: args.ec2_aggregate_url.clone(), + ec2_aggregate_ipv6: ec2_public.as_ref().and_then(|a| a.ipv6.clone()), + ec2_aggregate_ipv4: ec2_public.as_ref().and_then(|a| a.ipv4.clone()), race_interval_secs: args.race_every, restart_cooldown_secs: args.restart_cooldown, camera_seed: args.camera_seed, @@ -837,12 +862,6 @@ pub async fn run(args: FleetCoordArgs) -> Result<()> { spot_wait: Duration::from_secs(args.spot_wait), }) }; - let secrets_region = || { - args.secrets_region - .clone() - .or_else(|| std::env::var("AWS_REGION").ok()) - .context("reading a secret needs --secrets-region or AWS_REGION") - }; let dns_token = match &args.dns_token_secret { Some(name) => Some( secrets::Secrets::new(&secrets_region()?)? @@ -897,6 +916,24 @@ pub async fn run(args: FleetCoordArgs) -> Result<()> { "aggregate".into(), ]; argv.extend(args.aggregate_args.iter().cloned()); + if let (Some(a), Some(secret)) = (&ec2_public, &args.tls_secret) { + // The certificate file, written before the aggregator starts and kept current. + use crate::fleet::agent::tls; + let dir = args.ec2_aggregate_tls_dir.clone(); + let secrets = secrets::Secrets::new(&secrets_region()?)?; + tls::refresh(&secrets, secret, &dir).await?; + tokio::spawn(tls::run(secrets, secret.clone(), dir.clone())); + argv.extend([ + "--tls-file".into(), + dir.join(tls::FILE).to_string_lossy().into_owned(), + ]); + if a.ipv4.is_some() { + argv.extend(["--tls-listen".into(), format!("0.0.0.0:{}", tls::PORT)]); + } + if let Some(v6) = &a.ipv6 { + argv.extend(["--tls-listen".into(), format!("[{v6}]:{}", tls::PORT)]); + } + } tokio::spawn(crate::fleet::aggregate::child::supervise( argv, listen.clone(), diff --git a/tests/test_fleet_coord.rs b/tests/test_fleet_coord.rs index e1fd6c78f..b5ca9566d 100644 --- a/tests/test_fleet_coord.rs +++ b/tests/test_fleet_coord.rs @@ -2934,6 +2934,14 @@ struct DnsFixture { } async fn dns_fixture(pre: &[(&str, &str, &str)]) -> DnsFixture { + dns_fixture_with(pre, Config::default(), None).await +} + +async fn dns_fixture_with( + pre: &[(&str, &str, &str)], + config: Config, + ec2: Option>, +) -> DnsFixture { use fcvm::fleet::coord::dns::Cloudflare; let cf = FakeCf::default(); for (i, (kind, name, content)) in pre.iter().enumerate() { @@ -2957,9 +2965,9 @@ async fn dns_fixture(pre: &[(&str, &str, &str)]) -> DnsFixture { .unwrap() .with_confirm(Arc::new(gate.clone())); let (handle, _actor) = start( - State::new(Config::default(), now_ms()), - None, + State::new(config, now_ms()), None, + ec2, None, None, Some(writer), @@ -3221,18 +3229,74 @@ impl DnsFixture { } fn aggregate(&self) -> Value { + self.aggregate_of("agg:gpu-a") + } + + fn aggregate_of(&self, id: &str) -> Value { let state = &serde_json::from_str::(&self.handle.snapshot.borrow().text).unwrap()["state"]; state["aggregates"] .as_array() .unwrap() .iter() - .find(|a| a["id"] == "agg:gpu-a") + .find(|a| a["id"] == id) .cloned() .unwrap_or(Value::Null) } } +/// The EC2 half's aggregator serving browsers with TLS is named `agg-ec2-` at the +/// coordinator's own addresses, both on 443, and its URLs reach the page only once the +/// names answer. +#[tokio::test] +async fn the_ec2_aggregator_is_published_at_the_coordinators_addresses() { + let config = Config { + ec2_max: 1, + ec2_aggregate_ipv6: Some("2001:db8:20::5".into()), + ec2_aggregate_ipv4: Some("192.0.2.44".into()), + ..Config::default() + }; + let ec2: Arc = Arc::new(ec2::FakeEc2::new( + Duration::from_millis(10), + Duration::from_millis(10), + )); + let fx = dns_fixture_with(&[], config, Some(ec2)).await; + fx.wait_until("the EC2 half's region", |fx| { + !fx.aggregate_of("agg:ec2").is_null() + }) + .await; + let label = fx.label_for("2001:db8:20::5").await; + assert!( + label.starts_with("agg-ec2-") && label.len() == "agg-ec2-".len() + 8, + "{label}" + ); + fx.wait_until("the A record", |fx| { + fx.zone().contains(&( + "A".into(), + format!("{label}-v4.fcvm.example"), + "192.0.2.44".into(), + )) + }) + .await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert!( + fx.aggregate_of("agg:ec2")["frame_url"].is_null(), + "{}", + fx.aggregate_of("agg:ec2") + ); + fx.open(true); + fx.wait_until("the EC2 aggregator published", |fx| { + !fx.aggregate_of("agg:ec2")["frame_url"].is_null() + }) + .await; + let a = fx.aggregate_of("agg:ec2"); + assert_eq!(a["frame_url"], format!("wss://{label}.fcvm.example/frames")); + assert_eq!( + a["frame_url_v4"], + format!("wss://{label}-v4.fcvm.example/frames") + ); +} + /// A host aggregator that serves browsers is named like a launch, under /// `agg--`: its own IPv6 address and the host's IPv4 address on the /// aggregator's port, written, confirmed, and only then given to the page. Before that, From ad48244853b00cfd0a60fc2bd31815995ee9ce05 Mon Sep 17 00:00:00 2001 From: ejc3 Date: Sun, 25 Oct 2026 11:37:20 +0000 Subject: [PATCH 2/4] fleet: publish the host aggregator under agg-- names A host aggregator's stream was reachable only at the plain URL its agent reported, which an https page cannot open, so AggregateView.frame_url was unusable and frame_url_v4 was always null. Agent (`--aggregate-tls-port `, with --tls-secret): - runs the aggregator with --tls-file (the renderers' tls.json) and --tls-listen on :, and with an IPv6 prefix also on [::1]:443. ::1 is the host's side of the prefix, which fcvm never gives a VM; the agent puts it on the host's default IPv6 interface (ip -6 addr replace ... nodad) before it starts. - reports ipv6, and with --public-ipv4-from-imds ipv4 and v4_port, in the aggregate message (new optional fields). fleet-core: - Host keeps the reported addresses and its published names (aggregate_published). Event::Published for `agg-` applies only to the addresses the host reports now. - aggregate_id(host): `agg-` when the host name fits a DNS label. - canonical_ip and Published::names replace the two copies of the canonical comparison (dns.rs, view::frame_urls). - view: with published names an aggregate's frame_url/frame_url_v4 are its published names, else null; without DNS, the reported URL as before. Coordinator: - dns::facts includes each host aggregator (generation 0: one launch while its addresses stay the same), so its names go through the same mint/write/confirm/publish path and zone as slot names; owned() recognises agg--[-v4]. The write/confirm loop is unchanged. - The poster reads a host aggregator at its published name when it has one: the reported URL can be an address only that host reaches. Staging is unchanged until an agent is given --aggregate-port and --aggregate-tls-port. Tests: - a_host_aggregator_is_published_under_its_own_names (tests/test_fleet_coord.rs): no URL for a plain aggregator, records written, URLs only after confirmation, a new address hides the old URLs and the old records go. With the view returning the reported URL it fails: "the plain URL reached an https page". - the_agent_runs_and_feeds_its_aggregator now runs the aggregator with TLS, checks the reported ipv4/v4_port, https /healthz and 404 for /layout over TLS. - only_fleet_names_are_owned, labels_are_owned_and_not_reused, a_host_aggregator_is_named_only_when_its_host_name_fits_a_label, a_slot_address_is_the_gpu_index_plus_two_in_the_prefix. --- fleet/PROTOCOL.md | 27 +++++++-- fleet/core/src/apply.rs | 29 +++++++++- fleet/core/src/protocol.rs | 10 +++- fleet/core/src/state.rs | 65 +++++++++++++++++++++ fleet/core/src/view.rs | 44 ++++++++++----- src/cli/args.rs | 8 +++ src/fleet/agent/addr.rs | 37 ++++++++++++ src/fleet/agent/mod.rs | 106 ++++++++++++++++++++++++++++++----- src/fleet/coord/dns.rs | 85 +++++++++++++++++++--------- src/fleet/coord/mod.rs | 9 ++- src/network/routed.rs | 2 +- tests/test_fleet_agent.rs | 73 ++++++++++++++++++++---- tests/test_fleet_coord.rs | 112 +++++++++++++++++++++++++++++++++++++ 13 files changed, 533 insertions(+), 74 deletions(-) diff --git a/fleet/PROTOCOL.md b/fleet/PROTOCOL.md index c5f44f970..ea6c1cb72 100644 --- a/fleet/PROTOCOL.md +++ b/fleet/PROTOCOL.md @@ -388,7 +388,11 @@ tiles rounded up, whichever is smaller), and a claim when no tile can be lent. and `failures` carry it. - `aggregates` lists the regions an aggregator serves (see Aggregator): one per GPU host that has placed tiles, with the tiles inside it, and the EC2 half when the coordinator - runs its aggregator. `frame_url` is null until the host's aggregator is serving. + runs its aggregator. `frame_url` is null until the host's aggregator is serving. With + the coordinator's DNS (`--dns-token-secret`) a host aggregator's `frame_url` and + `frame_url_v4` are its published names (`wss://agg--./frames`, + `wss://agg---v4.:/frames`), and both stay null for an aggregator + that serves no browser (no `--aggregate-tls-port`). - `actions` holds the latest 50 intents, committed and refused, oldest first: the `viewer` that sent it, the intent `id`, its `args` (a refused word's are `{}`, so its text appears nowhere), and the outcome. @@ -456,8 +460,12 @@ fcvm fleet coord --listen [--agent-cmd '']... [--state-file asked for before it exists and never changes address, so no resolver holds a negative or stale answer for it. A launch's records stay until its slot has a newer published launch or leaves the fleet; names of the fleet's shape (`slot-`, `ec2-`, with or without a - token, with or without `-v4`) that nothing references are deleted, at start too; every - other name in the zone is left alone. + token, `agg--`, each with or without `-v4`) that nothing references are + deleted, at start too; every other name in the zone is left alone. A host aggregator + that reports public addresses is named the same way under `agg--` (a host + name of lowercase letters, digits and inner hyphens, at most 40 characters): AAAA its own + IPv6 address, A the host's IPv4 address, with its port in `frame_url_v4`. Its names last + while it reports the same addresses. - `--tls-secret` keeps the wildcard certificate renewed (see "Certificate"). ### Certificate @@ -765,7 +773,8 @@ round that counts it. fcvm fleet agent (--stdio | --dial [--access-secret ]) --image --base-port [--host ] [--publish-ip ] [--frame-host ] [--gpus =,...] [--fcvm ] [--fcvm-arg ]... - [--spot-notice-url ] [--aggregate-port [--aggregate-arg ]...] + [--spot-notice-url ] [--aggregate-port [--aggregate-arg ]... + [--aggregate-tls-port ]] [--scrub] [--store-ready-file ] [--host-from-imds] [--secret-region ] [--tls-secret [--tls-dir /run/fcvm-fleet/tls]] [--ipv6-prefix | --ipv6-prefix-from-imds] [--frame-zone ] @@ -820,6 +829,16 @@ every assignment or scene change, and reports it once it answers: {"type": "aggregate", "frame_url": "ws://:/frames"} ``` +`--aggregate-tls-port ` (with `--tls-secret`) also serves the stream to browsers with +TLS (the aggregator's `--tls-listen`, the renderers' certificate file): on `:`, +and with an IPv6 prefix on port 443 of `::1`, the host's side of the prefix (fcvm +never gives it to a VM), which the agent puts on the host's default IPv6 interface +(`ip -6 addr replace /128 dev nodad`) before it starts. The message then +carries what the coordinator names (`agg--`, see `--dns-token-secret`): +`ipv6` (port 443), and with `--public-ipv4-from-imds` the host's `ipv4` and the port as +`v4_port`. The aggregator's `--listen` stays on `:`, which the +host's security group does not admit. + ## Aggregator `fcvm fleet aggregate` joins one region's tiles into one stream: the agent runs one per diff --git a/fleet/core/src/apply.rs b/fleet/core/src/apply.rs index def65ba21..4efeb5c3c 100644 --- a/fleet/core/src/apply.rs +++ b/fleet/core/src/apply.rs @@ -197,6 +197,21 @@ fn apply_inner(state: &mut State, event: &Event) -> Option { } Event::Published { slot, names, .. } => { let f = &mut state.fleet; + // A host aggregator's names, for the addresses it reports now. + if let Some(h) = f + .hosts + .values_mut() + .find(|h| crate::state::aggregate_id(&h.name).as_deref() == Some(slot.as_str())) + { + if names.names( + h.aggregate_ipv6.as_deref(), + h.aggregate_ipv4.as_deref(), + h.aggregate_v4_port, + ) { + h.aggregate_published = Some(names.clone()); + } + return None; + } if let Some(s) = f .slots .values_mut() @@ -299,9 +314,17 @@ fn apply_inner(state: &mut State, event: &Event) -> Option { at, ); } - AgentMsg::Aggregate { frame_url } => { + AgentMsg::Aggregate { + frame_url, + ipv6, + ipv4, + v4_port, + } => { if let Some(h) = state.fleet.hosts.get_mut(host) { h.aggregate_url = Some(frame_url.clone()); + h.aggregate_ipv6 = ipv6.clone(); + h.aggregate_ipv4 = ipv4.clone(); + h.aggregate_v4_port = *v4_port; } } AgentMsg::Scrub { ok, .. } => { @@ -1012,6 +1035,10 @@ fn hello(f: &mut Fleet, host: &str, arch: &str, gpus: &[Gpu], capacity: usize, a joined_at: at, region: None, aggregate_url: None, + aggregate_ipv6: None, + aggregate_ipv4: None, + aggregate_v4_port: None, + aggregate_published: None, }, ); f.reclaimed.remove(host); diff --git a/fleet/core/src/protocol.rs b/fleet/core/src/protocol.rs index a9f34fa51..6dba4fdc8 100644 --- a/fleet/core/src/protocol.rs +++ b/fleet/core/src/protocol.rs @@ -195,9 +195,17 @@ pub enum AgentMsg { Draining { reason: String, }, - /// The host's aggregator is serving the host's region at this frame WebSocket. + /// The host's aggregator is serving the host's region at this frame WebSocket, and, + /// when it serves browsers with TLS, at these public addresses: its own IPv6 address + /// (port 443) and the host's IPv4 address on `v4_port`. The coordinator names them. Aggregate { frame_url: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + ipv6: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + ipv4: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + v4_port: Option, }, /// A GPU scrub finished (`ok`), or failed with `error`. Scrub { diff --git a/fleet/core/src/state.rs b/fleet/core/src/state.rs index 93ae170d9..c02f25cf2 100644 --- a/fleet/core/src/state.rs +++ b/fleet/core/src/state.rs @@ -165,6 +165,18 @@ pub struct Host { /// The host aggregator's frame WebSocket, once the agent reports it. #[serde(default)] pub aggregate_url: Option, + /// The aggregator's public addresses, as last reported (`AgentMsg::Aggregate`), and the + /// port of its IPv4 path. + #[serde(default)] + pub aggregate_ipv6: Option, + #[serde(default)] + pub aggregate_ipv4: Option, + #[serde(default)] + pub aggregate_v4_port: Option, + /// The aggregator's published DNS names (with `Config::names_published`), under the + /// label `agg--` (see [`aggregate_id`]). + #[serde(default)] + pub aggregate_published: Option, } /// What an agent (or the EC2 driver) last said about one slot. @@ -627,6 +639,38 @@ pub fn ec2_name(n: u32) -> String { format!("ec2-{n}") } +/// The id a host's aggregator is named under in DNS, `agg-`, when the host's name can +/// be part of a DNS label: lowercase letters, digits and inner hyphens, at most 40 +/// characters (an EC2 instance id is 19), and not `ec2`, kept for the EC2 half's aggregator. +pub fn aggregate_id(host: &str) -> Option { + let ok = !host.is_empty() + && host.len() <= 40 + && host != "ec2" + && !host.starts_with('-') + && !host.ends_with('-') + && host + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-'); + ok.then(|| format!("agg-{host}")) +} + +/// An address in its canonical text (`2001:db8::2` for `2001:0db8::2`); text that is not an +/// address is kept as it is. +pub fn canonical_ip(a: &str) -> String { + a.parse::() + .map(|ip| ip.to_string()) + .unwrap_or_else(|_| a.to_string()) +} + +impl Published { + /// Whether these names are for exactly these addresses and IPv4 port. + pub fn names(&self, ipv6: Option<&str>, ipv4: Option<&str>, v4_port: Option) -> bool { + self.ipv6 == ipv6.map(canonical_ip) + && self.ipv4 == ipv4.map(canonical_ip) + && self.v4_port == v4_port + } +} + /// Parse `slot-` back to its number. pub fn slot_number(id: &str) -> Option { id.strip_prefix("slot-")?.parse().ok() @@ -640,6 +684,27 @@ pub fn ec2_number(id: &str) -> Option { mod tests { use super::*; + #[test] + fn a_host_aggregator_is_named_only_when_its_host_name_fits_a_label() { + assert_eq!( + aggregate_id("i-0123456789abcdef0").as_deref(), + Some("agg-i-0123456789abcdef0") + ); + assert_eq!(aggregate_id("gpu-a").as_deref(), Some("agg-gpu-a")); + for bad in [ + "", + "ec2", + "-gpu", + "gpu-", + "GPU-A", + "gpu.example.com", + "gpu_a", + &"a".repeat(41), + ] { + assert_eq!(aggregate_id(bad), None, "{bad:?}"); + } + } + #[test] fn a_layout_leaves_every_renderer_half_a_second_after_its_post() { for t in [0.0, 0.004, 0.01, 0.034, 12.345, 861.999, 3600.5] { diff --git a/fleet/core/src/view.rs b/fleet/core/src/view.rs index d06d3d17b..b859f6f6d 100644 --- a/fleet/core/src/view.rs +++ b/fleet/core/src/view.rs @@ -272,9 +272,11 @@ pub struct AggregateView { pub half: Half, pub host: Option, pub rect: Rect, - /// The aggregator's frame WebSocket; null until it is serving. `frame_url_v4` is its - /// IPv4 sibling (every `*_url` has one): null for now, since an aggregator serves plain - /// WebSocket on the GPU host or in the coordinator, not under a published name. + /// The aggregator's frame WebSocket over IPv6 and over IPv4; null until it is serving. + /// With published names (the fleet's DNS), those of the addresses it serves now, once + /// published: `wss://agg--./frames` and + /// `wss://agg---v4.:/frames`. Otherwise the URL it was reported + /// at, with no IPv4 one. pub frame_url: Option, pub frame_url_v4: Option, /// The tiles inside the region, by tile id. @@ -484,13 +486,14 @@ pub fn view(state: &State) -> View { .filter_map(|s| Some((s.position?, tile_id(Half::Fcvm, s)?))) .collect(); placed.sort(); + let (frame_url, frame_url_v4) = host_aggregate_urls(state, h); Some(AggregateView { id: format!("agg:{}", h.name), half: Half::Fcvm, host: Some(h.name.clone()), rect, - frame_url: h.aggregate_url.clone(), - frame_url_v4: None, + frame_url, + frame_url_v4, tiles: placed.into_iter().map(|(_, id)| id).collect(), }) }) @@ -835,28 +838,39 @@ pub fn shared_scene(state: &State) -> SharedScene { SharedScene::new(&state.fleet.scene, state.config.image_aspect) } +/// A host aggregator's frame URLs (IPv6, IPv4). With published names, those of the +/// addresses it reports now, once published; otherwise the URL it reported, with no IPv4 one. +pub fn host_aggregate_urls(state: &State, h: &Host) -> (Option, Option) { + if !state.config.names_published { + return (h.aggregate_url.clone(), None); + } + match &h.aggregate_published { + Some(p) + if p.names( + h.aggregate_ipv6.as_deref(), + h.aggregate_ipv4.as_deref(), + h.aggregate_v4_port, + ) => + { + (p.frame_url.clone(), p.frame_url_v4.clone()) + } + _ => (None, None), + } +} + /// A tile's frame URLs (IPv6, IPv4). With published names, those of the launch the slot is /// on, once published; otherwise the URL its agent reported, with no IPv4 one. pub fn frame_urls(state: &State, s: &Slot) -> (Option, Option) { if !state.config.names_published { return (s.observed.frame_url.clone(), None); } - let canonical = |a: &Option| { - a.as_ref().map(|a| { - a.parse::() - .map(|ip| ip.to_string()) - .unwrap_or_else(|_| a.clone()) - }) - }; let o = &s.observed; match &s.published { // The names of exactly what the slot runs now: a new address or port at the same // generation waits for its own names. Some(p) if p.generation == o.generation - && p.ipv6 == canonical(&o.ipv6) - && p.ipv4 == canonical(&o.ipv4) - && p.v4_port == o.v4_port => + && p.names(o.ipv6.as_deref(), o.ipv4.as_deref(), o.v4_port) => { (p.frame_url.clone(), p.frame_url_v4.clone()) } diff --git a/src/cli/args.rs b/src/cli/args.rs index 440cc4c9d..18dd3c24a 100644 --- a/src/cli/args.rs +++ b/src/cli/args.rs @@ -453,6 +453,14 @@ pub struct FleetAgentArgs { #[arg(long = "aggregate-arg", allow_hyphen_values = true)] pub aggregate_args: Vec, + /// Serve the aggregator's stream to browsers with TLS (the --tls-secret certificate): + /// on this port of --publish-ip, reported with the host's public IPv4 address + /// (--public-ipv4-from-imds), and with an IPv6 prefix on port 443 of `::1`, + /// which the agent adds to the host's default IPv6 interface. The coordinator names + /// both (`agg--`). + #[arg(long, requires_all = ["aggregate_port", "tls_secret"])] + pub aggregate_tls_port: Option, + /// A file the host's boot writes once its store is mounted; its modification time is /// reported in the hello (the wake's `store_ready` phase) #[arg(long)] diff --git a/src/fleet/agent/addr.rs b/src/fleet/agent/addr.rs index 0babcc922..9c4f30790 100644 --- a/src/fleet/agent/addr.rs +++ b/src/fleet/agent/addr.rs @@ -33,6 +33,12 @@ impl Prefix { }) } + /// The host aggregator's address, `::1`: the host's side of the prefix, which + /// fcvm never gives a VM. + pub fn aggregator_address(&self) -> Ipv6Addr { + Ipv6Addr::from(self.base | 1) + } + /// The address of the slot on the GPU at `index` in the host's inventory. pub fn slot_address(&self, index: usize) -> Result { let host_bits = 128 - u32::from(self.len); @@ -63,6 +69,36 @@ pub async fn from_imds() -> Result { Prefix::parse(first) } +/// Put `addr` on the host's default IPv6 interface (the one fcvm's routed mode answers for +/// its VMs on), so the aggregator can listen there. Idempotent: an agent that restarts finds +/// it in place. +pub async fn add_host_address(addr: Ipv6Addr) -> Result<()> { + let iface = crate::network::routed::detect_default_ipv6_interface() + .await + .context("the host has no default IPv6 route, so no interface to put the aggregator's address on")?; + let out = tokio::process::Command::new("ip") + .args([ + "-6", + "addr", + "replace", + &format!("{addr}/128"), + "dev", + &iface, + "nodad", + ]) + .output() + .await + .context("running ip")?; + if !out.status.success() { + bail!( + "ip -6 addr replace {addr}/128 dev {iface}: {}", + String::from_utf8_lossy(&out.stderr).trim() + ); + } + tracing::info!(%addr, %iface, "aggregator address on the host"); + Ok(()) +} + /// fcvm's routed mode needs root, so an agent given a prefix must run as root; refusing at /// start beats every VM failing (and being retried) with the same configuration error. pub fn check_routed_user(routed: bool, root: bool) -> Result<()> { @@ -104,6 +140,7 @@ mod tests { assert_eq!(p.to_string(), "2001:db8:1:2:3::/80"); assert_eq!(p.slot_address(0).unwrap().to_string(), "2001:db8:1:2:3::2"); assert_eq!(p.slot_address(7).unwrap().to_string(), "2001:db8:1:2:3::9"); + assert_eq!(p.aggregator_address().to_string(), "2001:db8:1:2:3::1"); let small = Prefix::parse("2001:db8::/124").unwrap(); assert_eq!(small.slot_address(13).unwrap().to_string(), "2001:db8::f"); assert!(small.slot_address(14).is_err(), "past the prefix"); diff --git a/src/fleet/agent/mod.rs b/src/fleet/agent/mod.rs index df7de324a..070f9aa0d 100644 --- a/src/fleet/agent/mod.rs +++ b/src/fleet/agent/mod.rs @@ -32,14 +32,78 @@ pub struct Agent { pub spot_notice_url: Option, /// How often the spot notice URL is polled. pub spot_poll: Duration, - /// Run an aggregator for this host's band on this port, with this command - /// (`fcvm fleet aggregate ...`; `--listen` is added). - pub aggregate: Option<(u16, Vec)>, + /// Run an aggregator for this host's band. + pub aggregate: Option, /// A file the host's boot writes once its store is mounted; its modification time is /// reported in the hello. pub store_ready_file: Option, } +/// The host's aggregator. +#[derive(Debug, Clone)] +pub struct AggregateCfg { + /// Its private listener (`--listen :`): the layout, the stats, and the + /// stream for host-local readers. + pub port: u16, + /// The command (`fcvm fleet aggregate ...`; `--listen` and the TLS arguments are added). + pub argv: Vec, + /// Serving browsers with TLS. + pub public: Option, +} + +/// Where the aggregator serves browsers, and what the agent reports of it. +#[derive(Debug, Clone)] +pub struct AggregatePublic { + /// The certificate file (the renderers' `tls.json`). + pub tls_file: std::path::PathBuf, + /// The IPv4 port, on the publish address. + pub v4_port: u16, + /// The host's public IPv4 address, reported with `v4_port`. + pub ipv4: Option, + /// Its own IPv6 address, served on 443 (the host must hold it). + pub ipv6: Option, +} + +impl AggregateCfg { + /// The aggregator's command with its listeners. + pub fn command(&self, publish_ip: &str) -> (Vec, String) { + let listen = format!("{}:{}", bracket(publish_ip), self.port); + let mut argv = self.argv.clone(); + if let Some(p) = &self.public { + argv.push("--tls-file".into()); + argv.push(p.tls_file.to_string_lossy().into_owned()); + argv.push("--tls-listen".into()); + argv.push(format!("{}:{}", bracket(publish_ip), p.v4_port)); + if let Some(a) = p.ipv6 { + argv.push("--tls-listen".into()); + argv.push(format!("[{a}]:{}", tls::PORT)); + } + } + (argv, listen) + } + + /// The message that reports it serving. + pub fn report(&self, frame_url: String) -> AgentMsg { + let p = self.public.as_ref(); + let ipv4 = p.and_then(|p| p.ipv4.clone()); + AgentMsg::Aggregate { + frame_url, + ipv6: p.and_then(|p| p.ipv6).map(|a| a.to_string()), + v4_port: ipv4.as_ref().and(p.map(|p| p.v4_port)), + ipv4, + } + } +} + +/// An IPv6 literal in brackets, for `
:`. +fn bracket(ip: &str) -> String { + if ip.contains(':') && !ip.starts_with('[') { + format!("[{ip}]") + } else { + ip.to_string() + } +} + /// The modification time of `path` in Unix milliseconds, if it exists. fn mtime_ms(path: &std::path::Path) -> Option { let t = std::fs::metadata(path).ok()?.modified().ok()?; @@ -107,10 +171,10 @@ pub async fn run(args: FleetAgentArgs) -> Result<()> { image: args.image, extra_args: fcvm_args, frame_host: args.frame_host.unwrap_or_else(|| args.publish_ip.clone()), - ipv6_prefix, + ipv6_prefix: ipv6_prefix.clone(), frame_zone: args.frame_zone, tls_publish_base: args.tls_publish_base, - public_ipv4, + public_ipv4: public_ipv4.clone(), publish_ip: args.publish_ip, sysfs_root: args.sysfs_root, start_timeout: Duration::from_secs(args.start_timeout_secs), @@ -127,7 +191,7 @@ pub async fn run(args: FleetAgentArgs) -> Result<()> { spot_notice_url: args.spot_notice_url, spot_poll: Duration::from_secs(5), store_ready_file: args.store_ready_file.clone(), - aggregate: args.aggregate_port.map(|p| { + aggregate: args.aggregate_port.map(|port| { let mut argv = vec![ std::env::current_exe() .map(|e| e.to_string_lossy().into_owned()) @@ -136,9 +200,27 @@ pub async fn run(args: FleetAgentArgs) -> Result<()> { "aggregate".into(), ]; argv.extend(args.aggregate_args.iter().cloned()); - (p, argv) + AggregateCfg { + port, + argv, + public: args.aggregate_tls_port.map(|v4_port| AggregatePublic { + tls_file: args.tls_dir.join(tls::FILE), + v4_port, + ipv4: public_ipv4.clone(), + ipv6: ipv6_prefix.as_ref().map(addr::Prefix::aggregator_address), + }), + } }), }; + // The aggregator listens on its own IPv6 address, so the host must hold it first. + if let Some(a) = agent + .aggregate + .as_ref() + .and_then(|a| a.public.as_ref()) + .and_then(|p| p.ipv6) + { + addr::add_host_address(a).await?; + } let Some(url) = args.dial else { return serve(agent, tokio::io::stdin(), tokio::io::stdout()).await; }; @@ -248,18 +330,16 @@ where let (budget_tx, budget_rx) = watch::channel::>(None); // The aggregator's layout follows the assignment and the scene. let (layout_tx, layout_rx) = watch::channel::>(None); - let aggregator = agent.aggregate.clone().map(|(port, argv)| { - let listen = format!("{}:{port}", cfg.publish_ip); - let url = slot::frame_url(&cfg.frame_host, port); + let aggregator = agent.aggregate.clone().map(|a| { + let (argv, listen) = a.command(&cfg.publish_ip); + let report = a.report(slot::frame_url(&cfg.frame_host, a.port)); let out = out.clone(); tokio::spawn(crate::fleet::aggregate::child::supervise( argv, listen, layout_rx, move || { - let _ = out.send(AgentMsg::Aggregate { - frame_url: url.clone(), - }); + let _ = out.send(report.clone()); }, )) }); diff --git a/src/fleet/coord/dns.rs b/src/fleet/coord/dns.rs index f117d5a84..8ff1f09f0 100644 --- a/src/fleet/coord/dns.rs +++ b/src/fleet/coord/dns.rs @@ -7,6 +7,10 @@ //! - `