feat(aws): environment record, agent token, access point wait and resource removal
RuntimeApi (GetParameter, GetRole, GetSecretValue, DescribeAccessPoints, task definition retirement, DeleteSecret with a 7-day window, DeleteAccessPoint) with the decisions in tested functions over it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
shadowdaoandClaude Opus 5.5 committed
1 parent
e43ed26e20
commit
7fd6fa6aaa
3 files changed
+840
-1
No files matched your search
@@ -754,7 +754,7 @@ fn error_parts<E: ProvideErrorMetadata>(
|
||||
}
|
||||
|
||||
/// Yes / No (missing) / Unknown (denied) from a lookup's result.
|
||||
fn knowable<T, E>(
|
||||
pub(crate) fn knowable<T, E>(
|
||||
action: &str,
|
||||
res: Result<T, SdkError<E, HttpResponse>>,
|
||||
) -> Result<Knowable, AwsError>
|
||||
|
||||
@@ -0,0 +1,498 @@
|
||||
//! What the AWS host asks AWS around a task, besides running it
|
||||
//! (`tasks.rs`) and preparing its resources (`workspace.rs`): the
|
||||
//! environment's record, whether the caller is onboarded, the agent token of
|
||||
//! a running task (P2), waiting for a new access point, and removing a
|
||||
//! project's resources.
|
||||
//!
|
||||
//! Like the other `aws/` modules, the decisions are made by the functions
|
||||
//! here over a small trait ([`RuntimeApi`]) so a fake exercises them; the SDK
|
||||
//! calls are in [`SdkRuntimeApi`].
|
||||
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
use aws_config::SdkConfig;
|
||||
use aws_sdk_secretsmanager::error::ProvideErrorMetadata;
|
||||
|
||||
use super::environments::{knowable, parse_environment_record, EnvironmentInfo, Knowable};
|
||||
use super::error::{classify_sdk_error, AwsError};
|
||||
use super::tasks::{find_project_task, TaskApi, STOP_REASON};
|
||||
use super::tofu_modules::{
|
||||
access_point_root, environment_parameter_name, project_secret_name, task_family,
|
||||
user_file_system_token, user_role_names,
|
||||
};
|
||||
use super::workspace::WorkspaceApi;
|
||||
|
||||
/// The secret key holding the agent token (Task 4's payload).
|
||||
pub const AGENT_TOKEN_KEY: &str = "TRIPLEC_AGENT_TOKEN";
|
||||
/// Secrets Manager's recovery window when a project is removed.
|
||||
pub const SECRET_RECOVERY_DAYS: i64 = 7;
|
||||
/// A project secret is at most 64 KiB; anything bigger is not ours.
|
||||
const MAX_SECRET_READ: usize = 70_000;
|
||||
|
||||
fn other(message: impl Into<String>) -> AwsError {
|
||||
AwsError::Other {
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
|
||||
pub trait RuntimeApi: Send + Sync {
|
||||
/// `ssm:GetParameter` of an environment's record; `None` when absent.
|
||||
fn environment_record(
|
||||
&self,
|
||||
env: &str,
|
||||
) -> impl Future<Output = Result<Option<String>, AwsError>> + Send;
|
||||
/// `iam:GetRole`.
|
||||
fn role_exists(&self, name: &str) -> impl Future<Output = Result<Knowable, AwsError>> + Send;
|
||||
/// `secretsmanager:GetSecretValue`'s `SecretString`.
|
||||
fn secret_string(&self, name: &str) -> impl Future<Output = Result<String, AwsError>> + Send;
|
||||
/// The access point's `LifeCycleState`; `None` when it is gone.
|
||||
fn access_point_state(
|
||||
&self,
|
||||
id: &str,
|
||||
) -> impl Future<Output = Result<Option<String>, AwsError>> + Send;
|
||||
fn deregister_task_definition(
|
||||
&self,
|
||||
arn: &str,
|
||||
) -> impl Future<Output = Result<(), AwsError>> + Send;
|
||||
fn delete_task_definitions(
|
||||
&self,
|
||||
arns: &[String],
|
||||
) -> impl Future<Output = Result<(), AwsError>> + Send;
|
||||
/// Schedule deletion with a [`SECRET_RECOVERY_DAYS`] window. A secret
|
||||
/// that does not exist is success.
|
||||
fn delete_secret(&self, name: &str) -> impl Future<Output = Result<(), AwsError>> + Send;
|
||||
/// A missing access point is success.
|
||||
fn delete_access_point(&self, id: &str) -> impl Future<Output = Result<(), AwsError>> + Send;
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------ environment
|
||||
|
||||
/// The environment's record, read with one `GetParameter` and validated.
|
||||
pub async fn read_environment<R: RuntimeApi>(
|
||||
rt: &R,
|
||||
env: &str,
|
||||
account: &str,
|
||||
region: &str,
|
||||
) -> Result<EnvironmentInfo, AwsError> {
|
||||
let value = rt.environment_record(env).await?.ok_or_else(|| {
|
||||
other(format!(
|
||||
"The environment \"{env}\" was not found in {region}. It may have been removed; \
|
||||
check Settings → Hosts."
|
||||
))
|
||||
})?;
|
||||
parse_environment_record(env, &value, account, region).map_err(|e| {
|
||||
other(format!(
|
||||
"The record of environment \"{env}\" in {region} is not usable: {e}"
|
||||
))
|
||||
})
|
||||
}
|
||||
|
||||
/// Whether the caller's task role exists in the environment: the onboarding
|
||||
/// signal a workspace user can read. A denied lookup is not an answer, so
|
||||
/// it is let through (RunTask will say what is wrong).
|
||||
pub async fn check_onboarded<R: RuntimeApi>(
|
||||
rt: &R,
|
||||
env: &str,
|
||||
owner_key: &str,
|
||||
) -> Result<(), AwsError> {
|
||||
let (task_role, _) = user_role_names(env, owner_key);
|
||||
match rt.role_exists(&task_role).await? {
|
||||
Knowable::No => Err(other(format!(
|
||||
"You have not joined the environment \"{env}\" yet. Join it in Settings → Hosts \
|
||||
(or ask an administrator to onboard you), then start the project again."
|
||||
))),
|
||||
Knowable::Yes | Knowable::Unknown => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The agent token in a project secret's JSON. The error never echoes the
|
||||
/// secret.
|
||||
pub fn agent_token_from_secret(json: &str) -> Result<String, String> {
|
||||
if json.len() > MAX_SECRET_READ {
|
||||
return Err("The project secret is larger than a project secret can be".into());
|
||||
}
|
||||
let v: serde_json::Value = serde_json::from_str(json)
|
||||
.map_err(|_| "The project secret is not a JSON object".to_string())?;
|
||||
v.get(AGENT_TOKEN_KEY)
|
||||
.and_then(|t| t.as_str())
|
||||
.filter(|t| !t.is_empty() && t.bytes().all(|b| b.is_ascii_graphic()))
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| "The project secret holds no agent token".to_string())
|
||||
}
|
||||
|
||||
/// The token a running task of this project started with (P2: the caller
|
||||
/// may read their own secret; `prepare_workspace` wrote it before RunTask).
|
||||
pub async fn read_agent_token<R: RuntimeApi>(
|
||||
rt: &R,
|
||||
env: &str,
|
||||
owner_key: &str,
|
||||
project_id: &str,
|
||||
) -> Result<String, AwsError> {
|
||||
let name = project_secret_name(env, owner_key, project_id);
|
||||
let json = rt.secret_string(&name).await?;
|
||||
agent_token_from_secret(&json).map_err(other)
|
||||
}
|
||||
|
||||
/// How long a fresh access point may take to become `available`.
|
||||
#[derive(Debug, Clone, Copy)]
|
||||
pub struct PollTiming {
|
||||
pub interval: Duration,
|
||||
pub timeout: Duration,
|
||||
}
|
||||
|
||||
impl Default for PollTiming {
|
||||
fn default() -> Self {
|
||||
PollTiming {
|
||||
interval: Duration::from_secs(2),
|
||||
timeout: Duration::from_secs(90),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Wait until the access point is `available` (a task mounting one that is
|
||||
/// still `creating` fails to start).
|
||||
pub async fn wait_access_point<R: RuntimeApi>(
|
||||
rt: &R,
|
||||
id: &str,
|
||||
timing: PollTiming,
|
||||
) -> Result<(), AwsError> {
|
||||
let deadline = tokio::time::Instant::now() + timing.timeout;
|
||||
loop {
|
||||
match rt.access_point_state(id).await?.as_deref() {
|
||||
Some("available") => return Ok(()),
|
||||
Some("creating") | Some("updating") => {}
|
||||
Some(s) => {
|
||||
return Err(other(format!(
|
||||
"The project's EFS access point {id} is {s}; start the project again"
|
||||
)))
|
||||
}
|
||||
None => {
|
||||
return Err(other(format!(
|
||||
"The project's EFS access point {id} disappeared; start the project again"
|
||||
)))
|
||||
}
|
||||
}
|
||||
if tokio::time::Instant::now() + timing.interval > deadline {
|
||||
return Err(other(format!(
|
||||
"The project's EFS access point {id} did not become available in time; \
|
||||
start the project again in a minute"
|
||||
)));
|
||||
}
|
||||
tokio::time::sleep(timing.interval).await;
|
||||
}
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ ARNs
|
||||
|
||||
/// The cluster *name* of `arn:...:task/<cluster>/<id>` (ECS accepts a name
|
||||
/// wherever it takes a cluster).
|
||||
pub fn cluster_of_task(task_arn: &str) -> Option<&str> {
|
||||
let (_, rest) = task_arn.split_once(":task/")?;
|
||||
let (cluster, id) = rest.split_once('/')?;
|
||||
(task_arn.starts_with("arn:") && !cluster.is_empty() && !id.is_empty() && !id.contains('/'))
|
||||
.then_some(cluster)
|
||||
}
|
||||
|
||||
/// The cluster ARN of a task ARN (same partition, region and account).
|
||||
pub fn cluster_arn_of_task(task_arn: &str) -> Option<String> {
|
||||
let cluster = cluster_of_task(task_arn)?;
|
||||
let (prefix, _) = task_arn.split_once(":task/")?;
|
||||
Some(format!("{prefix}:cluster/{cluster}"))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- removal
|
||||
|
||||
/// What removing a project's AWS resources needs to know.
|
||||
pub struct RemoveRequest<'a> {
|
||||
pub env: &'a str,
|
||||
pub region: &'a str,
|
||||
pub owner_user_id: &'a str,
|
||||
pub owner_key: &'a str,
|
||||
pub project_id: &'a str,
|
||||
/// The task ARN the store has, if any (stopped too, in case the listing
|
||||
/// missed it).
|
||||
pub stored_task_arn: Option<&'a str>,
|
||||
pub stored_access_point_id: Option<&'a str>,
|
||||
/// Delete the EFS access point. Its data stays on the file system until
|
||||
/// an administrator removes it; the path is not reused.
|
||||
pub discard_data: bool,
|
||||
}
|
||||
|
||||
/// Stop the project's task, retire its task definition revisions, schedule
|
||||
/// its secret for deletion (7 days) and, with `discard_data`, delete its
|
||||
/// access point. A task that can't be stopped stops the removal (nothing
|
||||
/// billed may be left running); the rest is attempted in full and every
|
||||
/// failure reported.
|
||||
pub async fn remove_workspace_resources<T: TaskApi, W: WorkspaceApi, R: RuntimeApi>(
|
||||
tasks: &T,
|
||||
work: &W,
|
||||
rt: &R,
|
||||
req: &RemoveRequest<'_>,
|
||||
) -> Result<(), AwsError> {
|
||||
let cluster = format!("triple-c-{}", req.env);
|
||||
let family = task_family(req.env, req.owner_key, req.project_id);
|
||||
|
||||
// 1. The task.
|
||||
let live =
|
||||
find_project_task(tasks, &cluster, &family, req.owner_user_id, req.project_id).await?;
|
||||
let mut to_stop: Vec<String> = live.into_iter().map(|t| t.arn).collect();
|
||||
if let Some(arn) = req.stored_task_arn {
|
||||
if cluster_of_task(arn) == Some(cluster.as_str()) && !to_stop.iter().any(|a| a == arn) {
|
||||
to_stop.push(arn.to_string());
|
||||
}
|
||||
}
|
||||
for arn in &to_stop {
|
||||
tasks.stop_task(&cluster, arn, STOP_REASON).await?;
|
||||
}
|
||||
|
||||
let mut failures: Vec<String> = Vec::new();
|
||||
|
||||
// 2. Task definition revisions the caller registered for this project.
|
||||
let mut retired = Vec::new();
|
||||
match work.task_definition_revisions(&family).await {
|
||||
Ok(revisions) => {
|
||||
for arn in revisions {
|
||||
match work.describe_task_definition(&arn).await {
|
||||
Ok(Some(d))
|
||||
if d.owner.as_deref() == Some(req.owner_user_id)
|
||||
&& d.project.as_deref() == Some(req.project_id) =>
|
||||
{
|
||||
match rt.deregister_task_definition(&d.arn).await {
|
||||
Ok(()) => retired.push(d.arn),
|
||||
Err(e) => failures.push(format!("task definition {}: {e}", d.arn)),
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => failures.push(format!("task definition {arn}: {e}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => failures.push(format!("task definitions: {e}")),
|
||||
}
|
||||
// DeleteTaskDefinitions takes at most 10 at a time.
|
||||
for chunk in retired.chunks(10) {
|
||||
if let Err(e) = rt.delete_task_definitions(chunk).await {
|
||||
failures.push(format!("deleting retired task definitions: {e}"));
|
||||
}
|
||||
}
|
||||
|
||||
// 3. The secret.
|
||||
let secret = project_secret_name(req.env, req.owner_key, req.project_id);
|
||||
if let Err(e) = rt.delete_secret(&secret).await {
|
||||
failures.push(format!("secret {secret}: {e}"));
|
||||
}
|
||||
|
||||
// 4. The access point, only when asked.
|
||||
if req.discard_data {
|
||||
let mut ids: Vec<String> = req.stored_access_point_id.map(String::from).into_iter().collect();
|
||||
let token = user_file_system_token(req.env, req.owner_key);
|
||||
match work.find_file_system(&token).await {
|
||||
Ok(Some(fs)) if fs.owner.as_deref() == Some(req.owner_user_id) => {
|
||||
let root = access_point_root(req.owner_key, req.project_id);
|
||||
match work.access_points(&fs.id).await {
|
||||
Ok(points) => ids.extend(
|
||||
points
|
||||
.into_iter()
|
||||
.filter(|a| {
|
||||
a.file_system_id == fs.id
|
||||
&& a.root_path.as_deref() == Some(root.as_str())
|
||||
&& a.owner.as_deref() == Some(req.owner_user_id)
|
||||
&& a.project.as_deref() == Some(req.project_id)
|
||||
})
|
||||
.map(|a| a.id),
|
||||
),
|
||||
Err(e) => failures.push(format!("access points: {e}")),
|
||||
}
|
||||
}
|
||||
Ok(_) => {}
|
||||
Err(e) => failures.push(format!("file system: {e}")),
|
||||
}
|
||||
ids.sort();
|
||||
ids.dedup();
|
||||
for id in ids {
|
||||
if let Err(e) = rt.delete_access_point(&id).await {
|
||||
failures.push(format!("access point {id}: {e}"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if failures.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(other(format!(
|
||||
"Some of the project's AWS resources in {} could not be removed: {}",
|
||||
req.region,
|
||||
failures.join("; ")
|
||||
)))
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------- real API
|
||||
|
||||
pub struct SdkRuntimeApi {
|
||||
ssm: aws_sdk_ssm::Client,
|
||||
iam: aws_sdk_iam::Client,
|
||||
secrets: aws_sdk_secretsmanager::Client,
|
||||
efs: aws_sdk_efs::Client,
|
||||
ecs: aws_sdk_ecs::Client,
|
||||
}
|
||||
|
||||
impl SdkRuntimeApi {
|
||||
pub fn new(cfg: &SdkConfig) -> Self {
|
||||
SdkRuntimeApi {
|
||||
ssm: aws_sdk_ssm::Client::new(cfg),
|
||||
iam: aws_sdk_iam::Client::new(cfg),
|
||||
secrets: aws_sdk_secretsmanager::Client::new(cfg),
|
||||
efs: aws_sdk_efs::Client::new(cfg),
|
||||
ecs: aws_sdk_ecs::Client::new(cfg),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeApi for SdkRuntimeApi {
|
||||
async fn environment_record(&self, env: &str) -> Result<Option<String>, AwsError> {
|
||||
const ACTION: &str = "ssm:GetParameter";
|
||||
match self
|
||||
.ssm
|
||||
.get_parameter()
|
||||
.name(environment_parameter_name(env))
|
||||
.with_decryption(false)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(out) => Ok(out.parameter().and_then(|p| p.value()).map(str::to_string)),
|
||||
Err(e)
|
||||
if e.as_service_error()
|
||||
.is_some_and(|s| s.is_parameter_not_found()) =>
|
||||
{
|
||||
Ok(None)
|
||||
}
|
||||
Err(e) => Err(classify_sdk_error(ACTION, &e)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn role_exists(&self, name: &str) -> Result<Knowable, AwsError> {
|
||||
let res = self.iam.get_role().role_name(name).send().await;
|
||||
knowable("iam:GetRole", res)
|
||||
}
|
||||
|
||||
async fn secret_string(&self, name: &str) -> Result<String, AwsError> {
|
||||
const ACTION: &str = "secretsmanager:GetSecretValue";
|
||||
let out = self
|
||||
.secrets
|
||||
.get_secret_value()
|
||||
.secret_id(name)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| classify_sdk_error(ACTION, &e))?;
|
||||
out.secret_string()
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| other(format!("{ACTION} returned no SecretString")))
|
||||
}
|
||||
|
||||
async fn access_point_state(&self, id: &str) -> Result<Option<String>, AwsError> {
|
||||
const ACTION: &str = "elasticfilesystem:DescribeAccessPoints";
|
||||
match self.efs.describe_access_points().access_point_id(id).send().await {
|
||||
Ok(out) => Ok(out
|
||||
.access_points()
|
||||
.iter()
|
||||
.find(|a| a.access_point_id() == Some(id))
|
||||
.and_then(|a| a.life_cycle_state())
|
||||
.map(|s| s.as_str().to_string())),
|
||||
Err(e)
|
||||
if e.as_service_error()
|
||||
.is_some_and(|s| s.is_access_point_not_found()) =>
|
||||
{
|
||||
Ok(None)
|
||||
}
|
||||
Err(e) => Err(classify_sdk_error(ACTION, &e)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn deregister_task_definition(&self, arn: &str) -> Result<(), AwsError> {
|
||||
const ACTION: &str = "ecs:DeregisterTaskDefinition";
|
||||
self.ecs
|
||||
.deregister_task_definition()
|
||||
.task_definition(arn)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| classify_sdk_error(ACTION, &e))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn delete_task_definitions(&self, arns: &[String]) -> Result<(), AwsError> {
|
||||
const ACTION: &str = "ecs:DeleteTaskDefinitions";
|
||||
let out = self
|
||||
.ecs
|
||||
.delete_task_definitions()
|
||||
.set_task_definitions(Some(arns.to_vec()))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| classify_sdk_error(ACTION, &e))?;
|
||||
let failed: Vec<String> = out
|
||||
.failures()
|
||||
.iter()
|
||||
.map(|f| {
|
||||
format!(
|
||||
"{} ({})",
|
||||
f.arn().unwrap_or("?"),
|
||||
f.reason().unwrap_or("no reason given")
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
if failed.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(other(format!("{ACTION} failed for {}", failed.join(", "))))
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_secret(&self, name: &str) -> Result<(), AwsError> {
|
||||
const ACTION: &str = "secretsmanager:DeleteSecret";
|
||||
match self
|
||||
.secrets
|
||||
.delete_secret()
|
||||
.secret_id(name)
|
||||
.force_delete_without_recovery(false)
|
||||
.recovery_window_in_days(SECRET_RECOVERY_DAYS)
|
||||
.send()
|
||||
.await
|
||||
{
|
||||
Ok(_) => Ok(()),
|
||||
Err(e)
|
||||
if e.as_service_error()
|
||||
.is_some_and(|s| s.is_resource_not_found_exception()) =>
|
||||
{
|
||||
Ok(())
|
||||
}
|
||||
// Already scheduled: the goal.
|
||||
Err(e)
|
||||
if e.as_service_error().is_some_and(|s| {
|
||||
s.is_invalid_request_exception()
|
||||
&& s.message().is_some_and(|m| m.contains("scheduled for deletion"))
|
||||
}) =>
|
||||
{
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => Err(classify_sdk_error(ACTION, &e)),
|
||||
}
|
||||
}
|
||||
|
||||
async fn delete_access_point(&self, id: &str) -> Result<(), AwsError> {
|
||||
const ACTION: &str = "elasticfilesystem:DeleteAccessPoint";
|
||||
match self.efs.delete_access_point().access_point_id(id).send().await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(e)
|
||||
if e.as_service_error()
|
||||
.is_some_and(|s| s.is_access_point_not_found()) =>
|
||||
{
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => Err(classify_sdk_error(ACTION, &e)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
@@ -0,0 +1,341 @@
|
||||
use std::sync::Mutex;
|
||||
|
||||
use aws_sdk_ecs::operation::register_task_definition::RegisterTaskDefinitionInput;
|
||||
use aws_sdk_ecs::operation::run_task::RunTaskInput;
|
||||
|
||||
use super::*;
|
||||
use crate::aws::tasks::TaskInfo;
|
||||
use crate::aws::tofu_modules::owner_key;
|
||||
use crate::aws::workspace::{AccessPointRef, FileSystemRef, TaskDefinitionRef};
|
||||
|
||||
const REGION: &str = "eu-west-1";
|
||||
const ACCOUNT: &str = "111122223333";
|
||||
const USER: &str = "AROAEXAMPLEID:alice@example.com";
|
||||
const PROJECT: &str = "0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0";
|
||||
|
||||
fn ok() -> String {
|
||||
owner_key(REGION, USER)
|
||||
}
|
||||
|
||||
fn family() -> String {
|
||||
task_family("dev", &ok(), PROJECT)
|
||||
}
|
||||
|
||||
fn task_arn(id: &str) -> String {
|
||||
format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task/triple-c-dev/{id}")
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Fake {
|
||||
calls: Mutex<Vec<String>>,
|
||||
record: Option<String>,
|
||||
role: Option<Knowable>,
|
||||
secret: Option<String>,
|
||||
ap_states: Mutex<Vec<Option<String>>>,
|
||||
live_tasks: Vec<TaskInfo>,
|
||||
revisions: Vec<TaskDefinitionRef>,
|
||||
access_points: Vec<AccessPointRef>,
|
||||
fail_secret_delete: bool,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
fn log(&self, s: String) {
|
||||
self.calls.lock().unwrap().push(s);
|
||||
}
|
||||
fn calls(&self) -> Vec<String> {
|
||||
self.calls.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl RuntimeApi for Fake {
|
||||
async fn environment_record(&self, env: &str) -> Result<Option<String>, AwsError> {
|
||||
self.log(format!("record {env}"));
|
||||
Ok(self.record.clone())
|
||||
}
|
||||
async fn role_exists(&self, name: &str) -> Result<Knowable, AwsError> {
|
||||
self.log(format!("role {name}"));
|
||||
Ok(self.role.clone().unwrap_or(Knowable::Yes))
|
||||
}
|
||||
async fn secret_string(&self, name: &str) -> Result<String, AwsError> {
|
||||
self.log(format!("secret {name}"));
|
||||
self.secret.clone().ok_or_else(|| other("no secret"))
|
||||
}
|
||||
async fn access_point_state(&self, id: &str) -> Result<Option<String>, AwsError> {
|
||||
self.log(format!("ap {id}"));
|
||||
let mut s = self.ap_states.lock().unwrap();
|
||||
Ok(if s.len() > 1 { s.remove(0) } else { s.first().cloned().flatten() })
|
||||
}
|
||||
async fn deregister_task_definition(&self, arn: &str) -> Result<(), AwsError> {
|
||||
self.log(format!("deregister {arn}"));
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_task_definitions(&self, arns: &[String]) -> Result<(), AwsError> {
|
||||
self.log(format!("delete-td {}", arns.join(",")));
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_secret(&self, name: &str) -> Result<(), AwsError> {
|
||||
self.log(format!("delete-secret {name}"));
|
||||
if self.fail_secret_delete {
|
||||
return Err(other("denied"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
async fn delete_access_point(&self, id: &str) -> Result<(), AwsError> {
|
||||
self.log(format!("delete-ap {id}"));
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl TaskApi for Fake {
|
||||
async fn run_task(&self, _req: RunTaskInput) -> Result<TaskInfo, AwsError> {
|
||||
unreachable!("removal never runs a task")
|
||||
}
|
||||
async fn describe(&self, _c: &str, arns: &[String]) -> Result<Vec<TaskInfo>, AwsError> {
|
||||
Ok(self.live_tasks.iter().filter(|t| arns.contains(&t.arn)).cloned().collect())
|
||||
}
|
||||
async fn list_project_tasks(&self, _c: &str, _f: &str) -> Result<Vec<String>, AwsError> {
|
||||
Ok(self.live_tasks.iter().map(|t| t.arn.clone()).collect())
|
||||
}
|
||||
async fn stop_task(&self, cluster: &str, arn: &str, reason: &str) -> Result<(), AwsError> {
|
||||
self.log(format!("stop {cluster} {arn} {reason}"));
|
||||
Ok(())
|
||||
}
|
||||
async fn log_tail(&self, _g: &str, _s: &str, _l: i32) -> Result<Vec<String>, AwsError> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkspaceApi for Fake {
|
||||
async fn find_file_system(&self, _token: &str) -> Result<Option<FileSystemRef>, AwsError> {
|
||||
Ok(Some(FileSystemRef { id: "fs-1".into(), owner: Some(USER.into()) }))
|
||||
}
|
||||
async fn access_points(&self, _fs: &str) -> Result<Vec<AccessPointRef>, AwsError> {
|
||||
Ok(self.access_points.clone())
|
||||
}
|
||||
async fn create_access_point(
|
||||
&self,
|
||||
_fs: &str,
|
||||
_root: &str,
|
||||
_tags: &[(String, String)],
|
||||
) -> Result<String, AwsError> {
|
||||
unreachable!()
|
||||
}
|
||||
async fn put_project_secret(
|
||||
&self,
|
||||
_n: &str,
|
||||
_k: &str,
|
||||
_j: &str,
|
||||
_t: &[(String, String)],
|
||||
) -> Result<String, AwsError> {
|
||||
unreachable!()
|
||||
}
|
||||
async fn task_definition_revisions(&self, _f: &str) -> Result<Vec<String>, AwsError> {
|
||||
Ok(self.revisions.iter().map(|r| r.arn.clone()).collect())
|
||||
}
|
||||
async fn describe_task_definition(
|
||||
&self,
|
||||
arn: &str,
|
||||
) -> Result<Option<TaskDefinitionRef>, AwsError> {
|
||||
Ok(self.revisions.iter().find(|r| r.arn == arn).cloned())
|
||||
}
|
||||
async fn register_task_definition(
|
||||
&self,
|
||||
_req: RegisterTaskDefinitionInput,
|
||||
) -> Result<String, AwsError> {
|
||||
unreachable!()
|
||||
}
|
||||
}
|
||||
|
||||
fn record() -> String {
|
||||
serde_json::json!({
|
||||
"schema": 1, "name": "dev", "template_version": 1,
|
||||
"cluster_arn": format!("arn:aws:ecs:{REGION}:{ACCOUNT}:cluster/triple-c-dev"),
|
||||
"security_group_id": "sg-0123456789abcdef0",
|
||||
"subnet_ids": ["subnet-0a", "subnet-0b"],
|
||||
"assign_public_ip": true,
|
||||
"kms_key_arn": format!("arn:aws:kms:{REGION}:{ACCOUNT}:key/1234-abcd"),
|
||||
"log_group": "/triple-c/dev", "network_mode": "public", "tenancy": "strict"
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_environment_is_read_with_one_lookup_and_validated() {
|
||||
let fake = Fake { record: Some(record()), ..Default::default() };
|
||||
let env = read_environment(&fake, "dev", ACCOUNT, REGION).await.unwrap();
|
||||
assert_eq!(env.log_group, "/triple-c/dev");
|
||||
assert_eq!(fake.calls(), vec!["record dev"]);
|
||||
|
||||
let missing = Fake::default();
|
||||
let e = read_environment(&missing, "dev", ACCOUNT, REGION).await.unwrap_err();
|
||||
assert!(e.to_string().contains("was not found"), "{e}");
|
||||
|
||||
// A record from another account is refused, by field name only.
|
||||
let foreign = Fake { record: Some(record()), ..Default::default() };
|
||||
let e = read_environment(&foreign, "dev", "999999999999", REGION).await.unwrap_err();
|
||||
assert!(e.to_string().contains("cluster_arn"), "{e}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_missing_task_role_means_not_joined_and_a_denied_lookup_lets_the_start_try() {
|
||||
let no = Fake { role: Some(Knowable::No), ..Default::default() };
|
||||
let e = check_onboarded(&no, "dev", &ok()).await.unwrap_err();
|
||||
assert!(e.to_string().contains("not joined"), "{e}");
|
||||
assert_eq!(no.calls(), vec![format!("role triple-c-dev-{}-task", ok())]);
|
||||
let unknown = Fake { role: Some(Knowable::Unknown), ..Default::default() };
|
||||
assert!(check_onboarded(&unknown, "dev", &ok()).await.is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_agent_token_is_read_from_the_secret_without_echoing_it() {
|
||||
let json = r#"{"TRIPLEC_AGENT_TOKEN":"abc-DEF_123","OTHER":"SENTINEL-VALUE-7f3a"}"#;
|
||||
assert_eq!(agent_token_from_secret(json).unwrap(), "abc-DEF_123");
|
||||
for bad in [
|
||||
r#"{"OTHER":"SENTINEL-VALUE-7f3a"}"#,
|
||||
r#"{"TRIPLEC_AGENT_TOKEN":""}"#,
|
||||
r#"{"TRIPLEC_AGENT_TOKEN":"has space"}"#,
|
||||
"SENTINEL-VALUE-7f3a",
|
||||
] {
|
||||
let e = agent_token_from_secret(bad).unwrap_err();
|
||||
assert!(!e.contains("SENTINEL") && !e.contains("space"), "{e}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_token_comes_from_the_projects_own_secret() {
|
||||
let fake = Fake {
|
||||
secret: Some(r#"{"TRIPLEC_AGENT_TOKEN":"tok"}"#.into()),
|
||||
..Default::default()
|
||||
};
|
||||
assert_eq!(read_agent_token(&fake, "dev", &ok(), PROJECT).await.unwrap(), "tok");
|
||||
assert_eq!(fake.calls(), vec![format!("secret triple-c/dev/{}/{PROJECT}", ok())]);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn a_new_access_point_is_waited_for_until_available() {
|
||||
let fake = Fake {
|
||||
ap_states: Mutex::new(vec![Some("creating".into()), Some("creating".into()), Some("available".into())]),
|
||||
..Default::default()
|
||||
};
|
||||
wait_access_point(&fake, "fsap-1", PollTiming::default()).await.unwrap();
|
||||
assert_eq!(fake.calls().len(), 3);
|
||||
|
||||
let stuck = Fake { ap_states: Mutex::new(vec![Some("creating".into())]), ..Default::default() };
|
||||
let e = wait_access_point(&stuck, "fsap-1", PollTiming::default()).await.unwrap_err();
|
||||
assert!(e.to_string().contains("did not become available"), "{e}");
|
||||
|
||||
let gone = Fake { ap_states: Mutex::new(vec![Some("deleting".into())]), ..Default::default() };
|
||||
assert!(wait_access_point(&gone, "fsap-1", PollTiming::default()).await.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn task_arns_name_their_cluster() {
|
||||
let arn = task_arn("abc123");
|
||||
assert_eq!(cluster_of_task(&arn), Some("triple-c-dev"));
|
||||
assert_eq!(
|
||||
cluster_arn_of_task(&arn).as_deref(),
|
||||
Some(format!("arn:aws:ecs:{REGION}:{ACCOUNT}:cluster/triple-c-dev").as_str())
|
||||
);
|
||||
assert_eq!(cluster_of_task("arn:aws:ecs:r:a:task/abc"), None, "old format, no cluster");
|
||||
assert_eq!(cluster_of_task("task/triple-c-dev/abc"), None);
|
||||
assert_eq!(cluster_of_task("arn:aws:ecs:r:a:task/c/a/b"), None);
|
||||
}
|
||||
|
||||
fn live(id: &str) -> TaskInfo {
|
||||
TaskInfo {
|
||||
arn: task_arn(id),
|
||||
last_status: "RUNNING".into(),
|
||||
desired_status: "RUNNING".into(),
|
||||
health: None,
|
||||
stop_code: None,
|
||||
stopped_reason: None,
|
||||
task_definition_arn: format!(
|
||||
"arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:2",
|
||||
family()
|
||||
),
|
||||
runtime_id: None,
|
||||
exec_agent_running: false,
|
||||
owner: Some(USER.into()),
|
||||
project: Some(PROJECT.into()),
|
||||
}
|
||||
}
|
||||
|
||||
fn rev(n: u32, owner: &str) -> TaskDefinitionRef {
|
||||
TaskDefinitionRef {
|
||||
arn: format!("arn:aws:ecs:{REGION}:{ACCOUNT}:task-definition/{}:{n}", family()),
|
||||
family: family(),
|
||||
owner: Some(owner.into()),
|
||||
project: Some(PROJECT.into()),
|
||||
fingerprint: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn removal_stops_the_task_retires_own_revisions_and_schedules_the_secret() {
|
||||
let fake = Fake {
|
||||
live_tasks: vec![live("t1")],
|
||||
revisions: vec![rev(2, USER), rev(3, "AROAOTHER:mallory"), rev(1, USER)],
|
||||
access_points: vec![AccessPointRef {
|
||||
id: "fsap-found".into(),
|
||||
file_system_id: "fs-1".into(),
|
||||
root_path: Some(access_point_root(&ok(), PROJECT)),
|
||||
owner: Some(USER.into()),
|
||||
project: Some(PROJECT.into()),
|
||||
state: Some("available".into()),
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let key = ok();
|
||||
let t0 = task_arn("t0");
|
||||
let mut r = RemoveRequest {
|
||||
env: "dev",
|
||||
region: REGION,
|
||||
owner_user_id: USER,
|
||||
owner_key: &key,
|
||||
project_id: PROJECT,
|
||||
stored_task_arn: Some(&t0),
|
||||
stored_access_point_id: Some("fsap-stored"),
|
||||
discard_data: false,
|
||||
};
|
||||
remove_workspace_resources(&fake, &fake, &fake, &r).await.unwrap();
|
||||
let calls = fake.calls();
|
||||
let secret = project_secret_name("dev", &ok(), PROJECT);
|
||||
assert_eq!(
|
||||
calls,
|
||||
vec![
|
||||
format!("stop triple-c-dev {} {STOP_REASON}", task_arn("t1")),
|
||||
format!("stop triple-c-dev {} {STOP_REASON}", task_arn("t0")),
|
||||
format!("deregister {}", rev(2, USER).arn),
|
||||
format!("deregister {}", rev(1, USER).arn),
|
||||
format!("delete-td {},{}", rev(2, USER).arn, rev(1, USER).arn),
|
||||
format!("delete-secret {secret}"),
|
||||
],
|
||||
"another user's revision is left alone and the access point is kept"
|
||||
);
|
||||
|
||||
let keep = Fake { access_points: fake.access_points.clone(), ..Default::default() };
|
||||
r.discard_data = true;
|
||||
remove_workspace_resources(&keep, &keep, &keep, &r).await.unwrap();
|
||||
let calls = keep.calls();
|
||||
assert!(calls.contains(&"delete-ap fsap-found".to_string()), "{calls:?}");
|
||||
assert!(calls.contains(&"delete-ap fsap-stored".to_string()), "{calls:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_failed_step_is_reported_but_the_rest_still_runs() {
|
||||
let fake = Fake { fail_secret_delete: true, ..Default::default() };
|
||||
let key = ok();
|
||||
let r = RemoveRequest {
|
||||
env: "dev",
|
||||
region: REGION,
|
||||
owner_user_id: USER,
|
||||
owner_key: &key,
|
||||
project_id: PROJECT,
|
||||
stored_task_arn: None,
|
||||
stored_access_point_id: Some("fsap-1"),
|
||||
discard_data: true,
|
||||
};
|
||||
let e = remove_workspace_resources(&fake, &fake, &fake, &r).await.unwrap_err();
|
||||
assert!(e.to_string().contains("secret"), "{e}");
|
||||
assert!(fake.calls().contains(&"delete-ap fsap-1".to_string()));
|
||||
}
|
||||
Reference in new issue
Block a user