1use anyhow::Result;
2use aws_sdk_cloudwatchlogs::Client;
3use chrono::DateTime;
4
5use crate::config::AwsConfig;
6use crate::types::{LogEvent, LogGroup, LogStream};
7
8pub struct CloudWatchClient {
9 client: Client,
10 config: AwsConfig,
11}
12
13impl CloudWatchClient {
14 pub async fn new(config: AwsConfig) -> Result<Self> {
15 let client = config.cloudwatch_logs_client();
16 Ok(Self { client, config })
17 }
18
19 pub fn config(&self) -> &AwsConfig {
20 &self.config
21 }
22
23 pub fn dummy(config: AwsConfig) -> Self {
24 let aws_config = aws_config::SdkConfig::builder()
25 .behavior_version(aws_config::BehaviorVersion::latest())
26 .region(aws_config::Region::new(config.region.clone()))
27 .build();
28 let client = Client::new(&aws_config);
29 Self { client, config }
30 }
31
32 pub async fn list_log_groups(&self) -> Result<Vec<LogGroup>> {
33 let resp = self.client.describe_log_groups().send().await?;
34
35 Ok(resp.log_groups().iter().map(log_group).collect())
36 }
37
38 pub async fn get_log_group(&self, name: &str) -> Result<Option<LogGroup>> {
41 anyhow::ensure!(!name.is_empty(), "Log group name must not be empty");
42 let response = self
43 .client
44 .describe_log_groups()
45 .log_group_identifiers(name)
46 .send()
47 .await?;
48 Ok(response
49 .log_groups()
50 .iter()
51 .find(|group| group.log_group_name() == Some(name))
52 .map(log_group))
53 }
54
55 pub async fn list_log_streams(&self, log_group: &str) -> Result<Vec<LogStream>> {
56 let mut streams = Vec::new();
57 let mut pages = self
58 .client
59 .describe_log_streams()
60 .log_group_name(log_group)
61 .order_by(aws_sdk_cloudwatchlogs::types::OrderBy::LastEventTime)
62 .descending(true)
63 .into_paginator()
64 .send();
65
66 while let Some(resp) = pages.next().await {
67 let resp = resp?;
68 streams.extend(resp.log_streams().iter().map(|s| {
69 LogStream {
70 name: s.log_stream_name().unwrap_or("").to_string(),
71 creation_time: s
72 .creation_time()
73 .map(|t| DateTime::from_timestamp_millis(t).unwrap_or_default()),
74 last_event_time: s
75 .last_event_timestamp()
76 .map(|t| DateTime::from_timestamp_millis(t).unwrap_or_default()),
77 }
78 }));
79
80 if streams.len() >= 100 {
82 break;
83 }
84 }
85
86 Ok(streams)
87 }
88
89 pub async fn get_log_events(
90 &self,
91 log_group: &str,
92 log_stream: &str,
93 backward_token: Option<String>,
94 start_time: Option<i64>,
95 end_time: Option<i64>,
96 ) -> Result<(Vec<LogEvent>, bool, Option<String>)> {
97 let prev_token = backward_token.clone();
98 let mut request = self
99 .client
100 .get_log_events()
101 .log_group_name(log_group)
102 .log_stream_name(log_stream)
103 .set_start_from_head(Some(false))
104 .set_limit(Some(25));
105
106 if let Some(start) = start_time {
107 request = request.set_start_time(Some(start));
108 }
109
110 if let Some(end) = end_time {
111 request = request.set_end_time(Some(end));
112 }
113
114 if let Some(token) = backward_token {
115 request = request.set_next_token(Some(token));
116 }
117
118 let resp = request.send().await?;
119
120 let mut events: Vec<LogEvent> = resp
121 .events()
122 .iter()
123 .map(|e| LogEvent {
124 timestamp: DateTime::from_timestamp_millis(e.timestamp().unwrap_or(0))
125 .unwrap_or_default(),
126 message: e.message().unwrap_or("").to_string(),
127 ingestion_time: e.ingestion_time().and_then(DateTime::from_timestamp_millis),
128 })
129 .collect();
130
131 events.sort_by_key(|e| std::cmp::Reverse(e.timestamp));
132
133 let next_backward_token = resp.next_backward_token().map(|s| s.to_string());
134 let has_more = next_backward_token.is_some()
135 && next_backward_token != prev_token
136 && !events.is_empty();
137
138 Ok((events, has_more, next_backward_token))
139 }
140
141 pub async fn start_query(
142 &self,
143 log_group_names: Vec<String>,
144 query_string: String,
145 start_time: i64,
146 end_time: i64,
147 ) -> Result<String> {
148 let resp = self
149 .client
150 .start_query()
151 .set_log_group_names(Some(log_group_names))
152 .query_string(query_string)
153 .start_time(start_time / 1000)
154 .end_time(end_time / 1000)
155 .send()
156 .await?;
157
158 Ok(resp.query_id().unwrap_or("").to_string())
159 }
160
161 pub async fn get_query_results(
162 &self,
163 query_id: &str,
164 ) -> Result<(String, Vec<Vec<(String, String)>>)> {
165 let resp = self
166 .client
167 .get_query_results()
168 .query_id(query_id)
169 .send()
170 .await?;
171
172 let status = resp
173 .status()
174 .map(|s| s.as_str())
175 .unwrap_or("Unknown")
176 .to_string();
177
178 let results: Vec<Vec<(String, String)>> = resp
179 .results()
180 .iter()
181 .map(|result_row| {
182 result_row
183 .iter()
184 .map(|field| {
185 (
186 field.field().unwrap_or("").to_string(),
187 field.value().unwrap_or("").to_string(),
188 )
189 })
190 .collect()
191 })
192 .collect();
193
194 Ok((status, results))
195 }
196
197 pub async fn list_tags_for_log_group(
198 &self,
199 log_group_arn: &str,
200 ) -> Result<Vec<(String, String)>> {
201 let resp = self
202 .client
203 .list_tags_for_resource()
204 .resource_arn(log_group_arn)
205 .send()
206 .await?;
207
208 let tags = if let Some(tag_map) = resp.tags() {
209 tag_map
210 .iter()
211 .map(|(k, v)| (k.clone(), v.clone()))
212 .collect()
213 } else {
214 Vec::new()
215 };
216
217 Ok(tags)
218 }
219}
220
221fn log_group(group: &aws_sdk_cloudwatchlogs::types::LogGroup) -> LogGroup {
222 LogGroup {
223 name: group.log_group_name().unwrap_or_default().into(),
224 creation_time: group
225 .creation_time()
226 .map(|time| DateTime::from_timestamp_millis(time).unwrap_or_default()),
227 stored_bytes: group.stored_bytes(),
228 retention_days: group.retention_in_days(),
229 log_class: group.log_group_class().map(|class| class.as_str().into()),
230 arn: group.arn().map(str::to_owned),
231 log_group_arn: group.log_group_arn().map(str::to_owned),
232 deletion_protection_enabled: group.deletion_protection_enabled(),
233 }
234}