Skip to content
Merged
42 changes: 37 additions & 5 deletions fleet/PROTOCOL.md
Original file line number Diff line number Diff line change
Expand Up @@ -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-<host>-<token>.<zone>/frames`,
`wss://agg-<host>-<token>-v4.<zone>:<port>/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.
Expand All @@ -410,7 +414,8 @@ relaunching (its region would be black; claimed GPUs and dark tiles do not count
fcvm fleet coord --listen <addr:port> [--agent-cmd '<command>']... [--state-file <path>]
[--text-check bedrock --text-check-model <id> --text-check-region <region> | --text-check none]
[--max-tiles 100] [--ec2 none|fake] [--allowed-origin <scheme://host[:port]>]...
[--ec2-aggregate-listen <addr:port> --ec2-aggregate-url <ws url>] [--aggregate-arg <arg>]...
[--ec2-aggregate-listen <addr:port> (--ec2-aggregate-url <ws url> | --ec2-aggregate-tls
[--ec2-aggregate-tls-dir /run/fcvm-fleet/tls])] [--aggregate-arg <arg>]...
[--race-every <secs>] [--restart-cooldown 60] [--camera-seed 1] [--max-borrowed <n>]
[--trust-cf-connecting-ip] [--turnstile-secret <secret> --turnstile-site-key <key>]
[--admin-listen <addr:port>] [--ffmpeg ffmpeg]
Expand Down Expand Up @@ -470,8 +475,12 @@ fcvm fleet coord --listen <addr:port> [--agent-cmd '<command>']... [--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-<n>`, `ec2-<n>`, 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-<host>-<token>`, 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-<host>-<token>` (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.
- `--dns-subzone` makes the coordinator the authoritative DNS server of a subzone of
`--dns-zone` (the parent delegates it with NS records naming `--dns-subzone-ns`, and no DS
record) and names launches there instead: `<label>.<subzone>` AAAA and
Expand All @@ -487,6 +496,18 @@ fcvm fleet coord --listen <addr:port> [--agent-cmd '<command>']... [--state-file
dropped. With `--dns-token-secret` it deletes, once at start, the launch names earlier
written to the parent zone.
- `--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-<token>` (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

Expand Down Expand Up @@ -828,7 +849,8 @@ round that counts it.
fcvm fleet agent (--stdio | --dial <wss url> [--access-secret <secret>]) --image <renderer image> --base-port <port>
[--host <name>] [--publish-ip <ip>] [--frame-host <name or ip>]
[--gpus <sbdf>=<name>,...] [--fcvm <path>] [--fcvm-arg <arg>]...
[--spot-notice-url <url>] [--aggregate-port <port> [--aggregate-arg <arg>]...]
[--spot-notice-url <url>] [--aggregate-port <port> [--aggregate-arg <arg>]...
[--aggregate-tls-port <port>]]
[--scrub] [--store-ready-file <path>] [--host-from-imds] [--secret-region <region>]
[--tls-secret <secret> [--tls-dir /run/fcvm-fleet/tls]]
[--ipv6-prefix <prefix/len> | --ipv6-prefix-from-imds] [--frame-zone <zone>]
Expand Down Expand Up @@ -883,6 +905,16 @@ every assignment or scene change, and reports it once it answers:
{"type": "aggregate", "frame_url": "ws://<frame host>:<port>/frames"}
```

`--aggregate-tls-port <port>` (with `--tls-secret`) also serves the stream to browsers with
TLS (the aggregator's `--tls-listen`, the renderers' certificate file): on `<publish ip>:<port>`,
and with an IPv6 prefix on port 443 of `<prefix>::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 <addr>/128 dev <iface> nodad`) before it starts. The message then
carries what the coordinator names (`agg-<host>-<token>`, 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 `<publish ip>:<aggregate port>`, 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
Expand Down
4 changes: 4 additions & 0 deletions fleet/coord/Containerfile
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ FROM docker.io/library/ubuntu:24.04
RUN apt-get update -q && apt-get install -y -q --no-install-recommends ffmpeg ca-certificates \
&& rm -rf /var/lib/apt/lists/*
COPY fcvm /usr/local/bin/fcvm
# --ec2-aggregate-tls writes the certificate file under /run/fcvm-fleet (the task mounts a
# volume there; the root file system is read-only). The directory is the coordinator's own,
# so a volume that takes the image's ownership is writable by it.
RUN mkdir -p /run/fcvm-fleet && chown 65534:65534 /run/fcvm-fleet
USER 65534:65534
EXPOSE 8080
ENTRYPOINT ["/usr/local/bin/fcvm", "fleet", "coord"]
Expand Down
40 changes: 39 additions & 1 deletion fleet/core/src/apply.rs
Original file line number Diff line number Diff line change
Expand Up @@ -196,7 +196,33 @@ fn apply_inner(state: &mut State, event: &Event) -> Option<Receipt> {
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
.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()
Expand Down Expand Up @@ -299,9 +325,17 @@ fn apply_inner(state: &mut State, event: &Event) -> Option<Receipt> {
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, .. } => {
Expand Down Expand Up @@ -1026,6 +1060,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);
Expand Down
10 changes: 9 additions & 1 deletion fleet/core/src/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -205,9 +205,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<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
ipv4: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
v4_port: Option<u16>,
},
/// A GPU scrub finished (`ok`), or failed with `error`.
Scrub {
Expand Down
82 changes: 82 additions & 0 deletions fleet/core/src/state.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<String>,
/// 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-<token>`.
#[serde(default)]
pub ec2_aggregate_ipv6: Option<String>,
#[serde(default)]
pub ec2_aggregate_ipv4: Option<String>,
/// The owner's race cadence: a relaunch of every tile this often. `None`: no schedule.
pub race_interval_secs: Option<u64>,
/// After any race, `restart_now` is refused for this long, for everyone.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -165,6 +173,18 @@ pub struct Host {
/// The host aggregator's frame WebSocket, once the agent reports it.
#[serde(default)]
pub aggregate_url: Option<String>,
/// The aggregator's public addresses, as last reported (`AgentMsg::Aggregate`), and the
/// port of its IPv4 path.
#[serde(default)]
pub aggregate_ipv6: Option<String>,
#[serde(default)]
pub aggregate_ipv4: Option<String>,
#[serde(default)]
pub aggregate_v4_port: Option<u16>,
/// The aggregator's published DNS names (with `Config::names_published`), under the
/// label `agg-<host>-<token>` (see [`aggregate_id`]).
#[serde(default)]
pub aggregate_published: Option<Published>,
}

/// What an agent (or the EC2 driver) last said about one slot.
Expand Down Expand Up @@ -511,6 +531,9 @@ pub struct Fleet {
pub budget: crate::budget::Budget,
/// Coordinator components that are degraded: reason and since when.
pub degraded: BTreeMap<String, (String, u64)>,
/// The EC2 half's aggregator's published DNS names (`EC2_AGGREGATE_ID`).
#[serde(default)]
pub ec2_aggregate_published: Option<Published>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
Expand Down Expand Up @@ -560,6 +583,7 @@ impl State {
warming: Default::default(),
power: Power::default(),
budget: Default::default(),
ec2_aggregate_published: None,
},
config,
receipts: BTreeMap::new(),
Expand All @@ -583,6 +607,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();
Expand Down Expand Up @@ -636,6 +662,41 @@ pub fn ec2_name(n: u32) -> String {
format!("ec2-{n}")
}

/// The id a host's aggregator is named under in DNS, `agg-<host>`, 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<String> {
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}"))
}

/// 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 {
a.parse::<std::net::IpAddr>()
.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<u16>) -> bool {
self.ipv6 == ipv6.map(canonical_ip)
&& self.ipv4 == ipv4.map(canonical_ip)
&& self.v4_port == v4_port
}
}

/// Parse `slot-<n>` back to its number.
pub fn slot_number(id: &str) -> Option<u32> {
id.strip_prefix("slot-")?.parse().ok()
Expand All @@ -649,6 +710,27 @@ pub fn ec2_number(id: &str) -> Option<u32> {
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] {
Expand Down
Loading
Loading