use aws_config::{Region, SdkConfig};
use aws_sdk_acm::Client as AcmClient;
use aws_sdk_cloudwatch::Client as CwClient;
use aws_sdk_cloudwatchlogs::Client as CwLogsClient;
use aws_sdk_costexplorer::Client as CostExplorerClient;
use aws_sdk_ec2::Client as Ec2Client;
use aws_sdk_elasticbeanstalk::Client;
use aws_sdk_iam::Client as IamClient;
use aws_sdk_organizations::Client as OrgClient;
use aws_sdk_s3::Client as S3Client;
use aws_sdk_secretsmanager::Client as SecretsClient;
use aws_sdk_sqs::Client as SqsClient;
use aws_sdk_sts::Client as StsClient;
use chrono::{DateTime, Utc};
use color_eyre::eyre::{eyre, Result, WrapErr};
#[derive(Clone, Debug)]
pub struct Event {
pub at: Option<DateTime<Utc>>,
pub env: String,
pub application: String,
pub message: String,
pub severity: String,
pub version_label: Option<String>,
}
#[derive(Clone, Debug)]
pub struct CwAlarm {
pub name: String,
pub state: String, pub state_reason: String,
pub metric_name: String,
pub namespace: String,
}
#[derive(Clone, Debug, Default)]
pub struct MetricSeries {
pub id: String, pub label: String, pub points: Vec<(DateTime<Utc>, f64)>,
}
#[derive(Clone, Debug, Default)]
pub struct WorkerQueues {
pub main_url: Option<String>,
pub dlq_url: Option<String>,
pub main_stats: Option<QueueStats>,
pub dlq_stats: Option<QueueStats>,
}
#[derive(Clone, Debug, Default)]
pub struct QueueStats {
pub visible: i64,
pub in_flight: i64,
pub delayed: i64,
}
#[derive(Clone, Debug)]
pub struct QueueMessage {
pub id: String,
pub receipt_handle: String,
pub body: String,
pub receive_count: i64,
pub sent_at: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug)]
pub struct AcmCert {
pub arn: String,
pub domain: String,
}
#[derive(Clone, Debug)]
pub struct Instance {
pub id: String,
pub health: String, pub color: String, pub causes: Vec<String>,
pub instance_type: String,
pub availability_zone: String,
pub launched_at: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug)]
pub struct Application {
pub name: String,
pub description: String,
pub date_created: Option<DateTime<Utc>>,
pub date_updated: Option<DateTime<Utc>>,
pub version_count: usize,
pub templates: Vec<String>,
pub latest_version_label: Option<String>,
pub latest_version_created: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug, Default)]
pub struct EnvVpcContext {
pub vpc_id: Option<String>,
pub subnets: Vec<String>,
pub elb_subnets: Vec<String>,
pub security_groups: Vec<String>,
}
#[derive(Clone, Debug)]
pub struct SubnetInfo {
pub id: String,
pub availability_zone: String,
pub cidr_block: String,
pub name_tag: Option<String>,
}
#[derive(Clone, Debug)]
pub struct SecurityGroupInfo {
pub id: String,
pub group_name: String,
pub description: String,
}
#[derive(Clone, Debug)]
pub struct CustomPlatform {
pub arn: String,
pub branch: String,
pub version: String,
pub status: String,
pub lifecycle: String,
}
#[derive(Clone, Debug)]
pub struct AppVersion {
pub label: String,
pub description: String,
pub created: Option<DateTime<Utc>>,
}
#[derive(Clone, Debug)]
pub struct Environment {
pub name: String,
pub application: String,
pub status: String,
pub health: String,
pub platform: String, pub solution_stack: String,
pub tier: String, pub cname: String,
pub version_label: String,
pub arn: Option<String>,
pub updated: Option<DateTime<Utc>>,
pub id: Option<String>,
pub region: Option<String>,
}
#[derive(Clone, Debug)]
pub struct AwsContext {
pub region: String,
pub profile: Option<String>,
pub account_id: Option<String>,
pub caller_arn: Option<String>,
}
pub type CustomMetricQuery = (String, String, String, String, Vec<(String, String)>);
#[derive(Clone, Debug)]
pub struct LogEvent {
pub timestamp_ms: i64,
pub stream: String,
pub message: String,
}
#[derive(Clone, Debug)]
pub struct InsightsRow {
pub fields: Vec<(String, String)>,
}
#[derive(Clone, Debug)]
pub struct InsightsResults {
pub rows: Vec<InsightsRow>,
pub records_scanned: i64,
pub records_matched: i64,
}
#[derive(Clone, Debug)]
pub struct Identity {
pub account_id: Option<String>,
pub caller_arn: Option<String>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct EnvInstanceCounts {
pub healthy: i32,
pub total: i32,
}
pub struct AwsClient {
client: Client,
sqs: SqsClient,
cw: CwClient,
cw_logs: CwLogsClient,
s3: S3Client,
ec2: Ec2Client,
org: OrgClient,
cost: CostExplorerClient,
iam: IamClient,
secrets: SecretsClient,
acm: AcmClient,
config: SdkConfig,
pub context: AwsContext,
}
#[derive(Clone, Debug)]
pub struct OrgAccount {
pub id: String,
pub name: String,
pub email: Option<String>,
pub status: String,
}
fn cost_explorer_client(base: &SdkConfig) -> CostExplorerClient {
let cfg = base.to_builder().region(Region::new("us-east-1")).build();
CostExplorerClient::new(&cfg)
}
fn iam_client(base: &SdkConfig) -> IamClient {
let cfg = base.to_builder().region(Region::new("us-east-1")).build();
IamClient::new(&cfg)
}
#[derive(Clone, Debug)]
pub struct SecretSummary {
pub name: String,
pub arn: String,
pub description: Option<String>,
pub last_changed: Option<DateTime<Utc>>,
pub last_rotated: Option<DateTime<Utc>>,
pub kms_key_id: Option<String>,
}
#[derive(Clone, Debug)]
pub struct IamSimResult {
pub action: String,
pub resource: String,
pub decision: String,
pub matched_statements: Vec<String>,
pub missing_context: Vec<String>,
pub blocked_by_scp: bool,
pub blocked_by_boundary: bool,
}
#[derive(Clone, Debug, Default)]
pub struct EnvResources {
pub asgs: Vec<String>,
pub instances: Vec<String>,
pub launch_configs: Vec<String>,
pub launch_templates: Vec<String>,
pub load_balancers: Vec<String>,
pub triggers: Vec<String>,
pub queues: Vec<EnvResourceQueue>,
}
#[derive(Clone, Debug)]
pub struct EnvResourceQueue {
pub name: String,
pub url: String,
}
#[derive(Clone, Debug)]
pub struct ConfigOption {
pub namespace: String,
pub name: String,
pub value: Option<String>,
pub default_value: Option<String>,
pub value_type: String,
pub value_options: Vec<String>,
pub change_severity: Option<String>,
#[allow(dead_code)]
pub user_defined: Option<bool>,
pub min_value: Option<i32>,
pub max_value: Option<i32>,
pub max_length: Option<i32>,
}
#[derive(Clone, Debug)]
pub struct EnvCost {
pub env_name: String,
pub cost_usd: f64,
}
impl AwsClient {
pub async fn with(profile: Option<String>, region: Option<String>) -> Result<Self> {
let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
if let Some(p) = profile.clone() {
builder = builder.profile_name(p);
}
if let Some(r) = region.clone() {
builder = builder.region(Region::new(r));
}
let config = builder.load().await;
let resolved_region = config
.region()
.map(|r| r.as_ref().to_string())
.unwrap_or_else(|| "unknown".to_string());
if region.as_deref().is_some_and(|r| r != resolved_region) {
tracing::warn!(
target: "ebman::aws",
requested = ?region,
resolved = %resolved_region,
env_aws_region = ?std::env::var("AWS_REGION").ok(),
env_aws_default_region = ?std::env::var("AWS_DEFAULT_REGION").ok(),
"AwsClient::with region mismatch — explicit override was ignored by SDK"
);
}
let region = resolved_region;
let profile = profile.or_else(|| std::env::var("AWS_PROFILE").ok());
let client = Client::new(&config);
let sqs = SqsClient::new(&config);
let cw = CwClient::new(&config);
let cw_logs = CwLogsClient::new(&config);
let s3 = S3Client::new(&config);
let ec2 = Ec2Client::new(&config);
let org = OrgClient::new(&config);
let cost = cost_explorer_client(&config);
let iam = iam_client(&config);
let secrets = SecretsClient::new(&config);
Ok(Self {
client,
sqs,
cw,
cw_logs,
s3,
ec2,
org,
cost,
iam,
secrets,
acm: AcmClient::new(&config),
config,
context: AwsContext {
region,
profile,
account_id: None,
caller_arn: None,
},
})
}
pub async fn assume_role(target_name: &str, spec: &crate::config::AccountSpec) -> Result<Self> {
let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
if let Some(p) = spec.source_profile.as_ref() {
builder = builder.profile_name(p.clone());
}
if let Some(r) = spec.region.clone() {
builder = builder.region(Region::new(r));
}
let base_config = builder.load().await;
let sts = StsClient::new(&base_config);
let session_name = format!("ebman-{target_name}");
let mut req = sts
.assume_role()
.role_arn(spec.role_arn.clone())
.role_session_name(session_name);
if let Some(eid) = spec.external_id.as_ref() {
req = req.external_id(eid.clone());
}
let resp = req.send().await.wrap_err("sts:AssumeRole failed")?;
let creds = resp
.credentials
.ok_or_else(|| eyre!("sts:AssumeRole returned no credentials"))?;
let access_key = creds.access_key_id;
let secret_key = creds.secret_access_key;
let session_token = creds.session_token;
let aws_creds = aws_credential_types::Credentials::new(
access_key,
secret_key,
Some(session_token),
std::time::SystemTime::UNIX_EPOCH.checked_add(std::time::Duration::from_secs(
creds.expiration.secs() as u64,
)),
"ebman-assume-role",
);
let mut builder = aws_config::defaults(aws_config::BehaviorVersion::latest());
builder = builder.credentials_provider(aws_creds);
if let Some(r) = spec.region.clone() {
builder = builder.region(Region::new(r));
} else if let Some(r) = base_config.region().cloned() {
builder = builder.region(r);
}
let config = builder.load().await;
let region = config
.region()
.map(|r| r.as_ref().to_string())
.unwrap_or_else(|| "unknown".to_string());
let cost = cost_explorer_client(&config);
let iam = iam_client(&config);
let secrets = SecretsClient::new(&config);
Ok(Self {
client: Client::new(&config),
sqs: SqsClient::new(&config),
cw: CwClient::new(&config),
cw_logs: CwLogsClient::new(&config),
s3: S3Client::new(&config),
ec2: Ec2Client::new(&config),
org: OrgClient::new(&config),
cost,
iam,
secrets,
acm: AcmClient::new(&config),
config,
context: AwsContext {
region,
profile: Some(target_name.to_string()),
account_id: None,
caller_arn: None,
},
})
}
#[cfg(test)]
pub fn stub() -> Self {
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
Self::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
)
}
#[cfg(test)]
pub fn for_tests(
client: Client,
sqs: SqsClient,
cw: CwClient,
cw_logs: CwLogsClient,
s3: S3Client,
ec2: Ec2Client,
) -> Self {
let config = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let org = OrgClient::new(&config);
let cost = cost_explorer_client(&config);
let iam = iam_client(&config);
let secrets = SecretsClient::new(&config);
Self {
client,
sqs,
cw,
cw_logs,
s3,
ec2,
org,
cost,
iam,
secrets,
acm: AcmClient::new(&config),
config,
context: AwsContext {
region: "us-east-1".to_string(),
profile: None,
account_id: None,
caller_arn: None,
},
}
}
pub async fn verify_identity(&self) -> Result<Identity> {
let ident = StsClient::new(&self.config)
.get_caller_identity()
.send()
.await
.wrap_err("sts get-caller-identity failed")?;
Ok(Identity {
account_id: ident.account,
caller_arn: ident.arn,
})
}
pub async fn list_events(&self, max: i32) -> Result<Vec<Event>> {
self.list_events_inner(None, max).await
}
pub async fn list_events_for_env(&self, env_name: &str, max: i32) -> Result<Vec<Event>> {
self.list_events_inner(Some(env_name.to_string()), max)
.await
}
async fn list_events_inner(&self, env_name: Option<String>, max: i32) -> Result<Vec<Event>> {
let mut req = self.client.describe_events().max_records(max);
if let Some(n) = env_name {
req = req.environment_name(n);
}
let resp = req.send().await?;
let events = resp
.events
.unwrap_or_default()
.into_iter()
.map(|e| Event {
at: e
.event_date
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
env: e.environment_name.unwrap_or_default(),
application: e.application_name.unwrap_or_default(),
message: e.message.unwrap_or_default(),
severity: e
.severity
.map(|s| s.as_str().to_string())
.unwrap_or_else(|| "INFO".to_string()),
version_label: e.version_label.filter(|v| !v.is_empty()),
})
.collect();
Ok(events)
}
pub async fn describe_env_resources(&self, env_name: &str) -> Result<EnvResources> {
let resp = self
.client
.describe_environment_resources()
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeEnvironmentResources failed")?;
let res = resp
.environment_resources
.ok_or_else(|| eyre!("no environment_resources in response"))?;
Ok(EnvResources {
asgs: res
.auto_scaling_groups
.unwrap_or_default()
.into_iter()
.filter_map(|a| a.name)
.collect(),
instances: res
.instances
.unwrap_or_default()
.into_iter()
.filter_map(|i| i.id)
.collect(),
launch_configs: res
.launch_configurations
.unwrap_or_default()
.into_iter()
.filter_map(|l| l.name)
.collect(),
launch_templates: res
.launch_templates
.unwrap_or_default()
.into_iter()
.filter_map(|l| l.id)
.collect(),
load_balancers: res
.load_balancers
.unwrap_or_default()
.into_iter()
.filter_map(|l| l.name)
.collect(),
triggers: res
.triggers
.unwrap_or_default()
.into_iter()
.filter_map(|t| t.name)
.collect(),
queues: res
.queues
.unwrap_or_default()
.into_iter()
.filter_map(|q| {
let name = q.name?;
Some(EnvResourceQueue {
name,
url: q.url.unwrap_or_default(),
})
})
.collect(),
})
}
pub async fn describe_worker_queues(
&self,
application_name: &str,
env_name: &str,
) -> Result<WorkerQueues> {
let mut main_url: Option<String> = None;
let mut dlq_url: Option<String> = None;
if let Ok(resp) = self
.client
.describe_environment_resources()
.environment_name(env_name)
.send()
.await
{
if let Some(res) = resp.environment_resources {
for q in res.queues.unwrap_or_default() {
let name = q.name.unwrap_or_default();
let url = q.url.unwrap_or_default();
if url.is_empty() {
continue;
}
match name.as_str() {
"WorkerQueue" => main_url = Some(url),
"WorkerDeadLetterQueue" => dlq_url = Some(url),
_ => {}
}
}
}
}
if main_url.is_none() || dlq_url.is_none() {
if let Ok(resp) = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
{
for setting in resp.configuration_settings.unwrap_or_default() {
for opt in setting.option_settings.unwrap_or_default() {
let ns = opt.namespace.unwrap_or_default();
let name = opt.option_name.unwrap_or_default();
if ns != "aws:elasticbeanstalk:sqsd" {
continue;
}
match name.as_str() {
"WorkerQueueURL" => {
let v = opt.value.unwrap_or_default();
if !v.is_empty() && main_url.is_none() {
main_url = Some(v);
}
}
"DeadLetterQueueURL" => {
let v = opt.value.unwrap_or_default();
if !v.is_empty() && dlq_url.is_none() {
dlq_url = Some(v);
}
}
_ => {}
}
}
}
}
}
if let (Some(main), None) = (&main_url, &dlq_url) {
dlq_url = derive_dlq_url(main);
}
let main_stats = if let Some(u) = &main_url {
self.queue_stats(u).await.ok()
} else {
None
};
let dlq_stats = if let Some(u) = &dlq_url {
self.queue_stats(u).await.ok()
} else {
None
};
Ok(WorkerQueues {
main_url,
dlq_url,
main_stats,
dlq_stats,
})
}
pub async fn queue_stats(&self, queue_url: &str) -> Result<QueueStats> {
use aws_sdk_sqs::types::QueueAttributeName as Q;
let resp = self
.sqs
.get_queue_attributes()
.queue_url(queue_url)
.attribute_names(Q::ApproximateNumberOfMessages)
.attribute_names(Q::ApproximateNumberOfMessagesNotVisible)
.attribute_names(Q::ApproximateNumberOfMessagesDelayed)
.send()
.await?;
let attrs = resp.attributes.unwrap_or_default();
let parse = |k: Q| -> i64 {
attrs
.get(&k)
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(0)
};
Ok(QueueStats {
visible: parse(Q::ApproximateNumberOfMessages),
in_flight: parse(Q::ApproximateNumberOfMessagesNotVisible),
delayed: parse(Q::ApproximateNumberOfMessagesDelayed),
})
}
pub async fn peek_messages(&self, queue_url: &str, max: i32) -> Result<Vec<QueueMessage>> {
use aws_sdk_sqs::types::MessageSystemAttributeName as M;
let target = max.clamp(1, 100) as usize;
let mut out: Vec<QueueMessage> = Vec::new();
let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
let mut empty_in_a_row = 0;
for _ in 0..((target / 10).max(1) + 4) {
if out.len() >= target {
break;
}
let resp = self
.sqs
.receive_message()
.queue_url(queue_url)
.max_number_of_messages(((target - out.len()).clamp(1, 10)) as i32)
.visibility_timeout(5)
.wait_time_seconds(1)
.message_system_attribute_names(M::ApproximateReceiveCount)
.message_system_attribute_names(M::SentTimestamp)
.send()
.await
.wrap_err("ReceiveMessage failed")?;
let batch = resp.messages.unwrap_or_default();
if batch.is_empty() {
empty_in_a_row += 1;
if empty_in_a_row >= 2 {
break;
}
continue;
}
empty_in_a_row = 0;
for m in batch {
let id = m.message_id.clone().unwrap_or_default();
if !id.is_empty() && !seen.insert(id.clone()) {
continue;
}
let attrs = m.attributes.unwrap_or_default();
let receive_count = attrs
.get(&M::ApproximateReceiveCount)
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(0);
let sent_at = attrs
.get(&M::SentTimestamp)
.and_then(|v| v.parse::<i64>().ok())
.and_then(DateTime::from_timestamp_millis);
out.push(QueueMessage {
id,
receipt_handle: m.receipt_handle.unwrap_or_default(),
body: m.body.unwrap_or_default(),
receive_count,
sent_at,
});
if out.len() >= target {
break;
}
}
}
Ok(out)
}
pub async fn send_message(&self, queue_url: &str, body: &str) -> Result<()> {
self.sqs
.send_message()
.queue_url(queue_url)
.message_body(body)
.send()
.await?;
Ok(())
}
pub async fn delete_message(&self, queue_url: &str, receipt_handle: &str) -> Result<()> {
self.sqs
.delete_message()
.queue_url(queue_url)
.receipt_handle(receipt_handle)
.send()
.await?;
Ok(())
}
pub async fn fetch_env_costs(&self) -> Result<Vec<EnvCost>> {
use aws_sdk_costexplorer::types::{DateInterval, GroupDefinition, GroupDefinitionType};
let now = chrono::Utc::now().date_naive();
let start = (now - chrono::Duration::days(30))
.format("%Y-%m-%d")
.to_string();
let end = now.format("%Y-%m-%d").to_string();
let time_period = DateInterval::builder()
.start(start)
.end(end)
.build()
.wrap_err("Cost Explorer DateInterval missing field")?;
let group_by = GroupDefinition::builder()
.r#type(GroupDefinitionType::Tag)
.key("elasticbeanstalk:environment-name")
.build();
let resp = self
.cost
.get_cost_and_usage()
.time_period(time_period)
.granularity(aws_sdk_costexplorer::types::Granularity::Monthly)
.metrics("UnblendedCost")
.group_by(group_by)
.send()
.await
.wrap_err("GetCostAndUsage failed")?;
let mut totals: std::collections::HashMap<String, f64> = std::collections::HashMap::new();
for period in resp.results_by_time.unwrap_or_default() {
for group in period.groups.unwrap_or_default() {
let raw_key = match group.keys.as_ref().and_then(|k| k.first()) {
Some(k) => k.clone(),
None => continue,
};
let env_name = match raw_key.split_once('$') {
Some((_, v)) if !v.is_empty() => v.to_string(),
_ => continue,
};
let amount: f64 = group
.metrics
.as_ref()
.and_then(|m| m.get("UnblendedCost"))
.and_then(|m| m.amount.as_deref())
.and_then(|s| s.parse().ok())
.unwrap_or(0.0);
*totals.entry(env_name).or_insert(0.0) += amount;
}
}
let mut out: Vec<EnvCost> = totals
.into_iter()
.map(|(env_name, cost_usd)| EnvCost { env_name, cost_usd })
.collect();
out.sort_by(|a, b| {
b.cost_usd
.partial_cmp(&a.cost_usd)
.unwrap_or(std::cmp::Ordering::Equal)
});
Ok(out)
}
pub async fn list_alarms_for_env(&self, env_name: &str) -> Result<Vec<CwAlarm>> {
let mut out = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self.cw.describe_alarms();
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("DescribeAlarms failed")?;
for a in resp.metric_alarms.unwrap_or_default() {
let dims = a.dimensions.clone().unwrap_or_default();
let touches = dims.iter().any(|d| d.value.as_deref() == Some(env_name));
if !touches {
continue;
}
out.push(CwAlarm {
name: a.alarm_name.unwrap_or_default(),
state: a
.state_value
.map(|s| s.as_str().to_string())
.unwrap_or_default(),
state_reason: a.state_reason.unwrap_or_default(),
metric_name: a.metric_name.unwrap_or_default(),
namespace: a.namespace.unwrap_or_default(),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
Ok(out)
}
#[allow(clippy::too_many_arguments)]
pub async fn put_env_metric_alarm(
&self,
alarm_name: &str,
env_name: &str,
metric_name: &str,
threshold: f64,
comparison_operator: &str,
period_secs: i32,
evaluation_periods: i32,
statistic: &str,
) -> Result<()> {
use aws_sdk_cloudwatch::types::{ComparisonOperator, Dimension, Statistic};
let op = ComparisonOperator::from(comparison_operator);
if op.as_str() != comparison_operator {
return Err(eyre!(
"unknown comparison operator '{comparison_operator}' \
(valid: GreaterThanThreshold, GreaterThanOrEqualToThreshold, \
LessThanThreshold, LessThanOrEqualToThreshold)"
));
}
let stat = Statistic::from(statistic);
if stat.as_str() != statistic {
return Err(eyre!(
"unknown statistic '{statistic}' (valid: Average, Sum, Maximum, Minimum, SampleCount)"
));
}
let dim = Dimension::builder()
.name("EnvironmentName")
.value(env_name)
.build();
self.cw
.put_metric_alarm()
.alarm_name(alarm_name)
.alarm_description(format!("ebman: {metric_name} alarm on {env_name}"))
.namespace("AWS/ElasticBeanstalk")
.metric_name(metric_name)
.dimensions(dim)
.comparison_operator(op)
.threshold(threshold)
.period(period_secs)
.evaluation_periods(evaluation_periods)
.statistic(stat)
.treat_missing_data("notBreaching")
.send()
.await
.wrap_err("PutMetricAlarm failed")?;
Ok(())
}
pub async fn fetch_env_option_settings(
&self,
application_name: &str,
env_name: &str,
) -> Result<Vec<(String, String, String)>> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(env) failed")?;
let out = resp
.configuration_settings
.unwrap_or_default()
.into_iter()
.flat_map(|c| c.option_settings.unwrap_or_default())
.map(|o| {
(
o.namespace.unwrap_or_default(),
o.option_name.unwrap_or_default(),
o.value.unwrap_or_default(),
)
})
.collect();
Ok(out)
}
pub async fn fetch_env_vpc_context(
&self,
application_name: &str,
env_name: &str,
) -> Result<EnvVpcContext> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(env) failed")?;
let mut ctx = EnvVpcContext::default();
for setting in resp.configuration_settings.unwrap_or_default() {
for opt in setting.option_settings.unwrap_or_default() {
let ns = opt.namespace.unwrap_or_default();
let name = opt.option_name.unwrap_or_default();
let value = opt.value.unwrap_or_default();
match (ns.as_str(), name.as_str()) {
("aws:ec2:vpc", "VPCId") if !value.is_empty() => {
ctx.vpc_id = Some(value);
}
("aws:ec2:vpc", "Subnets") if !value.is_empty() => {
ctx.subnets = split_csv(&value);
}
("aws:ec2:vpc", "ELBSubnets") if !value.is_empty() => {
ctx.elb_subnets = split_csv(&value);
}
("aws:autoscaling:launchconfiguration", "SecurityGroups")
if !value.is_empty() =>
{
ctx.security_groups = split_csv(&value);
}
_ => {}
}
}
}
Ok(ctx)
}
pub async fn list_subnets_in_vpc(&self, vpc_id: &str) -> Result<Vec<SubnetInfo>> {
use aws_sdk_ec2::types::Filter;
let resp = self
.ec2
.describe_subnets()
.filters(
Filter::builder()
.name("vpc-id")
.values(vpc_id.to_string())
.build(),
)
.send()
.await
.wrap_err("DescribeSubnets failed")?;
let mut out: Vec<SubnetInfo> = resp
.subnets
.unwrap_or_default()
.into_iter()
.map(|s| {
let name_tag = s.tags.as_ref().and_then(|tags| {
tags.iter()
.find(|t| t.key.as_deref() == Some("Name"))
.and_then(|t| t.value.clone())
});
SubnetInfo {
id: s.subnet_id.unwrap_or_default(),
availability_zone: s.availability_zone.unwrap_or_default(),
cidr_block: s.cidr_block.unwrap_or_default(),
name_tag,
}
})
.collect();
out.sort_by(|a, b| {
a.availability_zone
.cmp(&b.availability_zone)
.then(a.cidr_block.cmp(&b.cidr_block))
});
Ok(out)
}
pub async fn list_security_groups_in_vpc(
&self,
vpc_id: &str,
) -> Result<Vec<SecurityGroupInfo>> {
use aws_sdk_ec2::types::Filter;
let resp = self
.ec2
.describe_security_groups()
.filters(
Filter::builder()
.name("vpc-id")
.values(vpc_id.to_string())
.build(),
)
.send()
.await
.wrap_err("DescribeSecurityGroups failed")?;
let mut out: Vec<SecurityGroupInfo> = resp
.security_groups
.unwrap_or_default()
.into_iter()
.map(|g| SecurityGroupInfo {
id: g.group_id.unwrap_or_default(),
group_name: g.group_name.unwrap_or_default(),
description: g.description.unwrap_or_default(),
})
.collect();
out.sort_by(|a, b| a.group_name.cmp(&b.group_name));
Ok(out)
}
pub async fn fetch_env_rds_config(
&self,
application_name: &str,
env_name: &str,
) -> Result<Vec<(String, String)>> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(rds) failed")?;
let mut out: Vec<(String, String)> = resp
.configuration_settings
.unwrap_or_default()
.into_iter()
.flat_map(|c| c.option_settings.unwrap_or_default())
.filter_map(|o| {
let ns = o.namespace?;
if ns != "aws:rds:dbinstance" {
return None;
}
let opt = o.option_name?;
let value = o.value.unwrap_or_default();
if value.is_empty() {
return None;
}
Some((opt, value))
})
.collect();
out.sort();
Ok(out)
}
pub async fn list_secrets(&self, name_filter: Option<&str>) -> Result<Vec<SecretSummary>> {
let mut out: Vec<SecretSummary> = Vec::new();
let mut next: Option<String> = None;
loop {
let mut req = self.secrets.list_secrets();
if let Some(t) = next.take() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("ListSecrets failed")?;
for s in resp.secret_list.unwrap_or_default() {
let name = match s.name {
Some(n) if !n.is_empty() => n,
_ => continue,
};
if let Some(needle) = name_filter {
if !name.contains(needle) {
continue;
}
}
out.push(SecretSummary {
name,
arn: s.arn.unwrap_or_default(),
description: s.description.filter(|d| !d.is_empty()),
last_changed: s
.last_changed_date
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
last_rotated: s
.last_rotated_date
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
kms_key_id: s.kms_key_id.filter(|k| !k.is_empty()),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next = Some(t),
_ => break,
}
}
out.sort_by_key(|r| std::cmp::Reverse(r.last_changed));
Ok(out)
}
pub async fn fetch_secret_value(&self, secret_id: &str) -> Result<String> {
let resp = self
.secrets
.get_secret_value()
.secret_id(secret_id)
.send()
.await
.wrap_err("GetSecretValue failed")?;
if let Some(s) = resp.secret_string {
return Ok(s);
}
if let Some(b) = resp.secret_binary {
return Ok(format!("(binary, {} bytes — not shown)", b.as_ref().len()));
}
Ok(String::new())
}
pub async fn simulate_principal_policy(
&self,
principal_arn: &str,
action_names: &[String],
resource_arns: &[String],
) -> Result<Vec<IamSimResult>> {
if action_names.is_empty() {
return Ok(Vec::new());
}
let resources: Vec<String> = if resource_arns.is_empty() {
vec!["*".to_string()]
} else {
resource_arns.to_vec()
};
let mut req = self
.iam
.simulate_principal_policy()
.policy_source_arn(principal_arn);
for a in action_names {
req = req.action_names(a);
}
for r in &resources {
req = req.resource_arns(r);
}
let resp = req
.send()
.await
.wrap_err("SimulatePrincipalPolicy failed")?;
let mut out: Vec<IamSimResult> = Vec::new();
for r in resp.evaluation_results.unwrap_or_default() {
let action = r.eval_action_name;
let resource = r.eval_resource_name.unwrap_or_default();
let decision = r.eval_decision.as_str().to_string();
let matched_statements: Vec<String> = r
.matched_statements
.unwrap_or_default()
.into_iter()
.filter_map(|s| {
let policy = s.source_policy_id?;
let sid = s
.start_position
.as_ref()
.map(|p| format!("{}:{}", p.line, p.column))
.unwrap_or_else(|| "0:0".into());
Some(format!("{policy} @ {sid}"))
})
.collect();
let missing_context: Vec<String> = r.missing_context_values.unwrap_or_default();
let blocked_by_scp = r
.organizations_decision_detail
.as_ref()
.is_some_and(|d| !d.allowed_by_organizations);
let blocked_by_boundary = r
.permissions_boundary_decision_detail
.as_ref()
.is_some_and(|d| !d.allowed_by_permissions_boundary);
out.push(IamSimResult {
action,
resource,
decision,
matched_statements,
missing_context,
blocked_by_scp,
blocked_by_boundary,
});
}
Ok(out)
}
pub async fn fetch_env_configuration_options(
&self,
application_name: &str,
env_name: &str,
) -> Result<Vec<ConfigOption>> {
let vocab_fut = self
.client
.describe_configuration_options()
.environment_name(env_name)
.send();
let settings_fut = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send();
let (vocab_resp, settings_resp) = tokio::try_join!(
async {
vocab_fut
.await
.wrap_err("DescribeConfigurationOptions failed")
},
async {
settings_fut
.await
.wrap_err("DescribeConfigurationSettings(options) failed")
},
)?;
let mut current: std::collections::HashMap<(String, String), String> =
std::collections::HashMap::new();
for c in settings_resp.configuration_settings.unwrap_or_default() {
for o in c.option_settings.unwrap_or_default() {
if let (Some(ns), Some(name)) = (o.namespace, o.option_name) {
if let Some(v) = o.value {
if !v.is_empty() {
current.insert((ns, name), v);
}
}
}
}
}
let mut out: Vec<ConfigOption> = vocab_resp
.options
.unwrap_or_default()
.into_iter()
.filter_map(|o| {
let namespace = o.namespace?;
let name = o.name?;
let value = current.get(&(namespace.clone(), name.clone())).cloned();
Some(ConfigOption {
namespace,
name,
value,
default_value: o.default_value,
value_type: o
.value_type
.map(|v| v.as_str().to_string())
.unwrap_or_default(),
value_options: o.value_options.unwrap_or_default(),
change_severity: o.change_severity,
user_defined: o.user_defined,
min_value: o.min_value,
max_value: o.max_value,
max_length: o.max_length,
})
})
.collect();
out.sort_by(|a, b| {
let a_set = a.value.is_some();
let b_set = b.value.is_some();
a.namespace
.cmp(&b.namespace)
.then_with(|| b_set.cmp(&a_set))
.then_with(|| a.name.cmp(&b.name))
});
Ok(out)
}
pub async fn fetch_env_listeners(
&self,
application_name: &str,
env_name: &str,
) -> Result<Vec<(String, String, String)>> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(listeners) failed")?;
let mut out: Vec<(String, String, String)> = resp
.configuration_settings
.unwrap_or_default()
.into_iter()
.flat_map(|c| c.option_settings.unwrap_or_default())
.filter_map(|o| {
let ns = o.namespace?;
let port = ns.strip_prefix("aws:elbv2:listener:")?.to_string();
let opt = o.option_name?;
let value = o.value.unwrap_or_default();
if value.is_empty() {
return None;
}
Some((port, opt, value))
})
.collect();
out.sort_by(|a, b| {
let rank_a = u8::from(a.0 != "default");
let rank_b = u8::from(b.0 != "default");
let port_a = a.0.parse::<u32>().unwrap_or(0);
let port_b = b.0.parse::<u32>().unwrap_or(0);
(rank_a, port_a, &a.1).cmp(&(rank_b, port_b, &b.1))
});
Ok(out)
}
pub async fn list_certificates(&self) -> Result<Vec<AcmCert>> {
use aws_sdk_acm::types::CertificateStatus;
let mut out: Vec<AcmCert> = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self
.acm
.list_certificates()
.certificate_statuses(CertificateStatus::Issued);
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("ListCertificates failed")?;
for c in resp.certificate_summary_list.unwrap_or_default() {
if let Some(arn) = c.certificate_arn {
out.push(AcmCert {
arn,
domain: c.domain_name.unwrap_or_default(),
});
}
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
out.sort_by(|a, b| a.domain.cmp(&b.domain));
Ok(out)
}
pub async fn fetch_env_vars(
&self,
application_name: &str,
env_name: &str,
) -> Result<Vec<(String, String)>> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.environment_name(env_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(env) failed")?;
let mut out: Vec<(String, String)> = resp
.configuration_settings
.unwrap_or_default()
.into_iter()
.flat_map(|c| c.option_settings.unwrap_or_default())
.filter(|o| {
o.namespace.as_deref() == Some("aws:elasticbeanstalk:application:environment")
})
.map(|o| {
(
o.option_name.unwrap_or_default(),
o.value.unwrap_or_default(),
)
})
.collect();
out.sort();
Ok(out)
}
pub async fn update_env_option_settings(
&self,
env_name: &str,
to_set: &[(String, String, String)],
to_remove: &[(String, String)],
) -> Result<()> {
use aws_sdk_elasticbeanstalk::types::{ConfigurationOptionSetting, OptionSpecification};
if to_set.is_empty() && to_remove.is_empty() {
return Err(eyre!("update_env_option_settings: nothing to do"));
}
let mut req = self.client.update_environment().environment_name(env_name);
for (ns, name, value) in to_set {
req = req.option_settings(
ConfigurationOptionSetting::builder()
.namespace(ns)
.option_name(name)
.value(value)
.build(),
);
}
for (ns, name) in to_remove {
req = req.options_to_remove(
OptionSpecification::builder()
.namespace(ns)
.option_name(name)
.build(),
);
}
req.send()
.await
.wrap_err("UpdateEnvironment(option_settings) failed")?;
Ok(())
}
pub async fn discover_env_log_groups(&self, env_name: &str) -> Result<Vec<String>> {
let prefix = format!("/aws/elasticbeanstalk/{env_name}/");
let mut out: Vec<String> = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self
.cw_logs
.describe_log_groups()
.log_group_name_prefix(&prefix);
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("DescribeLogGroups failed")?;
for g in resp.log_groups.unwrap_or_default() {
if let Some(name) = g.log_group_name {
out.push(name);
}
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
out.sort();
Ok(out)
}
pub async fn fetch_recent_log_events(
&self,
log_group: &str,
since_ms: i64,
limit: i32,
) -> Result<(Vec<LogEvent>, i64)> {
let resp = self
.cw_logs
.filter_log_events()
.log_group_name(log_group)
.start_time(since_ms)
.limit(limit)
.send()
.await
.wrap_err("FilterLogEvents failed")?;
let mut out: Vec<LogEvent> = Vec::new();
let mut max_ts = since_ms;
for e in resp.events.unwrap_or_default() {
let ts = e.timestamp.unwrap_or(since_ms);
if ts > max_ts {
max_ts = ts;
}
out.push(LogEvent {
timestamp_ms: ts,
stream: e.log_stream_name.unwrap_or_default(),
message: e.message.unwrap_or_default(),
});
}
let next_since = if max_ts > since_ms {
max_ts + 1
} else {
since_ms
};
Ok((out, next_since))
}
pub async fn run_insights_query(
&self,
log_groups: &[String],
start_ms: i64,
end_ms: i64,
query: &str,
) -> Result<InsightsResults> {
use aws_sdk_cloudwatchlogs::types::QueryStatus;
let start_s = start_ms / 1000;
let end_s = end_ms / 1000;
let mut req = self
.cw_logs
.start_query()
.start_time(start_s)
.end_time(end_s)
.query_string(query);
for g in log_groups {
req = req.log_group_names(g);
}
let start_resp = req.send().await.wrap_err("StartQuery failed")?;
let query_id = start_resp
.query_id
.ok_or_else(|| eyre!("StartQuery returned no query_id"))?;
loop {
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
let resp = self
.cw_logs
.get_query_results()
.query_id(&query_id)
.send()
.await
.wrap_err("GetQueryResults failed")?;
let status = resp.status.clone();
let scanned = resp
.statistics
.as_ref()
.map(|s| s.records_scanned as i64)
.unwrap_or(0);
let matched = resp
.statistics
.as_ref()
.map(|s| s.records_matched as i64)
.unwrap_or(0);
match status {
Some(QueryStatus::Scheduled) | Some(QueryStatus::Running) => continue,
Some(QueryStatus::Complete) => {
let rows: Vec<InsightsRow> = resp
.results
.unwrap_or_default()
.into_iter()
.map(|fields| InsightsRow {
fields: fields
.into_iter()
.map(|f| (f.field.unwrap_or_default(), f.value.unwrap_or_default()))
.collect(),
})
.collect();
return Ok(InsightsResults {
rows,
records_scanned: scanned,
records_matched: matched,
});
}
Some(QueryStatus::Failed) => {
return Err(eyre!("Insights query failed"));
}
Some(QueryStatus::Cancelled) => {
return Err(eyre!("Insights query was cancelled"));
}
Some(QueryStatus::Timeout) => {
return Err(eyre!("Insights query timed out (server-side 15min cap)"));
}
Some(other) => {
return Err(eyre!(
"unexpected Insights query status: {}",
other.as_str()
));
}
None => {
return Err(eyre!("Insights query returned no status"));
}
}
}
}
pub async fn delete_alarms(&self, names: &[String]) -> Result<()> {
if names.is_empty() {
return Ok(());
}
let mut req = self.cw.delete_alarms();
for n in names {
req = req.alarm_names(n);
}
req.send().await.wrap_err("DeleteAlarms failed")?;
Ok(())
}
pub async fn fetch_env_metrics(
&self,
env_name: &str,
range_secs: i64,
) -> Result<Vec<MetricSeries>> {
use aws_sdk_cloudwatch::types::{Dimension, Metric, MetricDataQuery, MetricStat};
let end = Utc::now();
let start = end - chrono::Duration::seconds(range_secs);
let dim = Dimension::builder()
.name("EnvironmentName")
.value(env_name)
.build();
let make_query = |id: &str, name: &str, stat: &str| -> MetricDataQuery {
let metric = Metric::builder()
.namespace("AWS/ElasticBeanstalk")
.metric_name(name)
.dimensions(dim.clone())
.build();
let ms = MetricStat::builder()
.metric(metric)
.period(60)
.stat(stat)
.build();
MetricDataQuery::builder().id(id).metric_stat(ms).build()
};
let resp = self
.cw
.get_metric_data()
.start_time(to_smithy(start))
.end_time(to_smithy(end))
.metric_data_queries(make_query("health", "EnvironmentHealth", "Maximum"))
.metric_data_queries(make_query("req4xx", "ApplicationRequests4xx", "Sum"))
.metric_data_queries(make_query("req5xx", "ApplicationRequests5xx", "Sum"))
.metric_data_queries(make_query("p90", "ApplicationLatencyP90", "Average"))
.send()
.await?;
let order = ["health", "req4xx", "req5xx", "p90"];
let labels: std::collections::HashMap<&str, (&str, &str)> = [
("health", ("Env Health (0–25)", "score")),
("req4xx", ("4xx Requests / min", "count")),
("req5xx", ("5xx Requests / min", "count")),
("p90", ("Latency P90", "s")),
]
.into_iter()
.collect();
let mut by_id: std::collections::HashMap<String, MetricSeries> =
std::collections::HashMap::new();
for r in resp.metric_data_results.unwrap_or_default() {
let id = r.id.unwrap_or_default();
let display = labels
.get(id.as_str())
.copied()
.map(|(d, _)| d.to_string())
.unwrap_or_else(|| id.clone());
let timestamps = r.timestamps.unwrap_or_default();
let values = r.values.unwrap_or_default();
let mut points: Vec<(DateTime<Utc>, f64)> = timestamps
.iter()
.zip(values.iter())
.filter_map(|(ts, v)| {
DateTime::<Utc>::from_timestamp(ts.secs(), ts.subsec_nanos()).map(|t| (t, *v))
})
.collect();
points.sort_by_key(|(t, _)| *t);
by_id.insert(
id.clone(),
MetricSeries {
id,
label: display,
points,
},
);
}
Ok(order.iter().filter_map(|id| by_id.remove(*id)).collect())
}
pub async fn fetch_custom_env_metrics(
&self,
env_name: &str,
range_secs: i64,
specs: &[CustomMetricQuery],
) -> Result<Vec<MetricSeries>> {
use aws_sdk_cloudwatch::types::{Dimension, Metric, MetricDataQuery, MetricStat};
if specs.is_empty() {
return Ok(Vec::new());
}
let end = Utc::now();
let start = end - chrono::Duration::seconds(range_secs);
let mut req = self
.cw
.get_metric_data()
.start_time(to_smithy(start))
.end_time(to_smithy(end));
let mut id_to_label: std::collections::HashMap<String, String> =
std::collections::HashMap::new();
for (i, (label, namespace, name, stat, dims)) in specs.iter().enumerate() {
let id = format!("m{i}");
let mut metric_builder = Metric::builder().namespace(namespace).metric_name(name);
if dims.is_empty() {
metric_builder = metric_builder.dimensions(
Dimension::builder()
.name("EnvironmentName")
.value(env_name)
.build(),
);
} else {
for (k, v) in dims {
metric_builder =
metric_builder.dimensions(Dimension::builder().name(k).value(v).build());
}
}
let ms = MetricStat::builder()
.metric(metric_builder.build())
.period(60)
.stat(stat)
.build();
id_to_label.insert(id.clone(), label.clone());
req =
req.metric_data_queries(MetricDataQuery::builder().id(id).metric_stat(ms).build());
}
let resp = req.send().await?;
let mut by_id: std::collections::HashMap<String, MetricSeries> =
std::collections::HashMap::new();
for r in resp.metric_data_results.unwrap_or_default() {
let id = r.id.unwrap_or_default();
let label = id_to_label.get(&id).cloned().unwrap_or_else(|| id.clone());
let timestamps = r.timestamps.unwrap_or_default();
let values = r.values.unwrap_or_default();
let mut points: Vec<(DateTime<Utc>, f64)> = timestamps
.iter()
.zip(values.iter())
.filter_map(|(ts, v)| {
DateTime::<Utc>::from_timestamp(ts.secs(), ts.subsec_nanos()).map(|t| (t, *v))
})
.collect();
points.sort_by_key(|(t, _)| *t);
by_id.insert(id.clone(), MetricSeries { id, label, points });
}
Ok((0..specs.len())
.filter_map(|i| by_id.remove(&format!("m{i}")))
.collect())
}
pub async fn purge_queue(&self, queue_url: &str) -> Result<()> {
self.sqs.purge_queue().queue_url(queue_url).send().await?;
Ok(())
}
pub async fn list_tags(&self, resource_arn: &str) -> Result<Vec<(String, String)>> {
let resp = self
.client
.list_tags_for_resource()
.resource_arn(resource_arn)
.send()
.await?;
let tags = resp
.resource_tags
.unwrap_or_default()
.into_iter()
.filter_map(|t| match (t.key, t.value) {
(Some(k), Some(v)) => Some((k, v)),
_ => None,
})
.collect();
Ok(tags)
}
pub async fn update_tags(
&self,
resource_arn: &str,
to_add: &[(String, String)],
to_remove: &[String],
) -> Result<()> {
use aws_sdk_elasticbeanstalk::types::Tag;
let mut req = self
.client
.update_tags_for_resource()
.resource_arn(resource_arn);
for (k, v) in to_add {
req = req.tags_to_add(Tag::builder().key(k).value(v).build());
}
for k in to_remove {
req = req.tags_to_remove(k);
}
req.send().await?;
Ok(())
}
pub async fn rebuild_env(&self, env_name: &str) -> Result<()> {
self.client
.rebuild_environment()
.environment_name(env_name)
.send()
.await?;
Ok(())
}
pub async fn restart_app_server(&self, env_name: &str) -> Result<()> {
self.client
.restart_app_server()
.environment_name(env_name)
.send()
.await?;
Ok(())
}
pub async fn swap_cnames(&self, source: &str, dest: &str) -> Result<()> {
self.client
.swap_environment_cnames()
.source_environment_name(source)
.destination_environment_name(dest)
.send()
.await?;
Ok(())
}
pub async fn create_config_template(
&self,
application_name: &str,
template_name: &str,
source_env_name: &str,
) -> Result<()> {
self.client
.create_configuration_template()
.application_name(application_name)
.template_name(template_name)
.environment_id(source_env_name)
.send()
.await
.wrap_err("CreateConfigurationTemplate failed")?;
Ok(())
}
pub async fn delete_config_template(
&self,
application_name: &str,
template_name: &str,
) -> Result<()> {
self.client
.delete_configuration_template()
.application_name(application_name)
.template_name(template_name)
.send()
.await
.wrap_err("DeleteConfigurationTemplate failed")?;
Ok(())
}
pub async fn list_compatible_platforms(&self, env_name: &str) -> Result<Vec<CustomPlatform>> {
use aws_sdk_elasticbeanstalk::types::{PlatformFilter, PlatformStatus};
let desc = self
.client
.describe_environments()
.environment_names(env_name)
.send()
.await
.wrap_err("DescribeEnvironments failed")?;
let env = desc
.environments
.unwrap_or_default()
.into_iter()
.next()
.ok_or_else(|| eyre!("env '{env_name}' not found"))?;
let current_arn = env.platform_arn.clone().unwrap_or_default();
let stack_or_arn = env
.solution_stack_name
.clone()
.unwrap_or_else(|| current_arn.clone());
let branch = platform_branch_from(&stack_or_arn);
let owner_filter = PlatformFilter::builder()
.r#type("PlatformStatus")
.operator("=")
.values(PlatformStatus::Ready.as_str())
.build();
let mut filters = vec![owner_filter];
if !branch.is_empty() {
filters.push(
PlatformFilter::builder()
.r#type("PlatformBranchName")
.operator("=")
.values(branch.clone())
.build(),
);
}
let mut next_token: Option<String> = None;
let mut out: Vec<CustomPlatform> = Vec::new();
loop {
let mut req = self.client.list_platform_versions();
for f in &filters {
req = req.filters(f.clone());
}
if let Some(t) = next_token.clone() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("ListPlatformVersions failed")?;
for p in resp.platform_summary_list.unwrap_or_default() {
out.push(CustomPlatform {
arn: p.platform_arn.unwrap_or_default(),
branch: p.platform_branch_name.unwrap_or_default(),
version: p.platform_version.unwrap_or_default(),
status: p
.platform_status
.map(|s| s.as_str().to_string())
.unwrap_or_default(),
lifecycle: p.platform_lifecycle_state.unwrap_or_default(),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
out.sort_by(|a, b| compare_versions(&b.version, &a.version));
Ok(out)
}
pub async fn upgrade_platform(&self, env_name: &str, platform_arn: &str) -> Result<()> {
self.client
.update_environment()
.environment_name(env_name)
.platform_arn(platform_arn)
.send()
.await
.wrap_err("UpdateEnvironment(platform_arn) failed")?;
Ok(())
}
pub async fn clone_env(&self, source_env_name: &str, target_env_name: &str) -> Result<()> {
let desc = self
.client
.describe_environments()
.environment_names(source_env_name)
.send()
.await
.wrap_err("DescribeEnvironments failed")?;
let env = desc
.environments
.unwrap_or_default()
.into_iter()
.next()
.ok_or_else(|| eyre!("source env '{source_env_name}' not found"))?;
let application = env
.application_name
.ok_or_else(|| eyre!("source env has no application_name"))?;
let env_id = env
.environment_id
.ok_or_else(|| eyre!("source env has no environment_id"))?;
let template = format!(
"__ebman-clone-{}-{}",
target_env_name,
chrono::Utc::now().timestamp()
);
self.client
.create_configuration_template()
.application_name(&application)
.template_name(&template)
.environment_id(&env_id)
.send()
.await
.wrap_err("CreateConfigurationTemplate failed")?;
let create_result = self
.client
.create_environment()
.application_name(&application)
.environment_name(target_env_name)
.template_name(&template)
.send()
.await;
let _ = self
.client
.delete_configuration_template()
.application_name(&application)
.template_name(&template)
.send()
.await;
create_result.wrap_err("CreateEnvironment failed")?;
Ok(())
}
pub async fn scale_env(&self, env_name: &str, min: i32, max: i32) -> Result<()> {
use aws_sdk_elasticbeanstalk::types::ConfigurationOptionSetting;
let opts = vec![
ConfigurationOptionSetting::builder()
.namespace("aws:autoscaling:asg")
.option_name("MinSize")
.value(min.to_string())
.build(),
ConfigurationOptionSetting::builder()
.namespace("aws:autoscaling:asg")
.option_name("MaxSize")
.value(max.to_string())
.build(),
];
self.client
.update_environment()
.environment_name(env_name)
.set_option_settings(Some(opts))
.send()
.await
.wrap_err("UpdateEnvironment(asg) failed")?;
Ok(())
}
pub async fn terminate_instance(&self, instance_id: &str) -> Result<()> {
self.ec2
.terminate_instances()
.instance_ids(instance_id)
.send()
.await
.wrap_err("ec2:TerminateInstances failed")?;
Ok(())
}
pub async fn abort_environment_update(&self, env_name: &str) -> Result<()> {
self.client
.abort_environment_update()
.environment_name(env_name)
.send()
.await
.wrap_err("AbortEnvironmentUpdate failed")?;
Ok(())
}
pub async fn list_custom_platforms(&self) -> Result<Vec<CustomPlatform>> {
use aws_sdk_elasticbeanstalk::types::PlatformFilter;
let filter = PlatformFilter::builder()
.r#type("PlatformOwner")
.operator("=")
.values("self")
.build();
let mut next_token: Option<String> = None;
let mut out: Vec<CustomPlatform> = Vec::new();
loop {
let mut req = self.client.list_platform_versions().filters(filter.clone());
if let Some(t) = next_token.clone() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("ListPlatformVersions failed")?;
for p in resp.platform_summary_list.unwrap_or_default() {
out.push(CustomPlatform {
arn: p.platform_arn.unwrap_or_default(),
branch: p.platform_branch_name.unwrap_or_default(),
version: p.platform_version.unwrap_or_default(),
status: p
.platform_status
.map(|s| s.as_str().to_string())
.unwrap_or_default(),
lifecycle: p.platform_lifecycle_state.unwrap_or_default(),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
Ok(out)
}
pub async fn delete_custom_platform(&self, platform_arn: &str) -> Result<()> {
self.client
.delete_platform_version()
.platform_arn(platform_arn)
.send()
.await
.wrap_err("DeletePlatformVersion failed")?;
Ok(())
}
pub async fn list_application_versions(
&self,
application_name: &str,
) -> Result<Vec<AppVersion>> {
let mut out: Vec<AppVersion> = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self
.client
.describe_application_versions()
.application_name(application_name);
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req
.send()
.await
.wrap_err("DescribeApplicationVersions failed")?;
for v in resp.application_versions.unwrap_or_default() {
out.push(AppVersion {
label: v.version_label.unwrap_or_default(),
description: v.description.unwrap_or_default(),
created: v
.date_created
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
out.sort_by_key(|v| std::cmp::Reverse(v.created));
Ok(out)
}
pub async fn delete_application_version(
&self,
application_name: &str,
version_label: &str,
delete_source_bundle: bool,
) -> Result<()> {
self.client
.delete_application_version()
.application_name(application_name)
.version_label(version_label)
.delete_source_bundle(delete_source_bundle)
.send()
.await
.wrap_err("DeleteApplicationVersion failed")?;
Ok(())
}
pub async fn create_storage_location(&self) -> Result<String> {
let resp = self
.client
.create_storage_location()
.send()
.await
.wrap_err("CreateStorageLocation failed")?;
resp.s3_bucket
.ok_or_else(|| eyre!("CreateStorageLocation returned no S3Bucket"))
}
pub async fn upload_bundle(
&self,
bucket: &str,
key: &str,
path: &std::path::Path,
) -> Result<()> {
self.upload_bundle_with(bucket, key, path, MULTIPART_THRESHOLD, MULTIPART_PART_SIZE)
.await
}
pub async fn upload_bundle_with(
&self,
bucket: &str,
key: &str,
path: &std::path::Path,
multipart_threshold: u64,
part_size: u64,
) -> Result<()> {
use aws_sdk_s3::primitives::ByteStream;
let metadata = tokio::fs::metadata(path)
.await
.wrap_err_with(|| format!("stat bundle {}", path.display()))?;
let size = metadata.len();
if !should_multipart(size, multipart_threshold) {
let body = ByteStream::from_path(path)
.await
.wrap_err_with(|| format!("read {}", path.display()))?;
self.s3
.put_object()
.bucket(bucket)
.key(key)
.body(body)
.send()
.await
.wrap_err_with(|| format!("S3 PutObject {bucket}/{key} failed"))?;
return Ok(());
}
let create = self
.s3
.create_multipart_upload()
.bucket(bucket)
.key(key)
.send()
.await
.wrap_err_with(|| format!("S3 CreateMultipartUpload {bucket}/{key} failed"))?;
let upload_id = create
.upload_id()
.ok_or_else(|| eyre!("CreateMultipartUpload returned no UploadId"))?
.to_string();
let plan = plan_part_lengths(size, part_size);
let mut completed_parts: Vec<aws_sdk_s3::types::CompletedPart> =
Vec::with_capacity(plan.len());
use tokio::io::AsyncReadExt;
let mut file = match tokio::fs::File::open(path).await {
Ok(f) => f,
Err(e) => {
let _ = self
.s3
.abort_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(eyre!("open {} for multipart upload: {e}", path.display()));
}
};
for (idx, part_len) in plan.iter().enumerate() {
let part_number = (idx + 1) as i32;
let mut buf = vec![0u8; *part_len as usize];
if let Err(e) = file.read_exact(&mut buf).await {
let _ = self
.s3
.abort_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(eyre!(
"read part {part_number} from {}: {e}",
path.display()
));
}
let resp = match self
.s3
.upload_part()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.part_number(part_number)
.body(ByteStream::from(buf))
.send()
.await
{
Ok(r) => r,
Err(e) => {
let _ = self
.s3
.abort_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(e).wrap_err_with(|| {
format!("S3 UploadPart {part_number} of {bucket}/{key} failed")
});
}
};
let e_tag = resp
.e_tag()
.ok_or_else(|| eyre!("UploadPart {part_number} returned no ETag"))?
.to_string();
completed_parts.push(
aws_sdk_s3::types::CompletedPart::builder()
.part_number(part_number)
.e_tag(e_tag)
.build(),
);
}
let completed = aws_sdk_s3::types::CompletedMultipartUpload::builder()
.set_parts(Some(completed_parts))
.build();
if let Err(e) = self
.s3
.complete_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.multipart_upload(completed)
.send()
.await
{
let _ = self
.s3
.abort_multipart_upload()
.bucket(bucket)
.key(key)
.upload_id(&upload_id)
.send()
.await;
return Err(e)
.wrap_err_with(|| format!("S3 CompleteMultipartUpload {bucket}/{key} failed"));
}
Ok(())
}
pub async fn create_app_version(
&self,
application_name: &str,
version_label: &str,
description: Option<&str>,
s3_bucket: &str,
s3_key: &str,
) -> Result<()> {
use aws_sdk_elasticbeanstalk::types::S3Location;
let source = S3Location::builder()
.s3_bucket(s3_bucket)
.s3_key(s3_key)
.build();
let mut req = self
.client
.create_application_version()
.application_name(application_name)
.version_label(version_label)
.source_bundle(source)
.auto_create_application(false);
if let Some(d) = description {
req = req.description(d);
}
req.send()
.await
.wrap_err("CreateApplicationVersion failed")?;
Ok(())
}
pub async fn deploy_version(&self, env_name: &str, version_label: &str) -> Result<()> {
self.client
.update_environment()
.environment_name(env_name)
.version_label(version_label)
.send()
.await
.wrap_err("UpdateEnvironment(version_label) failed")?;
Ok(())
}
pub async fn describe_template_settings(
&self,
application_name: &str,
template_name: &str,
) -> Result<Vec<(String, String, String)>> {
let resp = self
.client
.describe_configuration_settings()
.application_name(application_name)
.template_name(template_name)
.send()
.await
.wrap_err("DescribeConfigurationSettings(template) failed")?;
let mut out: Vec<(String, String, String)> = resp
.configuration_settings
.unwrap_or_default()
.into_iter()
.flat_map(|c| c.option_settings.unwrap_or_default())
.map(|o| {
(
o.namespace.unwrap_or_default(),
o.option_name.unwrap_or_default(),
o.value.unwrap_or_default(),
)
})
.collect();
out.sort();
Ok(out)
}
pub async fn apply_config_template(&self, env_name: &str, template_name: &str) -> Result<()> {
self.client
.update_environment()
.environment_name(env_name)
.template_name(template_name)
.send()
.await
.wrap_err("UpdateEnvironment(template_name) failed")?;
Ok(())
}
pub async fn terminate_env(&self, env_name: &str) -> Result<()> {
self.client
.terminate_environment()
.environment_name(env_name)
.send()
.await?;
Ok(())
}
pub async fn request_env_info_tail(&self, env_name: &str) -> Result<()> {
use aws_sdk_elasticbeanstalk::types::EnvironmentInfoType;
self.client
.request_environment_info()
.environment_name(env_name)
.info_type(EnvironmentInfoType::Tail)
.send()
.await
.wrap_err("RequestEnvironmentInfo failed")?;
Ok(())
}
pub async fn retrieve_env_info_tail(&self, env_name: &str) -> Result<Vec<(String, String)>> {
use aws_sdk_elasticbeanstalk::types::EnvironmentInfoType;
let resp = self
.client
.retrieve_environment_info()
.environment_name(env_name)
.info_type(EnvironmentInfoType::Tail)
.send()
.await
.wrap_err("RetrieveEnvironmentInfo failed")?;
let mut out = Vec::new();
for info in resp.environment_info.unwrap_or_default() {
if let (Some(id), Some(url)) = (info.ec2_instance_id, info.message) {
out.push((id, url));
}
}
Ok(out)
}
pub async fn fetch_url_text(url: &str) -> Result<String> {
use tokio::process::Command;
let out = Command::new("curl")
.args([
"-s",
"-S",
"--fail-with-body",
"--max-time",
"15",
"--no-buffer",
])
.arg(url)
.output()
.await
.wrap_err("could not invoke curl (is it installed?)")?;
if !out.status.success() {
return Err(eyre!(
"curl exit {}: {}",
out.status,
String::from_utf8_lossy(&out.stderr).trim()
));
}
Ok(String::from_utf8_lossy(&out.stdout).into_owned())
}
pub async fn fetch_env_instance_counts(&self, env_name: &str) -> Result<EnvInstanceCounts> {
let resp = self
.client
.describe_environment_health()
.environment_name(env_name)
.attribute_names(
aws_sdk_elasticbeanstalk::types::EnvironmentHealthAttribute::InstancesHealth,
)
.send()
.await
.wrap_err("DescribeEnvironmentHealth failed")?;
Ok(summarise_instance_health(resp.instances_health.as_ref()))
}
pub async fn list_instances(&self, env_name: &str) -> Result<Vec<Instance>> {
let resp = self
.client
.describe_instances_health()
.environment_name(env_name)
.attribute_names(aws_sdk_elasticbeanstalk::types::InstancesHealthAttribute::All)
.send()
.await?;
let instances = resp
.instance_health_list
.unwrap_or_default()
.into_iter()
.map(|i| Instance {
id: i.instance_id.unwrap_or_default(),
health: i.health_status.unwrap_or_default(),
color: i.color.unwrap_or_default(),
causes: i.causes.unwrap_or_default(),
instance_type: i.instance_type.unwrap_or_default(),
availability_zone: i.availability_zone.unwrap_or_default(),
launched_at: i
.launched_at
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
})
.collect();
Ok(instances)
}
pub async fn list_org_accounts(&self) -> Result<Vec<OrgAccount>> {
let mut out: Vec<OrgAccount> = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self.org.list_accounts();
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req
.send()
.await
.wrap_err("organizations:ListAccounts failed")?;
for a in resp.accounts.unwrap_or_default() {
out.push(OrgAccount {
id: a.id.unwrap_or_default(),
name: a.name.unwrap_or_default(),
email: a.email,
status: a.status.map(|s| s.as_str().to_string()).unwrap_or_default(),
});
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
out.sort_by(|a, b| {
let sa = (a.status != "ACTIVE", a.name.to_lowercase());
let sb = (b.status != "ACTIVE", b.name.to_lowercase());
sa.cmp(&sb)
});
Ok(out)
}
pub async fn list_applications(&self) -> Result<Vec<Application>> {
let resp = self.client.describe_applications().send().await?;
let apps = resp
.applications
.unwrap_or_default()
.into_iter()
.map(|a| Application {
name: a.application_name.unwrap_or_default(),
description: a.description.unwrap_or_default(),
date_created: a
.date_created
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
date_updated: a
.date_updated
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
version_count: a.versions.map(|v| v.len()).unwrap_or(0),
templates: a.configuration_templates.unwrap_or_default(),
latest_version_label: None,
latest_version_created: None,
})
.collect();
Ok(apps)
}
pub async fn list_environments(&self) -> Result<Vec<Environment>> {
let mut all = Vec::new();
let mut next_token: Option<String> = None;
loop {
let mut req = self.client.describe_environments().include_deleted(false);
if let Some(t) = next_token.take() {
req = req.next_token(t);
}
let resp = req.send().await.wrap_err("DescribeEnvironments failed")?;
if let Some(envs) = resp.environments {
all.extend(envs.into_iter().map(map_env));
}
match resp.next_token {
Some(t) if !t.is_empty() => next_token = Some(t),
_ => break,
}
}
Ok(all)
}
pub async fn list_solution_stacks(&self) -> Result<Vec<String>> {
let resp = self
.client
.list_available_solution_stacks()
.send()
.await
.wrap_err("ListAvailableSolutionStacks failed")?;
Ok(resp.solution_stacks.unwrap_or_default())
}
}
fn map_env(e: aws_sdk_elasticbeanstalk::types::EnvironmentDescription) -> Environment {
let solution_stack = e.solution_stack_name.clone().unwrap_or_default();
let raw_platform = e
.solution_stack_name
.clone()
.or(e.platform_arn.clone())
.unwrap_or_default();
let tier = e
.tier
.as_ref()
.and_then(|t| t.name.as_deref())
.map(normalize_tier)
.unwrap_or_else(|| "?".into());
Environment {
name: e.environment_name.unwrap_or_default(),
application: e.application_name.unwrap_or_default(),
status: e
.status
.map(|s| s.as_str().to_string())
.unwrap_or_else(|| "-".into()),
health: e
.health
.map(|h| h.as_str().to_string())
.unwrap_or_else(|| "-".into()),
platform: platform_family(&raw_platform),
solution_stack,
tier,
cname: e.cname.unwrap_or_default(),
version_label: e.version_label.unwrap_or_default(),
arn: e.environment_arn,
updated: e
.date_updated
.and_then(|d| DateTime::from_timestamp(d.secs(), d.subsec_nanos())),
id: e.environment_id,
region: None,
}
}
fn platform_branch_from(stack_or_arn: &str) -> String {
if let Some(rest) = stack_or_arn.split(" running ").nth(1) {
return rest.trim().to_string();
}
if stack_or_arn.starts_with("arn:") {
let parts: Vec<&str> = stack_or_arn.split('/').collect();
if parts.len() >= 2 {
return parts[parts.len() - 2].to_string();
}
}
String::new()
}
fn compare_versions(a: &str, b: &str) -> std::cmp::Ordering {
use std::cmp::Ordering;
let parse = |s: &str| {
s.split('.')
.map(|p| p.split('-').next().unwrap_or(p).parse::<u64>().ok())
.collect::<Vec<_>>()
};
let av = parse(a);
let bv = parse(b);
for i in 0..av.len().max(bv.len()) {
let aa = av.get(i).and_then(|x| *x);
let bb = bv.get(i).and_then(|x| *x);
match (aa, bb) {
(Some(x), Some(y)) => match x.cmp(&y) {
Ordering::Equal => continue,
o => return o,
},
(Some(_), None) => return Ordering::Greater,
(None, Some(_)) => return Ordering::Less,
(None, None) => break,
}
}
a.cmp(b)
}
pub fn summarise_instance_health(
summary: Option<&aws_sdk_elasticbeanstalk::types::InstanceHealthSummary>,
) -> EnvInstanceCounts {
let Some(s) = summary else {
return EnvInstanceCounts::default();
};
let g = |v: Option<i32>| v.unwrap_or(0);
let ok = g(s.ok);
let info = g(s.info);
let healthy = ok + info;
let total = g(s.no_data)
+ g(s.unknown)
+ g(s.pending)
+ ok
+ info
+ g(s.warning)
+ g(s.degraded)
+ g(s.severe);
EnvInstanceCounts { healthy, total }
}
pub fn parse_window_ms(input: &str) -> Option<i64> {
let s = input.trim().to_lowercase();
if s.is_empty() {
return None;
}
let unit = s.chars().last()?;
let num: i64 = s[..s.len() - unit.len_utf8()].parse().ok()?;
if num <= 0 {
return None;
}
let ms = match unit {
'm' => num * 60_000,
'h' => num * 60 * 60_000,
'd' => num * 24 * 60 * 60_000,
_ => return None,
};
Some(ms)
}
pub fn format_insights_results(
results: &InsightsResults,
query: &str,
log_groups: &[String],
) -> String {
let mut out = String::new();
out.push_str(&format!(
"query: {query}\nlog groups: {}\nmatched: {} / scanned: {}\n",
if log_groups.is_empty() {
"(none)".to_string()
} else {
log_groups.join(", ")
},
results.records_matched,
results.records_scanned,
));
out.push_str(&"─".repeat(60));
out.push('\n');
if results.rows.is_empty() {
out.push_str("(no rows matched the query)\n");
return out;
}
let headers: Vec<String> = results.rows[0]
.fields
.iter()
.map(|(k, _)| k.clone())
.filter(|k| k != "@ptr")
.collect();
const COL_MAX: usize = 60;
let mut widths: Vec<usize> = headers.iter().map(|h| h.len()).collect();
for row in &results.rows {
for (i, h) in headers.iter().enumerate() {
if let Some((_, v)) = row.fields.iter().find(|(k, _)| k == h) {
let cells = v.chars().count().min(COL_MAX);
if cells > widths[i] {
widths[i] = cells;
}
}
}
}
let mut header_line = String::new();
for (i, h) in headers.iter().enumerate() {
if i > 0 {
header_line.push_str(" ");
}
header_line.push_str(&format!("{:<w$}", h, w = widths[i]));
}
out.push_str(&header_line);
out.push('\n');
let mut sep_line = String::new();
for (i, w) in widths.iter().enumerate() {
if i > 0 {
sep_line.push_str(" ");
}
sep_line.push_str(&"─".repeat(*w));
}
out.push_str(&sep_line);
out.push('\n');
for row in &results.rows {
let mut line = String::new();
for (i, h) in headers.iter().enumerate() {
if i > 0 {
line.push_str(" ");
}
let raw = row
.fields
.iter()
.find(|(k, _)| k == h)
.map(|(_, v)| v.as_str())
.unwrap_or("");
let trimmed: String = if raw.chars().count() > COL_MAX {
let mut s: String = raw.chars().take(COL_MAX.saturating_sub(1)).collect();
s.push('…');
s
} else {
raw.to_string()
};
line.push_str(&format!("{:<w$}", trimmed, w = widths[i]));
}
out.push_str(&line);
out.push('\n');
}
out
}
pub const MULTIPART_THRESHOLD: u64 = 64 * 1024 * 1024;
pub const MULTIPART_PART_SIZE: u64 = 16 * 1024 * 1024;
pub fn should_multipart(size: u64, threshold: u64) -> bool {
size >= threshold
}
pub fn plan_part_lengths(total_size: u64, part_size: u64) -> Vec<u64> {
if total_size == 0 || part_size == 0 {
return Vec::new();
}
let full = total_size / part_size;
let remainder = total_size % part_size;
let mut out = Vec::with_capacity(full as usize + if remainder > 0 { 1 } else { 0 });
for _ in 0..full {
out.push(part_size);
}
if remainder > 0 {
out.push(remainder);
}
out
}
pub fn stack_family_version(stack: &str) -> Option<(String, String)> {
let version_token = stack.split_whitespace().find(|tok| {
tok.strip_prefix('v')
.map(|rest| {
!rest.is_empty()
&& rest
.split('.')
.all(|p| !p.is_empty() && p.chars().all(|c| c.is_ascii_digit()))
})
.unwrap_or(false)
})?;
let version = version_token.trim_start_matches('v').to_string();
let key = stack
.split_whitespace()
.filter(|tok| *tok != version_token)
.collect::<Vec<_>>()
.join(" ");
Some((key, version))
}
pub fn latest_stack_versions(stacks: &[String]) -> std::collections::HashMap<String, String> {
let mut out: std::collections::HashMap<String, String> = std::collections::HashMap::new();
for s in stacks {
if let Some((key, ver)) = stack_family_version(s) {
match out.get(&key) {
Some(cur) if compare_versions(&ver, cur) != std::cmp::Ordering::Greater => {}
_ => {
out.insert(key, ver);
}
}
}
}
out
}
pub fn newer_stack_version(
env_stack: &str,
latest: &std::collections::HashMap<String, String>,
) -> Option<String> {
let (key, ver) = stack_family_version(env_stack)?;
let newest = latest.get(&key)?;
if compare_versions(newest, &ver) == std::cmp::Ordering::Greater {
Some(newest.clone())
} else {
None
}
}
pub async fn list_environments_in_region(
profile: Option<String>,
region: String,
) -> Result<Vec<Environment>> {
let client = AwsClient::with(profile, Some(region.clone())).await?;
let mut envs = client.list_environments().await?;
for e in &mut envs {
e.region = Some(region.clone());
}
Ok(envs)
}
pub async fn list_environments_for_account(
name: &str,
spec: &crate::config::AccountSpec,
region: Option<String>,
) -> Result<Vec<Environment>> {
let mut spec = spec.clone();
if region.is_some() {
spec.region = region.clone();
}
let client = AwsClient::assume_role(name, &spec).await?;
let resolved_region = client.context.region.clone();
let mut envs = client.list_environments().await?;
for e in &mut envs {
e.region = Some(resolved_region.clone());
}
Ok(envs)
}
fn platform_family(raw: &str) -> String {
if raw.is_empty() {
return String::new();
}
if raw.contains(" running on ") {
for seg in raw.split('/') {
if let Some((family, _)) = seg.split_once(" running on ") {
return family.trim().to_string();
}
}
}
if let Some((_, after)) = raw.rsplit_once(" running ") {
return after.trim().to_string();
}
raw.to_string()
}
fn to_smithy(d: DateTime<Utc>) -> aws_sdk_cloudwatch::primitives::DateTime {
aws_sdk_cloudwatch::primitives::DateTime::from_secs(d.timestamp())
}
fn split_csv(value: &str) -> Vec<String> {
value
.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
}
fn derive_dlq_url(main: &str) -> Option<String> {
let trimmed = main.trim_end_matches('/');
if trimmed.ends_with("-dlq") {
return None;
}
Some(format!("{trimmed}-dlq"))
}
fn normalize_tier(name: &str) -> String {
match name {
"WebServer" => "Web".into(),
"Worker" => "Worker".into(),
other => other.to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn platform_family_from_solution_stack() {
assert_eq!(
platform_family("64bit Amazon Linux 2 v3.5.0 running Java 17"),
"Java 17"
);
assert_eq!(
platform_family("64bit Amazon Linux 2 v3.7.0 running Tomcat 9 Corretto 17"),
"Tomcat 9 Corretto 17"
);
assert_eq!(
platform_family("64bit Amazon Linux 2023 v6.1.0 running Node.js 18"),
"Node.js 18"
);
}
#[test]
fn platform_family_from_arn() {
assert_eq!(
platform_family(
"arn:aws:elasticbeanstalk:us-east-1::platform/Java 17 running on 64bit Amazon Linux 2/3.5.0"
),
"Java 17"
);
}
#[test]
fn platform_family_handles_empty_and_unknown() {
assert_eq!(platform_family(""), "");
assert_eq!(platform_family("just a string"), "just a string");
}
#[test]
fn stack_family_version_splits_solution_stack() {
assert_eq!(
stack_family_version("64bit Amazon Linux 2023 v6.1.0 running Node.js 18"),
Some((
"64bit Amazon Linux 2023 running Node.js 18".to_string(),
"6.1.0".to_string()
))
);
}
#[test]
fn stack_family_version_rejects_versionless() {
assert_eq!(stack_family_version(""), None);
assert_eq!(stack_family_version("some platform with no version"), None);
assert_eq!(stack_family_version("running via vN stack"), None);
}
#[test]
fn latest_stack_versions_keeps_newest_per_family() {
let stacks = vec![
"64bit Amazon Linux 2 v3.1.0 running Node.js 14".to_string(),
"64bit Amazon Linux 2 v3.10.0 running Node.js 14".to_string(),
"64bit Amazon Linux 2 v3.2.0 running Node.js 14".to_string(),
"64bit Amazon Linux 2023 v6.1.0 running Node.js 18".to_string(),
];
let latest = latest_stack_versions(&stacks);
assert_eq!(
latest.get("64bit Amazon Linux 2 running Node.js 14"),
Some(&"3.10.0".to_string())
);
assert_eq!(
latest.get("64bit Amazon Linux 2023 running Node.js 18"),
Some(&"6.1.0".to_string())
);
}
#[test]
fn newer_stack_version_flags_only_superseded() {
let latest = latest_stack_versions(&[
"64bit Amazon Linux 2023 v6.1.0 running Node.js 18".to_string()
]);
assert_eq!(
newer_stack_version("64bit Amazon Linux 2023 v6.0.3 running Node.js 18", &latest),
Some("6.1.0".to_string())
);
assert_eq!(
newer_stack_version("64bit Amazon Linux 2023 v6.1.0 running Node.js 18", &latest),
None
);
assert_eq!(
newer_stack_version("64bit Amazon Linux 2023 v1.0.0 running Node.js 20", &latest),
None
);
assert_eq!(newer_stack_version("", &latest), None);
}
#[test]
fn normalize_tier_maps_known_names() {
assert_eq!(normalize_tier("WebServer"), "Web");
assert_eq!(normalize_tier("Worker"), "Worker");
assert_eq!(normalize_tier("Other"), "Other");
}
#[test]
fn derive_dlq_url_appends_suffix() {
assert_eq!(
derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue"),
Some("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue-dlq".to_string())
);
}
#[test]
fn should_multipart_crosses_threshold() {
assert!(!should_multipart(0, 64));
assert!(!should_multipart(63, 64));
assert!(should_multipart(64, 64));
assert!(should_multipart(1_000_000, 64));
}
#[test]
fn plan_part_lengths_exact_multiple() {
assert_eq!(plan_part_lengths(48, 16), vec![16, 16, 16]);
}
#[test]
fn plan_part_lengths_partial_last_part() {
assert_eq!(plan_part_lengths(17, 8), vec![8, 8, 1]);
}
#[test]
fn plan_part_lengths_zero_and_under_one_part() {
assert!(plan_part_lengths(0, 16).is_empty());
assert_eq!(plan_part_lengths(5, 16), vec![5]);
assert!(plan_part_lengths(100, 0).is_empty());
}
#[test]
fn summarise_instance_health_rolls_up_buckets() {
use aws_sdk_elasticbeanstalk::types::InstanceHealthSummary;
let s = InstanceHealthSummary::builder()
.ok(2)
.info(1)
.warning(1)
.degraded(0)
.severe(1)
.pending(0)
.no_data(0)
.unknown(0)
.build();
let counts = super::summarise_instance_health(Some(&s));
assert_eq!(counts.healthy, 3, "ok + info");
assert_eq!(counts.total, 5, "ok + info + warning + degraded + severe");
let s = InstanceHealthSummary::builder()
.pending(2)
.no_data(1)
.build();
let counts = super::summarise_instance_health(Some(&s));
assert_eq!(counts.healthy, 0);
assert_eq!(counts.total, 3);
let counts = super::summarise_instance_health(None);
assert_eq!(counts.healthy, 0);
assert_eq!(counts.total, 0);
let s = InstanceHealthSummary::builder().build();
let counts = super::summarise_instance_health(Some(&s));
assert_eq!(counts.healthy, 0);
assert_eq!(counts.total, 0);
}
#[test]
fn parse_window_ms_accepts_minutes_hours_days() {
assert_eq!(super::parse_window_ms("30m"), Some(30 * 60_000));
assert_eq!(super::parse_window_ms("1h"), Some(60 * 60_000));
assert_eq!(super::parse_window_ms("6h"), Some(6 * 60 * 60_000));
assert_eq!(super::parse_window_ms("24h"), Some(24 * 60 * 60_000));
assert_eq!(super::parse_window_ms("7d"), Some(7 * 24 * 60 * 60_000));
assert_eq!(super::parse_window_ms(" 2h "), Some(2 * 60 * 60_000));
assert_eq!(super::parse_window_ms("3H"), Some(3 * 60 * 60_000));
}
#[test]
fn parse_window_ms_rejects_malformed_input() {
assert_eq!(super::parse_window_ms(""), None);
assert_eq!(super::parse_window_ms("30"), None);
assert_eq!(super::parse_window_ms("h"), None);
assert_eq!(super::parse_window_ms("1y"), None);
assert_eq!(super::parse_window_ms("2w"), None);
assert_eq!(super::parse_window_ms("60s"), None);
assert_eq!(super::parse_window_ms("0h"), None);
assert_eq!(super::parse_window_ms("-1h"), None);
assert_eq!(super::parse_window_ms("hour"), None);
}
#[test]
fn format_insights_results_renders_table() {
let results = InsightsResults {
rows: vec![
InsightsRow {
fields: vec![
("@timestamp".into(), "2026-05-23T10:00:00Z".into()),
("@message".into(), "POST /checkout 200 42ms".into()),
("@ptr".into(), "CWL_PTR_X".into()),
],
},
InsightsRow {
fields: vec![
("@timestamp".into(), "2026-05-23T10:00:01Z".into()),
("@message".into(), "GET /healthcheck 200 1ms".into()),
("@ptr".into(), "CWL_PTR_Y".into()),
],
},
],
records_scanned: 1234,
records_matched: 2,
};
let body = super::format_insights_results(
&results,
"fields @timestamp, @message",
&["/aws/elasticbeanstalk/prod/var/log/web.stdout.log".to_string()],
);
assert!(
body.contains("matched: 2 / scanned: 1234"),
"stats line present"
);
assert!(body.contains("@timestamp"), "@timestamp header present");
assert!(body.contains("@message"), "@message header present");
assert!(
!body.contains("@ptr"),
"@ptr field should be filtered out of the rendered table"
);
assert!(body.contains("POST /checkout"), "first row body present");
assert!(body.contains("GET /healthcheck"), "second row body present");
}
#[test]
fn format_insights_results_empty_input_shows_no_rows_stub() {
let results = InsightsResults {
rows: vec![],
records_scanned: 1000,
records_matched: 0,
};
let body = super::format_insights_results(
&results,
"fields @message | filter @message like /never/",
&["/aws/elasticbeanstalk/prod/var/log/web.stdout.log".to_string()],
);
assert!(body.contains("no rows matched"), "empty-input stub fires");
assert!(
body.contains("matched: 0 / scanned: 1000"),
"stats line still present"
);
}
#[test]
fn format_insights_results_truncates_long_values() {
let huge = "x".repeat(200);
let results = InsightsResults {
rows: vec![InsightsRow {
fields: vec![("@message".into(), huge.clone())],
}],
records_scanned: 1,
records_matched: 1,
};
let body = super::format_insights_results(&results, "fields @message", &[]);
assert!(
!body.contains(&huge),
"raw 200-char value should not appear untouched"
);
assert!(
body.contains("…"),
"truncation marker should signal the cut to the operator"
);
}
#[tokio::test]
async fn upload_bundle_uses_multipart_when_size_meets_threshold() {
use aws_sdk_s3::operation::complete_multipart_upload::CompleteMultipartUploadOutput;
use aws_sdk_s3::operation::create_multipart_upload::CreateMultipartUploadOutput;
use aws_sdk_s3::operation::upload_part::UploadPartOutput;
const BUCKET: &str = "elasticbeanstalk-eu-west-2-123";
const KEY: &str = "applications/big-app/v1";
const UPLOAD_ID: &str = "test-upload-id";
let cmu_rule = mock!(aws_sdk_s3::Client::create_multipart_upload)
.match_requests(|req| req.bucket() == Some(BUCKET) && req.key() == Some(KEY))
.then_output(|| {
CreateMultipartUploadOutput::builder()
.upload_id(UPLOAD_ID)
.build()
});
let up_rule_1 = mock!(aws_sdk_s3::Client::upload_part)
.match_requests(|req| {
req.bucket() == Some(BUCKET)
&& req.key() == Some(KEY)
&& req.upload_id() == Some(UPLOAD_ID)
&& req.part_number() == Some(1)
})
.then_output(|| UploadPartOutput::builder().e_tag("\"etag-1\"").build());
let up_rule_2 = mock!(aws_sdk_s3::Client::upload_part)
.match_requests(|req| {
req.bucket() == Some(BUCKET)
&& req.key() == Some(KEY)
&& req.upload_id() == Some(UPLOAD_ID)
&& req.part_number() == Some(2)
})
.then_output(|| UploadPartOutput::builder().e_tag("\"etag-2\"").build());
let up_rule_3 = mock!(aws_sdk_s3::Client::upload_part)
.match_requests(|req| {
req.bucket() == Some(BUCKET)
&& req.key() == Some(KEY)
&& req.upload_id() == Some(UPLOAD_ID)
&& req.part_number() == Some(3)
})
.then_output(|| UploadPartOutput::builder().e_tag("\"etag-3\"").build());
let cmpu_rule = mock!(aws_sdk_s3::Client::complete_multipart_upload)
.match_requests(|req| {
req.bucket() == Some(BUCKET)
&& req.key() == Some(KEY)
&& req.upload_id() == Some(UPLOAD_ID)
&& req.multipart_upload().map(|m| m.parts().len()) == Some(3)
})
.then_output(|| CompleteMultipartUploadOutput::builder().build());
let s3 = mock_client!(
aws_sdk_s3,
[&cmu_rule, &up_rule_1, &up_rule_2, &up_rule_3, &cmpu_rule]
);
let cfg = SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
s3,
Ec2Client::new(&cfg),
);
let tmp =
std::env::temp_dir().join(format!("ebman-test-multipart-{}.bin", std::process::id()));
let bytes = vec![0xABu8; 17];
std::fs::write(&tmp, &bytes).expect("write tempfile");
let res = client.upload_bundle_with(BUCKET, KEY, &tmp, 1, 8).await;
let _ = std::fs::remove_file(&tmp);
res.expect("multipart upload should succeed");
assert_eq!(cmu_rule.num_calls(), 1, "CreateMultipartUpload");
assert_eq!(up_rule_1.num_calls(), 1, "UploadPart #1");
assert_eq!(up_rule_2.num_calls(), 1, "UploadPart #2");
assert_eq!(up_rule_3.num_calls(), 1, "UploadPart #3");
assert_eq!(cmpu_rule.num_calls(), 1, "CompleteMultipartUpload");
}
#[tokio::test]
async fn upload_bundle_aborts_multipart_on_upload_part_failure() {
use aws_sdk_s3::operation::create_multipart_upload::CreateMultipartUploadOutput;
use aws_sdk_s3::operation::upload_part::{UploadPartError, UploadPartOutput};
use aws_smithy_mocks::mock;
const BUCKET: &str = "elasticbeanstalk-eu-west-2-123";
const KEY: &str = "applications/abort-test/v1";
const UPLOAD_ID: &str = "test-abort-upload-id";
let cmu_rule = mock!(aws_sdk_s3::Client::create_multipart_upload).then_output(|| {
CreateMultipartUploadOutput::builder()
.upload_id(UPLOAD_ID)
.build()
});
let up_ok = mock!(aws_sdk_s3::Client::upload_part)
.match_requests(|req| req.part_number() == Some(1))
.then_output(|| UploadPartOutput::builder().e_tag("\"etag-1\"").build());
let up_fail = mock!(aws_sdk_s3::Client::upload_part)
.match_requests(|req| req.part_number() == Some(2))
.then_error(|| {
UploadPartError::unhandled(
aws_smithy_types::error::ErrorMetadata::builder().build(),
)
});
let abort_rule = mock!(aws_sdk_s3::Client::abort_multipart_upload)
.match_requests(|req| {
req.bucket() == Some(BUCKET)
&& req.key() == Some(KEY)
&& req.upload_id() == Some(UPLOAD_ID)
})
.then_output(|| {
aws_sdk_s3::operation::abort_multipart_upload::AbortMultipartUploadOutput::builder()
.build()
});
let s3 = mock_client!(aws_sdk_s3, [&cmu_rule, &up_ok, &up_fail, &abort_rule]);
let cfg = SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
s3,
Ec2Client::new(&cfg),
);
let tmp = std::env::temp_dir().join(format!("ebman-test-abort-{}.bin", std::process::id()));
std::fs::write(&tmp, vec![0xCDu8; 16]).expect("write tempfile");
let res = client.upload_bundle_with(BUCKET, KEY, &tmp, 1, 8).await;
let _ = std::fs::remove_file(&tmp);
assert!(res.is_err(), "upload should surface UploadPart failure");
assert_eq!(abort_rule.num_calls(), 1, "AbortMultipartUpload must fire");
}
#[test]
fn derive_dlq_url_skips_already_dlq() {
assert_eq!(
derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/foo-dlq"),
None
);
}
#[test]
fn derive_dlq_url_strips_trailing_slash() {
assert_eq!(
derive_dlq_url("https://sqs.us-east-1.amazonaws.com/123/foo/"),
Some("https://sqs.us-east-1.amazonaws.com/123/foo-dlq".to_string())
);
}
use aws_smithy_mocks::{mock, mock_client};
fn client_with_eb(eb: Client) -> AwsClient {
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
AwsClient::for_tests(
eb,
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
)
}
fn client_with_cw(cw: CwClient) -> AwsClient {
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
cw,
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
)
}
fn client_with_eb_and_s3(eb: Client, s3: S3Client) -> AwsClient {
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
AwsClient::for_tests(
eb,
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
s3,
Ec2Client::new(&cfg),
)
}
fn client_with_sqs(sqs: SqsClient) -> AwsClient {
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
AwsClient::for_tests(
Client::new(&cfg),
sqs,
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
)
}
macro_rules! client_with_sub {
($field:ident = $value:expr) => {{
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let mut c = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
);
c.$field = $value;
c
}};
}
#[tokio::test]
async fn list_secrets_maps_secretlistentry_to_summary() {
use aws_sdk_secretsmanager::operation::list_secrets::ListSecretsOutput;
use aws_sdk_secretsmanager::types::SecretListEntry;
use aws_smithy_types::DateTime as SmithyDt;
let rule = mock!(aws_sdk_secretsmanager::Client::list_secrets).then_output(|| {
ListSecretsOutput::builder()
.secret_list(
SecretListEntry::builder()
.name("prod/db-password")
.arn("arn:aws:secretsmanager:us-east-1:123:secret:prod/db-password-AbCdEf")
.description("Production DB master password")
.last_changed_date(SmithyDt::from_secs(1_700_000_000))
.build(),
)
.secret_list(
SecretListEntry::builder()
.name("staging/api-key")
.arn("arn:aws:secretsmanager:us-east-1:123:secret:staging/api-key-XyZ")
.last_changed_date(SmithyDt::from_secs(1_600_000_000))
.build(),
)
.build()
});
let secrets = mock_client!(aws_sdk_secretsmanager, [&rule]);
let client = client_with_sub!(secrets = secrets);
let all = client.list_secrets(None).await.expect("ok");
assert_eq!(all.len(), 2);
assert_eq!(all[0].name, "prod/db-password");
assert_eq!(all[1].name, "staging/api-key");
assert_eq!(
all[0].description.as_deref(),
Some("Production DB master password")
);
assert!(all[0].last_changed.is_some());
}
#[tokio::test]
async fn list_certificates_filters_to_issued_and_extracts_domain() {
use aws_sdk_acm::operation::list_certificates::ListCertificatesOutput;
use aws_sdk_acm::types::{CertificateStatus, CertificateSummary};
let rule = mock!(aws_sdk_acm::Client::list_certificates)
.match_requests(|req| {
req.certificate_statuses()
.contains(&CertificateStatus::Issued)
})
.then_output(|| {
ListCertificatesOutput::builder()
.certificate_summary_list(
CertificateSummary::builder()
.certificate_arn("arn:aws:acm:us-east-1:123:certificate/abcd")
.domain_name("*.example.com")
.build(),
)
.certificate_summary_list(
CertificateSummary::builder()
.certificate_arn("arn:aws:acm:us-east-1:123:certificate/efgh")
.domain_name("api.example.com")
.build(),
)
.build()
});
let acm = mock_client!(aws_sdk_acm, [&rule]);
let client = client_with_sub!(acm = acm);
let certs = client.list_certificates().await.expect("ok");
assert_eq!(certs.len(), 2);
assert_eq!(certs[0].domain, "*.example.com");
assert_eq!(certs[1].domain, "api.example.com");
assert_eq!(rule.num_calls(), 1, "ListCertificates fired once");
}
#[tokio::test]
async fn list_org_accounts_sorts_active_first_then_by_name() {
use aws_sdk_organizations::operation::list_accounts::ListAccountsOutput;
use aws_sdk_organizations::types::{Account, AccountStatus};
let rule = mock!(aws_sdk_organizations::Client::list_accounts).then_output(|| {
ListAccountsOutput::builder()
.accounts(
Account::builder()
.id("999999999999")
.name("zzz-closed")
.email("zzz@example.com")
.status(AccountStatus::Suspended)
.build(),
)
.accounts(
Account::builder()
.id("222222222222")
.name("staging")
.email("staging@example.com")
.status(AccountStatus::Active)
.build(),
)
.accounts(
Account::builder()
.id("111111111111")
.name("prod")
.email("prod@example.com")
.status(AccountStatus::Active)
.build(),
)
.build()
});
let org = mock_client!(aws_sdk_organizations, [&rule]);
let client = client_with_sub!(org = org);
let accounts = client.list_org_accounts().await.expect("ok");
assert_eq!(accounts.len(), 3);
assert_eq!(accounts[0].name, "prod");
assert_eq!(accounts[1].name, "staging");
assert_eq!(accounts[2].name, "zzz-closed");
}
#[tokio::test]
async fn fetch_env_costs_extracts_env_name_from_tag_group_key() {
use aws_sdk_costexplorer::operation::get_cost_and_usage::GetCostAndUsageOutput;
use aws_sdk_costexplorer::types::{Granularity, Group, MetricValue, ResultByTime};
let rule = mock!(aws_sdk_costexplorer::Client::get_cost_and_usage)
.match_requests(|req| {
req.granularity() == Some(&Granularity::Monthly)
&& req.metrics().iter().any(|m| m == "UnblendedCost")
&& req
.group_by()
.iter()
.any(|g| g.key() == Some("elasticbeanstalk:environment-name"))
})
.then_output(|| {
let mut metrics = std::collections::HashMap::new();
metrics.insert(
"UnblendedCost".to_string(),
MetricValue::builder().amount("150.25").unit("USD").build(),
);
GetCostAndUsageOutput::builder()
.results_by_time(
ResultByTime::builder()
.groups(
Group::builder()
.keys("elasticbeanstalk:environment-name$uflexi-prod")
.set_metrics(Some(metrics))
.build(),
)
.build(),
)
.build()
});
let cost = mock_client!(aws_sdk_costexplorer, [&rule]);
let client = client_with_sub!(cost = cost);
let costs = client.fetch_env_costs().await.expect("ok");
assert_eq!(costs.len(), 1);
assert_eq!(costs[0].env_name, "uflexi-prod");
assert!(
(costs[0].cost_usd - 150.25).abs() < f64::EPSILON,
"amount parsed from string"
);
}
#[tokio::test]
async fn worker_queues_resolves_via_describe_environment_resources_when_autocreated() {
use aws_sdk_elasticbeanstalk::operation::describe_environment_resources::DescribeEnvironmentResourcesOutput;
use aws_sdk_elasticbeanstalk::types::{EnvironmentResourceDescription, Queue};
let der = mock!(Client::describe_environment_resources).then_output(|| {
DescribeEnvironmentResourcesOutput::builder()
.environment_resources(
EnvironmentResourceDescription::builder()
.queues(
Queue::builder()
.name("WorkerQueue")
.url("https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue")
.build(),
)
.queues(
Queue::builder()
.name("WorkerDeadLetterQueue")
.url(
"https://sqs.us-east-1.amazonaws.com/123/awseb-e-foo-queue-dlq",
)
.build(),
)
.build(),
)
.build()
});
let dcs = mock!(Client::describe_configuration_settings).then_output(|| {
aws_sdk_elasticbeanstalk::operation::describe_configuration_settings::DescribeConfigurationSettingsOutput::builder()
.build()
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&der, &dcs]);
let client = client_with_eb(eb);
let _ = client.describe_worker_queues("eb-app", "eb-env").await;
assert_eq!(
der.num_calls(),
1,
"describe_environment_resources should be the primary path"
);
}
#[tokio::test]
async fn peek_messages_loops_and_dedupes_across_batches() {
use aws_sdk_sqs::operation::receive_message::ReceiveMessageOutput;
use aws_sdk_sqs::types::Message;
fn msg(id: &'static str) -> Message {
Message::builder().message_id(id).body(id).build()
}
let rule = mock!(aws_sdk_sqs::Client::receive_message)
.sequence()
.output(|| {
ReceiveMessageOutput::builder()
.messages(msg("msg-1"))
.messages(msg("msg-2"))
.build()
})
.output(|| {
ReceiveMessageOutput::builder()
.messages(msg("msg-1")) .messages(msg("msg-3"))
.build()
})
.output(|| ReceiveMessageOutput::builder().build())
.output(|| ReceiveMessageOutput::builder().build())
.build();
let sqs = mock_client!(aws_sdk_sqs, [&rule]);
let client = client_with_sqs(sqs);
let out = client
.peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 10)
.await
.expect("peek should succeed");
let ids: Vec<String> = out.iter().map(|m| m.id.clone()).collect();
assert_eq!(ids, vec!["msg-1", "msg-2", "msg-3"]);
}
#[tokio::test]
async fn peek_messages_stops_after_two_empty_batches() {
use aws_sdk_sqs::operation::receive_message::ReceiveMessageOutput;
let rule = mock!(aws_sdk_sqs::Client::receive_message)
.sequence()
.output(|| ReceiveMessageOutput::builder().build())
.output(|| ReceiveMessageOutput::builder().build())
.output(|| {
ReceiveMessageOutput::builder()
.messages(
aws_sdk_sqs::types::Message::builder()
.message_id("late")
.body("late")
.build(),
)
.build()
})
.build();
let sqs = mock_client!(aws_sdk_sqs, [&rule]);
let client = client_with_sqs(sqs);
let out = client
.peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 10)
.await
.expect("peek should succeed");
assert!(
out.is_empty(),
"should have stopped before consuming the 'late' message"
);
assert_eq!(
rule.num_calls(),
2,
"exactly two empty-batch calls should terminate the loop"
);
}
#[tokio::test]
async fn list_environments_maps_describe_environments_to_env_rows() {
use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsOutput;
use aws_sdk_elasticbeanstalk::types::{EnvironmentDescription, EnvironmentTier};
let de = mock!(Client::describe_environments).then_output(|| {
DescribeEnvironmentsOutput::builder()
.environments(
EnvironmentDescription::builder()
.environment_name("api-prod")
.application_name("api")
.status("Ready".into())
.health("Green".into())
.cname("api-prod.eba.amazonaws.com")
.version_label("build-42")
.solution_stack_name("64bit Amazon Linux 2 v3.5.0 running Java 17")
.tier(EnvironmentTier::builder().name("WebServer").build())
.build(),
)
.build()
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&de]);
let client = client_with_eb(eb);
let envs = client.list_environments().await.expect("ok");
assert_eq!(envs.len(), 1);
let e = &envs[0];
assert_eq!(e.name, "api-prod");
assert_eq!(e.application, "api");
assert_eq!(e.tier, "Web", "tier normalises WebServer → Web");
assert_eq!(e.platform, "Java 17");
assert_eq!(e.version_label, "build-42");
}
#[tokio::test]
async fn list_application_versions_pages_through_next_token() {
use aws_sdk_elasticbeanstalk::operation::describe_application_versions::DescribeApplicationVersionsOutput;
use aws_sdk_elasticbeanstalk::types::ApplicationVersionDescription;
let page1 = mock!(Client::describe_application_versions)
.match_requests(|req| {
req.application_name() == Some("uflexi") && req.next_token().is_none()
})
.then_output(|| {
DescribeApplicationVersionsOutput::builder()
.application_versions(
ApplicationVersionDescription::builder()
.version_label("build-101")
.description("first")
.build(),
)
.application_versions(
ApplicationVersionDescription::builder()
.version_label("build-100")
.description("zeroth")
.build(),
)
.next_token("PAGE_2")
.build()
});
let page2 = mock!(Client::describe_application_versions)
.match_requests(|req| req.next_token() == Some("PAGE_2"))
.then_output(|| {
DescribeApplicationVersionsOutput::builder()
.application_versions(
ApplicationVersionDescription::builder()
.version_label("build-099")
.description("rolled")
.build(),
)
.build()
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&page1, &page2]);
let client = client_with_eb(eb);
let versions = client
.list_application_versions("uflexi")
.await
.expect("ok");
let labels: Vec<&str> = versions.iter().map(|v| v.label.as_str()).collect();
assert_eq!(
labels,
vec!["build-101", "build-100", "build-099"],
"all three versions from both pages should be returned",
);
assert_eq!(page1.num_calls(), 1, "first page fetched once");
assert_eq!(page2.num_calls(), 1, "second page fetched once");
}
#[test]
fn split_csv_trims_and_drops_empties() {
assert_eq!(
split_csv("subnet-a,subnet-b, subnet-c, ,subnet-d"),
vec!["subnet-a", "subnet-b", "subnet-c", "subnet-d"]
);
assert!(split_csv("").is_empty());
assert!(split_csv(",,,").is_empty());
}
#[tokio::test]
async fn fetch_env_vpc_context_pulls_vpc_id_subnets_and_sgs() {
use aws_sdk_elasticbeanstalk::operation::describe_configuration_settings::DescribeConfigurationSettingsOutput;
use aws_sdk_elasticbeanstalk::types::{
ConfigurationOptionSetting, ConfigurationSettingsDescription,
};
let dcs = mock!(Client::describe_configuration_settings).then_output(|| {
DescribeConfigurationSettingsOutput::builder()
.configuration_settings(
ConfigurationSettingsDescription::builder()
.option_settings(
ConfigurationOptionSetting::builder()
.namespace("aws:ec2:vpc")
.option_name("VPCId")
.value("vpc-123")
.build(),
)
.option_settings(
ConfigurationOptionSetting::builder()
.namespace("aws:ec2:vpc")
.option_name("Subnets")
.value("subnet-a,subnet-b")
.build(),
)
.option_settings(
ConfigurationOptionSetting::builder()
.namespace("aws:ec2:vpc")
.option_name("ELBSubnets")
.value("subnet-x,subnet-y")
.build(),
)
.option_settings(
ConfigurationOptionSetting::builder()
.namespace("aws:autoscaling:launchconfiguration")
.option_name("SecurityGroups")
.value("sg-1,sg-2,sg-3")
.build(),
)
.option_settings(
ConfigurationOptionSetting::builder()
.namespace("aws:elasticbeanstalk:application:environment")
.option_name("LOG_LEVEL")
.value("debug")
.build(),
)
.build(),
)
.build()
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&dcs]);
let client = client_with_eb(eb);
let ctx = client
.fetch_env_vpc_context("api", "api-prod")
.await
.expect("ok");
assert_eq!(ctx.vpc_id.as_deref(), Some("vpc-123"));
assert_eq!(ctx.subnets, vec!["subnet-a", "subnet-b"]);
assert_eq!(ctx.elb_subnets, vec!["subnet-x", "subnet-y"]);
assert_eq!(ctx.security_groups, vec!["sg-1", "sg-2", "sg-3"]);
}
#[tokio::test]
async fn list_subnets_in_vpc_filters_orders_and_extracts_name_tag() {
use aws_sdk_ec2::operation::describe_subnets::DescribeSubnetsOutput;
use aws_sdk_ec2::types::{Subnet, Tag};
let ds = mock!(aws_sdk_ec2::Client::describe_subnets).then_output(|| {
DescribeSubnetsOutput::builder()
.subnets(
Subnet::builder()
.subnet_id("subnet-2b")
.availability_zone("us-east-1b")
.cidr_block("10.0.2.0/24")
.tags(Tag::builder().key("Name").value("private-2b").build())
.build(),
)
.subnets(
Subnet::builder()
.subnet_id("subnet-1a")
.availability_zone("us-east-1a")
.cidr_block("10.0.1.0/24")
.build(),
)
.subnets(
Subnet::builder()
.subnet_id("subnet-1a-overlap")
.availability_zone("us-east-1a")
.cidr_block("10.0.0.0/24")
.build(),
)
.build()
});
let ec2 = mock_client!(aws_sdk_ec2, [&ds]);
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
ec2,
);
let subnets = client.list_subnets_in_vpc("vpc-abc").await.expect("ok");
let ids: Vec<&str> = subnets.iter().map(|s| s.id.as_str()).collect();
assert_eq!(ids, vec!["subnet-1a-overlap", "subnet-1a", "subnet-2b"]);
assert_eq!(subnets[2].name_tag.as_deref(), Some("private-2b"));
assert!(subnets[1].name_tag.is_none());
}
#[tokio::test]
async fn update_env_option_settings_builds_correct_request_shape() {
use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
let rule = mock!(Client::update_environment)
.match_requests(|input| {
if input.environment_name.as_deref() != Some("api-prod") {
return false;
}
let options = input.option_settings();
if options.len() != 2 {
return false;
}
if options[0].namespace.as_deref() != Some("aws:autoscaling:asg")
|| options[0].option_name.as_deref() != Some("MinSize")
|| options[0].value.as_deref() != Some("2")
{
return false;
}
if options[1].namespace.as_deref() != Some("aws:autoscaling:launchconfiguration")
|| options[1].option_name.as_deref() != Some("InstanceType")
|| options[1].value.as_deref() != Some("t3.medium")
{
return false;
}
let removes = input.options_to_remove();
if removes.len() != 1 {
return false;
}
removes[0].namespace.as_deref()
== Some("aws:elasticbeanstalk:application:environment")
&& removes[0].option_name.as_deref() == Some("OLD_VAR")
})
.then_output(|| UpdateEnvironmentOutput::builder().build());
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
let client = client_with_eb(eb);
let to_set = vec![
(
"aws:autoscaling:asg".to_string(),
"MinSize".to_string(),
"2".to_string(),
),
(
"aws:autoscaling:launchconfiguration".to_string(),
"InstanceType".to_string(),
"t3.medium".to_string(),
),
];
let to_remove = vec![(
"aws:elasticbeanstalk:application:environment".to_string(),
"OLD_VAR".to_string(),
)];
client
.update_env_option_settings("api-prod", &to_set, &to_remove)
.await
.expect("expected request shape to match");
assert_eq!(rule.num_calls(), 1);
}
#[tokio::test]
async fn update_env_option_settings_rejects_empty_input_before_dispatch() {
use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
let trip = mock!(Client::update_environment)
.then_output(|| UpdateEnvironmentOutput::builder().build());
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&trip]);
let client = client_with_eb(eb);
let err = client
.update_env_option_settings("api-prod", &[], &[])
.await
.expect_err("expected guard to fire");
assert!(
err.to_string().contains("nothing to do"),
"expected nothing-to-do guard, got {err}"
);
assert_eq!(
trip.num_calls(),
0,
"guard should short-circuit before any SDK call"
);
}
#[tokio::test]
async fn update_env_option_settings_surfaces_aws_errors() {
use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentError;
use aws_sdk_elasticbeanstalk::types::error::InsufficientPrivilegesException;
let err_rule = mock!(Client::update_environment).then_error(|| {
UpdateEnvironmentError::InsufficientPrivilegesException(
InsufficientPrivilegesException::builder()
.message("not authorized to call UpdateEnvironment")
.build(),
)
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&err_rule]);
let client = client_with_eb(eb);
let err = client
.update_env_option_settings(
"api-prod",
&[("aws:autoscaling:asg".into(), "MinSize".into(), "2".into())],
&[],
)
.await
.expect_err("expected AWS error to propagate");
assert!(
err.to_string()
.contains("UpdateEnvironment(option_settings)"),
"expected wrapped error context, got {err}"
);
}
#[tokio::test]
async fn list_security_groups_in_vpc_orders_by_name() {
use aws_sdk_ec2::operation::describe_security_groups::DescribeSecurityGroupsOutput;
use aws_sdk_ec2::types::SecurityGroup;
let dsg = mock!(aws_sdk_ec2::Client::describe_security_groups).then_output(|| {
DescribeSecurityGroupsOutput::builder()
.security_groups(
SecurityGroup::builder()
.group_id("sg-z")
.group_name("zeta")
.description("z group")
.build(),
)
.security_groups(
SecurityGroup::builder()
.group_id("sg-a")
.group_name("alpha")
.description("a group")
.build(),
)
.build()
});
let ec2 = mock_client!(aws_sdk_ec2, [&dsg]);
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
ec2,
);
let sgs = client
.list_security_groups_in_vpc("vpc-abc")
.await
.expect("ok");
assert_eq!(sgs.len(), 2);
assert_eq!(sgs[0].group_name, "alpha");
assert_eq!(sgs[1].group_name, "zeta");
}
#[tokio::test]
async fn list_environments_throttling_error_is_recognised_by_predicate() {
use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
let rule = mock!(Client::describe_environments).then_error(|| {
DescribeEnvironmentsError::generic(
aws_smithy_types::error::ErrorMetadata::builder()
.code("ThrottlingException")
.message("Rate exceeded")
.build(),
)
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
let client = client_with_eb(eb);
let err = client
.list_environments()
.await
.expect_err("expected throttling error to propagate");
let s = crate::app::flatten_err_to_string(&err);
assert!(
crate::app::is_throttling_error(&s),
"is_throttling_error should fire on the flattened SDK throttling string, got {s:?}"
);
assert!(
!s.contains("StatusCode") && !s.contains("Extensions"),
"throttling toast should be clean, got {s:?}"
);
}
#[tokio::test]
async fn list_environments_expired_token_surfaces_clean_user_message() {
use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
let rule = mock!(Client::describe_environments).then_error(|| {
DescribeEnvironmentsError::generic(
aws_smithy_types::error::ErrorMetadata::builder()
.code("ExpiredTokenException")
.message("The security token included in the request is expired")
.build(),
)
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
let client = client_with_eb(eb);
let err = client
.list_environments()
.await
.expect_err("expected expired-token error to propagate");
let s = crate::app::flatten_err_to_string(&err);
assert!(
!crate::app::is_throttling_error(&s),
"ExpiredToken should not fire the throttling predicate, got {s:?}"
);
assert!(
!s.contains("StatusCode") && !s.contains("Extensions") && !s.contains("SdkBody"),
"expired-token toast should be clean, got {s:?}"
);
}
#[tokio::test]
async fn fetch_env_metrics_batches_and_reorders_by_canonical_id() {
use aws_sdk_cloudwatch::operation::get_metric_data::GetMetricDataOutput;
use aws_sdk_cloudwatch::types::MetricDataResult;
use aws_smithy_types::DateTime as SdkDateTime;
let ts = SdkDateTime::from_secs(1_700_000_000);
let mk_result = move |id: &str, value: f64| {
MetricDataResult::builder()
.id(id)
.timestamps(ts)
.values(value)
.build()
};
let rule = mock!(aws_sdk_cloudwatch::Client::get_metric_data)
.match_requests(|req| {
let ids: Vec<&str> = req
.metric_data_queries()
.iter()
.filter_map(|q| q.id())
.collect();
ids == ["health", "req4xx", "req5xx", "p90"]
})
.then_output(move || {
GetMetricDataOutput::builder()
.metric_data_results(mk_result("req5xx", 12.0))
.metric_data_results(mk_result("health", 25.0))
.metric_data_results(mk_result("p90", 0.42))
.metric_data_results(mk_result("req4xx", 3.0))
.build()
});
let cw = mock_client!(aws_sdk_cloudwatch, [&rule]);
let client = client_with_cw(cw);
let series = client
.fetch_env_metrics("uflexi-prod", 900)
.await
.expect("metric fetch should succeed");
assert_eq!(rule.num_calls(), 1, "expected exactly one batched call");
let ids: Vec<&str> = series.iter().map(|s| s.id.as_str()).collect();
assert_eq!(ids, vec!["health", "req4xx", "req5xx", "p90"]);
let by_id: std::collections::HashMap<&str, &str> = series
.iter()
.map(|s| (s.id.as_str(), s.label.as_str()))
.collect();
assert_eq!(by_id["health"], "Env Health (0–25)");
assert_eq!(by_id["req4xx"], "4xx Requests / min");
assert_eq!(by_id["req5xx"], "5xx Requests / min");
assert_eq!(by_id["p90"], "Latency P90");
let p90 = series.iter().find(|s| s.id == "p90").unwrap();
assert_eq!(p90.points.len(), 1);
assert!((p90.points[0].1 - 0.42).abs() < f64::EPSILON);
}
#[tokio::test]
async fn deploy_from_path_chain_dispatches_each_stage() {
use aws_sdk_elasticbeanstalk::operation::create_application_version::CreateApplicationVersionOutput;
use aws_sdk_elasticbeanstalk::operation::create_storage_location::CreateStorageLocationOutput;
use aws_sdk_elasticbeanstalk::operation::update_environment::UpdateEnvironmentOutput;
use aws_sdk_s3::operation::put_object::PutObjectOutput;
const BUCKET: &str = "elasticbeanstalk-us-east-1-123456789012";
const APP: &str = "uflexi-webapp";
const ENV: &str = "uflexi-prod";
const LABEL: &str = "build-2026-05-20-1234567890";
const KEY: &str = "applications/uflexi-webapp/build-2026-05-20-1234567890";
let bundle_bytes: Vec<u8> = b"PK\x03\x04 ... a real zip would start here".to_vec();
let csl_rule = mock!(Client::create_storage_location).then_output(|| {
CreateStorageLocationOutput::builder()
.s3_bucket(BUCKET)
.build()
});
let put_rule = mock!(aws_sdk_s3::Client::put_object)
.match_requests(|req| req.bucket() == Some(BUCKET) && req.key() == Some(KEY))
.then_output(|| PutObjectOutput::builder().build());
let cav_rule = mock!(Client::create_application_version)
.match_requests(|req| {
req.application_name() == Some(APP)
&& req.version_label() == Some(LABEL)
&& req.source_bundle().and_then(|s| s.s3_bucket()) == Some(BUCKET)
&& req.source_bundle().and_then(|s| s.s3_key()) == Some(KEY)
&& req.auto_create_application() == Some(false)
})
.then_output(|| CreateApplicationVersionOutput::builder().build());
let upd_rule = mock!(Client::update_environment)
.match_requests(|req| {
req.environment_name() == Some(ENV) && req.version_label() == Some(LABEL)
})
.then_output(|| UpdateEnvironmentOutput::builder().build());
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&csl_rule, &cav_rule, &upd_rule]);
let s3 = mock_client!(aws_sdk_s3, [&put_rule]);
let client = client_with_eb_and_s3(eb, s3);
let bucket = client
.create_storage_location()
.await
.expect("CreateStorageLocation should return the managed bucket");
assert_eq!(bucket, BUCKET);
let tmp =
std::env::temp_dir().join(format!("ebman-test-bundle-{}.zip", std::process::id()));
std::fs::write(&tmp, &bundle_bytes).expect("write tempfile");
let upload_res = client
.upload_bundle_with(&bucket, KEY, &tmp, u64::MAX, 8 * 1024 * 1024)
.await;
let _ = std::fs::remove_file(&tmp);
upload_res.expect("PutObject should succeed");
client
.create_app_version(APP, LABEL, Some("test deploy"), &bucket, KEY)
.await
.expect("CreateApplicationVersion should succeed");
client
.deploy_version(ENV, LABEL)
.await
.expect("UpdateEnvironment should succeed");
assert_eq!(csl_rule.num_calls(), 1, "CreateStorageLocation");
assert_eq!(put_rule.num_calls(), 1, "S3 PutObject");
assert_eq!(cav_rule.num_calls(), 1, "CreateApplicationVersion");
assert_eq!(upd_rule.num_calls(), 1, "UpdateEnvironment");
}
#[tokio::test]
async fn list_environments_surfaces_aws_errors_with_op_context() {
use aws_sdk_elasticbeanstalk::operation::describe_environments::DescribeEnvironmentsError;
let rule = mock!(Client::describe_environments).then_error(|| {
DescribeEnvironmentsError::generic(
aws_smithy_types::error::ErrorMetadata::builder()
.code("InternalServerError")
.message("retry later")
.build(),
)
});
let eb = mock_client!(aws_sdk_elasticbeanstalk, [&rule]);
let client = client_with_eb(eb);
let err = client
.list_environments()
.await
.expect_err("expected AWS error to propagate");
assert!(
err.to_string().contains("DescribeEnvironments"),
"expected operation context, got {err}"
);
}
#[tokio::test]
async fn peek_messages_surfaces_sqs_errors_with_op_context() {
use aws_sdk_sqs::operation::receive_message::ReceiveMessageError;
let rule = mock!(aws_sdk_sqs::Client::receive_message).then_error(|| {
ReceiveMessageError::generic(
aws_smithy_types::error::ErrorMetadata::builder()
.code("QueueDoesNotExist")
.message("queue gone")
.build(),
)
});
let sqs = mock_client!(aws_sdk_sqs, [&rule]);
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
sqs,
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
Ec2Client::new(&cfg),
);
let err = client
.peek_messages("https://sqs.us-east-1.amazonaws.com/123/q", 5)
.await
.expect_err("expected SQS error to propagate");
assert!(
err.to_string().contains("ReceiveMessage"),
"expected operation context, got {err}"
);
}
#[tokio::test]
async fn list_subnets_in_vpc_surfaces_ec2_errors_with_op_context() {
use aws_sdk_ec2::operation::describe_subnets::DescribeSubnetsError;
let rule = mock!(aws_sdk_ec2::Client::describe_subnets).then_error(|| {
DescribeSubnetsError::generic(
aws_smithy_types::error::ErrorMetadata::builder()
.code("InvalidVpcID.NotFound")
.message("vpc-xxx not found")
.build(),
)
});
let ec2 = mock_client!(aws_sdk_ec2, [&rule]);
let cfg = aws_config::SdkConfig::builder()
.region(Region::new("us-east-1"))
.behavior_version(aws_config::BehaviorVersion::latest())
.build();
let client = AwsClient::for_tests(
Client::new(&cfg),
SqsClient::new(&cfg),
CwClient::new(&cfg),
CwLogsClient::new(&cfg),
S3Client::new(&cfg),
ec2,
);
let err = client
.list_subnets_in_vpc("vpc-xxx")
.await
.expect_err("expected EC2 error to propagate");
assert!(
err.to_string().contains("DescribeSubnets"),
"expected operation context, got {err}"
);
}
}