1use crate::config::AwsConfig;
2use anyhow::Result;
3use aws_sdk_rds::types::{DbInstance, PendingModifiedValues};
4use aws_smithy_types::{date_time::Format, DateTime};
5
6mod enrichment;
7pub use enrichment::InventoryOptions;
8
9#[derive(Debug, Clone, Default, PartialEq, Eq)]
12pub struct RdsInstance {
13 pub identifier: String,
14 pub arn: String,
15 pub status: String,
16 pub engine: String,
17 pub engine_version: String,
18 pub instance_class: String,
19 pub region: String,
20 pub availability_zone: String,
21 pub endpoint: String,
22 pub port: Option<i32>,
23 pub cluster_identifier: String,
24 pub role: String,
25 pub upgrade_rollout_order: String,
26 pub extended_support: String,
27 pub recommendations: String,
28 pub cpu: String,
29 pub current_activity: String,
30 pub maintenance: String,
31 pub vpc: String,
32 pub multi_az: Option<bool>,
33 pub storage_type: String,
34 pub storage_gib: Option<i32>,
35 pub iops: Option<i32>,
36 pub storage_throughput: Option<i32>,
37 pub security_groups: String,
38 pub subnet_group: String,
39 pub certificate_expiry: String,
40 pub pending_changes: String,
41 pub character_set: String,
42 pub option_group: String,
43 pub created_time: String,
44 pub encrypted: Option<bool>,
45 pub priority_tier: Option<i32>,
46 pub replica_lag: String,
47 pub replica_state: String,
48 pub parameter_group: String,
49 pub clone_group_id: String,
50 pub is_replica: bool,
51}
52
53pub struct RdsClient {
54 config: AwsConfig,
55}
56
57impl RdsClient {
58 pub fn new(config: AwsConfig) -> Self {
59 Self { config }
60 }
61
62 pub async fn list_instances(&self) -> Result<Vec<RdsInstance>> {
63 let mut stream = self
64 .config
65 .rds_client()
66 .describe_db_instances()
67 .into_paginator()
68 .items()
69 .send();
70 let mut instances = Vec::new();
71 while let Some(instance) = stream.next().await {
72 instances.push(RdsInstance::from_sdk(&instance?, &self.config.region));
73 }
74 instances.sort_by(|a, b| a.identifier.cmp(&b.identifier));
75 Ok(instances)
76 }
77
78 pub async fn list_instances_with_options(
79 &self,
80 options: InventoryOptions,
81 ) -> Result<Vec<RdsInstance>> {
82 let mut instances = self.list_instances().await?;
83 self.enrich(&mut instances, options).await;
84 Ok(instances)
85 }
86}
87
88fn timestamp(value: Option<&DateTime>) -> String {
89 value
90 .and_then(|time| time.fmt(Format::DateTime).ok())
91 .unwrap_or_default()
92}
93
94fn pending_changes(values: Option<&PendingModifiedValues>) -> String {
95 let Some(v) = values else {
96 return String::new();
97 };
98 let changes = [
101 (v.db_instance_class().is_some(), "Size"),
102 (v.allocated_storage().is_some(), "Storage"),
103 (v.master_user_password().is_some(), "Password"),
104 (v.port().is_some(), "Port"),
105 (v.backup_retention_period().is_some(), "Backup retention"),
106 (v.multi_az().is_some(), "Multi-AZ"),
107 (v.engine_version().is_some(), "Engine version"),
108 (v.license_model().is_some(), "License"),
109 (v.iops().is_some(), "IOPS"),
110 (v.storage_throughput().is_some(), "Storage throughput"),
111 (v.db_instance_identifier().is_some(), "DB identifier"),
112 (v.storage_type().is_some(), "Storage type"),
113 (v.ca_certificate_identifier().is_some(), "Certificate"),
114 (v.db_subnet_group_name().is_some(), "Subnet group"),
115 (v.pending_cloudwatch_logs_exports().is_some(), "Log exports"),
116 (!v.processor_features().is_empty(), "Processor features"),
117 (v.automation_mode().is_some(), "Automation mode"),
118 (
119 v.resume_full_automation_mode_time().is_some(),
120 "Resume automation",
121 ),
122 (v.multi_tenant().is_some(), "Multi-tenant"),
123 (
124 v.iam_database_authentication_enabled().is_some(),
125 "IAM authentication",
126 ),
127 (v.dedicated_log_volume().is_some(), "Dedicated log volume"),
128 (v.engine().is_some(), "Engine"),
129 (
130 !v.additional_storage_volumes().is_empty(),
131 "Additional storage volumes",
132 ),
133 ];
134 changes
135 .into_iter()
136 .filter_map(|(pending, label)| pending.then_some(label))
137 .collect::<Vec<_>>()
138 .join(", ")
139}
140
141impl RdsInstance {
142 fn from_sdk(instance: &DbInstance, region: &str) -> Self {
143 let is_replica = instance
144 .read_replica_source_db_instance_identifier()
145 .is_some()
146 || instance
147 .read_replica_source_db_cluster_identifier()
148 .is_some();
149 let cluster_identifier = instance
150 .db_cluster_identifier()
151 .unwrap_or_default()
152 .to_string();
153 let role = if !cluster_identifier.is_empty() {
154 String::new()
155 } else if is_replica {
156 "Read replica".into()
157 } else if instance.db_instance_identifier().is_some() {
158 "Instance".into()
159 } else {
160 String::new()
161 };
162 Self {
163 identifier: instance.db_instance_identifier().unwrap_or_default().into(),
164 arn: instance.db_instance_arn().unwrap_or_default().into(),
165 status: instance.db_instance_status().unwrap_or_default().into(),
166 engine: instance.engine().unwrap_or_default().into(),
167 engine_version: instance.engine_version().unwrap_or_default().into(),
168 instance_class: instance.db_instance_class().unwrap_or_default().into(),
169 region: region.into(),
170 availability_zone: instance.availability_zone().unwrap_or_default().into(),
171 endpoint: instance
172 .endpoint()
173 .and_then(|e| e.address())
174 .unwrap_or_default()
175 .into(),
176 port: instance.endpoint().and_then(|e| e.port()),
177 cluster_identifier,
178 role,
179 is_replica,
180 upgrade_rollout_order: instance
181 .upgrade_rollout_order()
182 .map(|v| v.as_str().into())
183 .unwrap_or_default(),
184 extended_support: match instance.engine_lifecycle_support() {
185 Some("open-source-rds-extended-support") => "Enabled".into(),
186 Some("open-source-rds-extended-support-disabled") => "Disabled".into(),
187 value => value.unwrap_or_default().into(),
188 },
189 vpc: instance
190 .db_subnet_group()
191 .and_then(|g| g.vpc_id())
192 .unwrap_or_default()
193 .into(),
194 multi_az: instance.multi_az(),
195 storage_type: instance.storage_type().unwrap_or_default().into(),
196 storage_gib: instance.allocated_storage(),
197 iops: instance.iops(),
198 storage_throughput: instance.storage_throughput(),
199 security_groups: instance
200 .vpc_security_groups()
201 .iter()
202 .filter_map(|g| g.vpc_security_group_id())
203 .chain(
204 instance
205 .db_security_groups()
206 .iter()
207 .filter_map(|g| g.db_security_group_name()),
208 )
209 .collect::<Vec<_>>()
210 .join(", "),
211 subnet_group: instance
212 .db_subnet_group()
213 .and_then(|g| g.db_subnet_group_name())
214 .unwrap_or_default()
215 .into(),
216 certificate_expiry: timestamp(
217 instance.certificate_details().and_then(|c| c.valid_till()),
218 ),
219 pending_changes: pending_changes(instance.pending_modified_values()),
220 character_set: [
221 instance.character_set_name(),
222 instance.nchar_character_set_name(),
223 ]
224 .into_iter()
225 .flatten()
226 .collect::<Vec<_>>()
227 .join(", "),
228 option_group: instance
229 .option_group_memberships()
230 .iter()
231 .filter_map(|g| g.option_group_name())
232 .collect::<Vec<_>>()
233 .join(", "),
234 created_time: timestamp(instance.instance_create_time()),
235 encrypted: instance.storage_encrypted(),
236 priority_tier: instance.promotion_tier(),
237 replica_state: instance
238 .status_infos()
239 .iter()
240 .filter(|s| s.status_type() == Some("read replication"))
241 .filter_map(|s| s.status())
242 .collect::<Vec<_>>()
243 .join(", "),
244 parameter_group: instance
245 .db_parameter_groups()
246 .iter()
247 .filter_map(|g| g.db_parameter_group_name())
248 .collect::<Vec<_>>()
249 .join(", "),
250 ..Default::default()
251 }
252 }
253}
254
255#[cfg(test)]
256mod tests;