Something went wrong. Try again.
Web frontend and supporting services for lance.blue
Something went wrong. Try again.
14 kB · 390 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391//! The AWS calls behind a match: manifest to S3 with a presigned GET,//! RunTask, DescribeTasks. Confined here the way atrium is confined to//! atproto. Credentials and region come from the instance (IMDS); nothing is//! configured in code.
use std::time::Duration;
use aws_sdk_ecs::types::{ AwsVpcConfiguration, ContainerOverride, KeyValuePair, NetworkConfiguration, TaskOverride,};
use crate::config::MatchConfig;
pub struct Aws { s3: aws_sdk_s3::Client, ecs: aws_sdk_ecs::Client,}
/// Never carries AWS error values into logs wholesale - like atrium errors,/// they can embed request details. Log the category; the operator has/// CloudTrail for the rest.#[derive(Debug)]pub struct AwsFailed(pub &'static str);
impl Aws { pub async fn new() -> Aws { let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await; Aws { s3: aws_sdk_s3::Client::new(&config), ecs: aws_sdk_ecs::Client::new(&config), } }
pub async fn put_manifest( &self, bucket: &str, match_id: &str, manifest: &serde_json::Value, ) -> Result<String, AwsFailed> { let key = format!("matches/{match_id}/manifest.json"); self.s3 .put_object() .bucket(bucket) .key(&key) .content_type("application/json") .body(manifest.to_string().into_bytes().into()) .send() .await .map_err(|_| AwsFailed("manifest upload"))?; Ok(key) }
/// A sanitised camo for one match, beside that match's manifest rather /// than in a shared place: it is fetched once at launch and has no life /// after the match, and a per-match key means one player's camo is never /// served under a name another match could guess. /// /// `name` separates the slots' files under that one prefix - `camo.png` /// for the player, `camo-bot.png` for the opponent's rolled one. pub async fn put_camo( &self, bucket: &str, match_id: &str, name: &str, png: Vec<u8>, ) -> Result<String, AwsFailed> { let key = format!("matches/{match_id}/{name}"); self.s3 .put_object() .bucket(bucket) .key(&key) .content_type("image/png") .body(png.into()) .send() .await .map_err(|_| AwsFailed("camo upload"))?; Ok(key) }
/// The presigned URL is a bearer credential; it goes into the container /// override and nowhere else - never logged, never stored. pub async fn presign_get(&self, bucket: &str, key: &str) -> Result<String, AwsFailed> { let presigned = self .s3 .get_object() .bucket(bucket) .key(key) .presigned( aws_sdk_s3::presigning::PresigningConfig::expires_in(Duration::from_secs(900)) .expect("static expiry is valid"), ) .await .map_err(|_| AwsFailed("presign"))?; Ok(presigned.uri().to_string()) }
/// Presign a PUT for one artifact, so the container sends the bytes to S3 /// instead of through here. /// /// Minted per upload rather than at manifest build time, and that is the /// whole point: a presigned URL dies with the credentials that signed it, /// these are IMDS instance-profile credentials, and IMDS rotates them on /// its own schedule and will not issue a fresh set on request. Signing at /// launch would tie every upload URL's life to how long a match runs, with /// a ceiling that is both low and unpredictable. Signing here means the /// signature only has to survive one PUT. /// /// The signature covers the bucket, the key and the method, so the container /// can only write the one artifact it asked for. /// Read back a camo this service staged for a match. /// /// Its own reader because it is not an artifact: a camo is written by us /// before the match under `matches/<id>/<name>`, and everything the /// container uploads lands under `matches/<id>/artifacts/`. Reading one /// with the other's key finds nothing, quietly, which is exactly what the /// social card did until somebody noticed the machines were grey. pub async fn get_camo( &self, bucket: &str, match_id: &str, name: &str, ) -> Result<Option<Vec<u8>>, AwsFailed> { let sent = self .s3 .get_object() .bucket(bucket) .key(format!("matches/{match_id}/{name}")) .send() .await; let object = match sent { Ok(object) => object, // A match with no camo staged for a slot is the ordinary case and // not a failure. Told apart here because the caller caches on the // difference: a card drawn without colours nobody chose can stand // for a day, one drawn without colours S3 would not hand over // must not. Err(e) => { return if e.as_service_error().is_some_and(|e| e.is_no_such_key()) { Ok(None) } else { Err(AwsFailed("camo read")) }; } }; let bytes = object .body .collect() .await .map_err(|_| AwsFailed("camo body"))?; Ok(Some(bytes.into_bytes().to_vec())) }
/// Read an artifact a match uploaded. /// /// The only read of the artifacts bucket. Everything else here signs a /// URL and lets the container do the transfer; a finished match's result /// has to be read by us, because it is what a score is computed from. /// Whether an artifact is there, without pulling it. The page needs to /// know a board exists to offer the card drawn on it, and a board render /// is megabytes. pub async fn has_artifact( &self, bucket: &str, match_id: &str, name: &str, ) -> Result<bool, AwsFailed> { match self .s3 .head_object() .bucket(bucket) .key(format!("matches/{match_id}/artifacts/{name}")) .send() .await { Ok(_) => Ok(true), Err(e) => { // A head on a missing key answers 404 with no modelled error // variant, unlike get_object's NoSuchKey. if e.raw_response().is_some_and(|r| r.status().as_u16() == 404) { Ok(false) } else { Err(AwsFailed("artifact head")) } } } }
pub async fn get_artifact( &self, bucket: &str, match_id: &str, name: &str, ) -> Result<Vec<u8>, AwsFailed> { let object = self .s3 .get_object() .bucket(bucket) .key(format!("matches/{match_id}/artifacts/{name}")) .send() .await .map_err(|_| AwsFailed("artifact read"))?; let bytes = object .body .collect() .await .map_err(|_| AwsFailed("artifact body"))?; Ok(bytes.into_bytes().to_vec()) }
pub async fn presign_artifact_put( &self, bucket: &str, match_id: &str, name: &str, expires_in: Duration, ) -> Result<String, AwsFailed> { let presigned = self .s3 .put_object() .bucket(bucket) .key(format!("matches/{match_id}/artifacts/{name}")) .presigned( aws_sdk_s3::presigning::PresigningConfig::expires_in(expires_in) .map_err(|_| AwsFailed("presign config"))?, ) .await .map_err(|_| AwsFailed("presign artifact put"))?; Ok(presigned.uri().to_string()) }
pub async fn run_task( &self, config: &MatchConfig, manifest_url: &str, size: &super::sizing::TaskSize, ) -> Result<String, AwsFailed> { let network = NetworkConfiguration::builder() .awsvpc_configuration( AwsVpcConfiguration::builder() .set_subnets(Some(config.subnets.clone())) .security_groups(&config.task_security_group) // No NAT gateway: without a public IP the image pull // hangs. .assign_public_ip(aws_sdk_ecs::types::AssignPublicIp::Enabled) .build() .map_err(|_| AwsFailed("network configuration"))?, ) .build();
// The match's size rides the same per-task override the manifest URL // does: the task definition stays the one-human shape, and a bigger // match asks for more only for its own task. let overrides = TaskOverride::builder() .cpu(size.cpu) .memory(&size.memory) .container_overrides( ContainerOverride::builder() .name("arena") .environment( KeyValuePair::builder() .name("ARENA_MANIFEST_URL") .value(manifest_url) .build(), ) .build(), ) .build();
let output = self .ecs .run_task() .cluster(&config.cluster) .task_definition(&config.task_family) .launch_type(aws_sdk_ecs::types::LaunchType::Fargate) .network_configuration(network) .overrides(overrides) .send() .await .map_err(|_| AwsFailed("run task"))?;
output .tasks() .first() .and_then(|task| task.task_arn()) .map(str::to_owned) .ok_or(AwsFailed("task arn missing")) }
/// Last status plus the private IP once the ENI has one. DescribeTasks /// exposes the IP as an attachment detail, so no EC2 call is needed. pub async fn task_status(&self, cluster: &str, task_arn: &str) -> Result<TaskState, AwsFailed> { let output = self .ecs .describe_tasks() .cluster(cluster) .tasks(task_arn) .send() .await .map_err(|_| AwsFailed("describe task"))?; let task = output.tasks().first().ok_or(AwsFailed("task not found"))?;
let status = task.last_status().unwrap_or("UNKNOWN").to_owned(); let ip = task .attachments() .iter() .flat_map(|attachment| attachment.details()) .find(|detail| detail.name() == Some("privateIPv4Address")) .and_then(|detail| detail.value()) .map(str::to_owned); Ok(TaskState { status, ip, timing: TaskTiming { provision_ms: delta_ms(task.created_at(), task.connectivity_at()), pull_ms: delta_ms(task.pull_started_at(), task.pull_stopped_at()), start_ms: delta_ms(task.pull_stopped_at(), task.started_at()), total_ms: delta_ms(task.created_at(), task.started_at()), }, }) }}
pub struct TaskState { pub status: String, pub ip: Option<String>, pub timing: TaskTiming,}
/// What ECS can say about a start that the container cannot. The task is only/// alive for the last of these: everything before `start_ms` happens before any/// of arena's own phase marks exist, and the pull in particular is invisible/// from inside. Every field is None until the task reaches the phase, and stays/// None if ECS omits the timestamp.#[derive(Debug)]pub struct TaskTiming { /// RunTask accepted to ENI attached. pub provision_ms: Option<i64>, /// The image pull. The number `plan/arena-image.md` has only ever guessed at. pub pull_ms: Option<i64>, /// Pull finished to container running - Fargate's own start overhead. pub start_ms: Option<i64>, /// RunTask accepted to container running. pub total_ms: Option<i64>,}
fn delta_ms( from: Option<&aws_sdk_ecs::primitives::DateTime>, to: Option<&aws_sdk_ecs::primitives::DateTime>,) -> Option<i64> { let millis = |t: &aws_sdk_ecs::primitives::DateTime| { t.secs() * 1000 + i64::from(t.subsec_nanos()) / 1_000_000 }; Some(millis(to?) - millis(from?))}
#[cfg(test)]mod tests { use super::*; use aws_sdk_ecs::primitives::DateTime;
#[test] fn both_present_gives_the_millisecond_gap() { let from = DateTime::from_millis(1_000); let to = DateTime::from_millis(2_500); assert_eq!(delta_ms(Some(&from), Some(&to)), Some(1_500)); }
#[test] fn a_missing_endpoint_gives_no_timing() { let present = DateTime::from_millis(1_000); assert_eq!(delta_ms(None, Some(&present)), None); assert_eq!(delta_ms(Some(&present), None), None); }
#[test] fn both_missing_gives_no_timing() { assert_eq!(delta_ms(None, None), None); }
/// ECS timestamps are not guaranteed monotonic against the phase order /// this module assumes (a clock can be corrected between two events); /// `delta_ms` does not clamp, so a `to` earlier than `from` comes out /// negative rather than panicking or getting silently flipped. #[test] fn to_before_from_is_a_negative_delta() { let later = DateTime::from_millis(5_000); let earlier = DateTime::from_millis(1_000); assert_eq!(delta_ms(Some(&later), Some(&earlier)), Some(-4_000)); }}