rusticity_core/rds/
enrichment.rs

1//! Extra reads are opt-in per visible column; failures never hide the inventory.
2use super::{RdsClient, RdsInstance};
3use anyhow::Result;
4use aws_sdk_cloudwatch::types::{
5    Dimension, Metric, MetricDataQuery, MetricStat, ScanBy, StatusCode,
6};
7use aws_sdk_rds::types::{DbCluster, DbRecommendation, ResourcePendingMaintenanceActions};
8use aws_smithy_types::DateTime;
9use std::collections::HashMap;
10
11#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
12pub struct InventoryOptions {
13    pub clusters: bool,
14    pub recommendations: bool,
15    pub maintenance: bool,
16    pub cpu: bool,
17    pub current_activity: bool,
18    pub replica_lag: bool,
19}
20
21impl InventoryOptions {
22    pub fn needs_metrics(self) -> bool {
23        self.cpu || self.current_activity || self.replica_lag
24    }
25}
26
27impl RdsClient {
28    pub(super) async fn enrich(&self, instances: &mut [RdsInstance], options: InventoryOptions) {
29        if instances.is_empty() {
30            return;
31        }
32        if options.clusters && instances.iter().any(|i| !i.cluster_identifier.is_empty()) {
33            match self.clusters().await {
34                Ok(clusters) => apply_clusters(instances, &clusters),
35                Err(error) => {
36                    tracing::warn!("RDS cluster metadata unavailable: {error}");
37                    for instance in instances
38                        .iter_mut()
39                        .filter(|i| !i.cluster_identifier.is_empty())
40                    {
41                        instance.role = "Unavailable".into();
42                        instance.clone_group_id = "Unavailable".into();
43                    }
44                }
45            }
46        }
47        if options.recommendations {
48            match self.recommendations().await {
49                Ok(rows) => apply_recommendations(instances, &rows),
50                Err(error) => {
51                    tracing::warn!("RDS recommendations unavailable: {error}");
52                    for instance in instances.iter_mut() {
53                        instance.recommendations = "Unavailable".into();
54                    }
55                }
56            }
57        }
58        if options.maintenance {
59            match self.maintenance().await {
60                Ok(rows) => apply_maintenance(instances, &rows),
61                Err(error) => {
62                    tracing::warn!("RDS maintenance unavailable: {error}");
63                    for instance in instances.iter_mut() {
64                        instance.maintenance = "Unavailable".into();
65                    }
66                }
67            }
68        }
69        if options.needs_metrics() {
70            // At most 300 metrics per request (CloudWatch limit: 500).
71            for chunk in instances.chunks_mut(100) {
72                if let Err(error) = self.metrics(chunk, options).await {
73                    tracing::warn!("RDS metrics unavailable: {error}");
74                    for instance in chunk {
75                        if options.cpu {
76                            instance.cpu = "Unavailable".into();
77                        }
78                        if options.current_activity {
79                            instance.current_activity = "Unavailable".into();
80                        }
81                        if options.replica_lag && is_replica_candidate(instance) {
82                            instance.replica_lag = "Unavailable".into();
83                        }
84                    }
85                }
86            }
87        }
88    }
89
90    async fn clusters(&self) -> Result<Vec<DbCluster>> {
91        Ok(self
92            .config
93            .rds_client()
94            .describe_db_clusters()
95            .into_paginator()
96            .items()
97            .send()
98            .collect::<Result<Vec<_>, _>>()
99            .await?)
100    }
101    async fn recommendations(&self) -> Result<Vec<DbRecommendation>> {
102        Ok(self
103            .config
104            .rds_client()
105            .describe_db_recommendations()
106            .into_paginator()
107            .items()
108            .send()
109            .collect::<Result<Vec<_>, _>>()
110            .await?)
111    }
112    async fn maintenance(&self) -> Result<Vec<ResourcePendingMaintenanceActions>> {
113        Ok(self
114            .config
115            .rds_client()
116            .describe_pending_maintenance_actions()
117            .into_paginator()
118            .items()
119            .send()
120            .collect::<Result<Vec<_>, _>>()
121            .await?)
122    }
123
124    async fn metrics(
125        &self,
126        instances: &mut [RdsInstance],
127        options: InventoryOptions,
128    ) -> Result<()> {
129        let mut queries = Vec::new();
130        let mut targets = HashMap::new();
131        for (index, instance) in instances.iter().enumerate() {
132            let lag_name = if instance.engine.starts_with("aurora") {
133                "AuroraReplicaLag"
134            } else {
135                "ReplicaLag"
136            };
137            for (enabled, metric, kind) in [
138                (options.cpu, "CPUUtilization", 0),
139                (options.current_activity, "DBLoad", 1),
140                (
141                    options.replica_lag && is_replica_candidate(instance),
142                    lag_name,
143                    2,
144                ),
145            ] {
146                if !enabled {
147                    continue;
148                }
149                let id = format!("m{index}_{kind}");
150                targets.insert(id.clone(), (index, kind));
151                let metric = Metric::builder()
152                    .namespace("AWS/RDS")
153                    .metric_name(metric)
154                    .dimensions(
155                        Dimension::builder()
156                            .name("DBInstanceIdentifier")
157                            .value(&instance.identifier)
158                            .build(),
159                    )
160                    .build();
161                queries.push(
162                    MetricDataQuery::builder()
163                        .id(id)
164                        .return_data(true)
165                        .metric_stat(
166                            MetricStat::builder()
167                                .metric(metric)
168                                .period(60)
169                                .stat("Average")
170                                .build(),
171                        )
172                        .build(),
173                );
174            }
175        }
176        if queries.is_empty() {
177            return Ok(());
178        }
179        let end = chrono::Utc::now().timestamp();
180        let mut pages = self
181            .config
182            .cloudwatch_client()
183            .get_metric_data()
184            .set_metric_data_queries(Some(queries))
185            .start_time(DateTime::from_secs(end - 900))
186            .end_time(DateTime::from_secs(end))
187            .scan_by(ScanBy::TimestampDescending)
188            .into_paginator()
189            .send();
190        let mut latest: HashMap<String, (DateTime, f64)> = HashMap::new();
191        let mut failed = Vec::new();
192        while let Some(page) = pages.next().await {
193            for result in page?.metric_data_results() {
194                let id = result.id().unwrap_or_default();
195                if matches!(
196                    result.status_code(),
197                    Some(StatusCode::Forbidden | StatusCode::InternalError)
198                ) {
199                    failed.push(id.to_string());
200                }
201                for (time, value) in result.timestamps().iter().zip(result.values()) {
202                    if value.is_finite() && latest.get(id).is_none_or(|(old, _)| time > old) {
203                        latest.insert(id.into(), (*time, *value));
204                    }
205                }
206            }
207        }
208        for (id, (index, kind)) in targets {
209            let instance = &mut instances[index];
210            let value = if failed.contains(&id) {
211                "Unavailable".into()
212            } else if let Some((_, value)) = latest.get(&id) {
213                match kind {
214                    0 => format!("{value:.1}%"),
215                    1 => format!("{value:.2} AAS"),
216                    _ if *value < 0.0 => String::new(), // ReplicaLag -1 means unknown, not negative lag.
217                    _ if instance.engine.starts_with("aurora") => format!("{value:.1} ms"),
218                    _ => format!("{value:.1} s"),
219                }
220            } else {
221                String::new()
222            };
223            match kind {
224                0 => instance.cpu = value,
225                1 => instance.current_activity = value,
226                _ => instance.replica_lag = value,
227            }
228        }
229        Ok(())
230    }
231}
232
233fn is_replica_candidate(instance: &RdsInstance) -> bool {
234    instance.is_replica || (!instance.cluster_identifier.is_empty() && instance.role != "Writer")
235}
236
237pub(super) fn apply_clusters(instances: &mut [RdsInstance], clusters: &[DbCluster]) {
238    for instance in instances
239        .iter_mut()
240        .filter(|i| !i.cluster_identifier.is_empty())
241    {
242        if let Some(cluster) = clusters
243            .iter()
244            .find(|c| c.db_cluster_identifier() == Some(instance.cluster_identifier.as_str()))
245        {
246            instance.clone_group_id = cluster.clone_group_id().unwrap_or_default().into();
247            if let Some(member) = cluster
248                .db_cluster_members()
249                .iter()
250                .find(|m| m.db_instance_identifier() == Some(instance.identifier.as_str()))
251            {
252                instance.role = match member.is_cluster_writer() {
253                    Some(true) => "Writer",
254                    Some(false) => "Reader",
255                    None => "",
256                }
257                .into();
258                instance.is_replica = member.is_cluster_writer() == Some(false);
259                instance.priority_tier = member.promotion_tier().or(instance.priority_tier);
260            }
261        }
262    }
263}
264
265pub(super) fn apply_recommendations(instances: &mut [RdsInstance], rows: &[DbRecommendation]) {
266    for instance in instances {
267        // Without an ARN, we cannot attribute this resource's optional results.
268        if instance.arn.is_empty() {
269            continue;
270        }
271        instance.recommendations = rows
272            .iter()
273            .filter(|r| {
274                r.resource_arn() == Some(instance.arn.as_str())
275                    && r.status().is_some_and(|s| s.eq_ignore_ascii_case("active"))
276            })
277            .count()
278            .to_string();
279    }
280}
281
282pub(super) fn apply_maintenance(
283    instances: &mut [RdsInstance],
284    rows: &[ResourcePendingMaintenanceActions],
285) {
286    for instance in instances {
287        // Without an ARN, we cannot attribute this resource's optional results.
288        if instance.arn.is_empty() {
289            continue;
290        }
291        instance.maintenance = rows
292            .iter()
293            .filter(|r| r.resource_identifier() == Some(instance.arn.as_str()))
294            .flat_map(|r| r.pending_maintenance_action_details())
295            .filter_map(|a| a.action())
296            .collect::<Vec<_>>()
297            .join(", ");
298        if instance.maintenance.is_empty() {
299            instance.maintenance = "None".into();
300        }
301    }
302}
rusticity_core/rds/enrichment.rs

rusticity_core/rds/
enrichment.rs
