Skip to content

Commit 9d7c0df

Browse files
raphael-goetzraphael-goetz
committed
feat: enhanced logging
1 parent eca3def commit 9d7c0df

11 files changed

Lines changed: 280 additions & 135 deletions

‎src/sagittarius/flow_service_client_impl.rs‎

Lines changed: 86 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -71,12 +71,14 @@ impl SagittariusFlowClient {
7171
return;
7272
}
7373

74-
log::info!("Will export flows into file because env is set to `DEVELOPMENT`");
75-
7674
let json = match serde_json::to_vec_pretty(&flows) {
7775
Ok(b) => b,
7876
Err(e) => {
79-
log::error!("Failed to serialize flows to JSON: {:?}", e);
77+
log::error!(
78+
"Failed to serialize development flow export flow_count={} error={:?}",
79+
flows.flows.len(),
80+
e
81+
);
8082
return;
8183
}
8284
};
@@ -85,16 +87,48 @@ impl SagittariusFlowClient {
8587
let tmp_path = Path::new("flowExport.json.tmp");
8688

8789
if let Err(e) = fs::write(tmp_path, &json).await {
88-
log::error!("Failed to write {}: {}", tmp_path.display(), e);
90+
log::error!(
91+
"Failed to write development flow export path={} error={}",
92+
tmp_path.display(),
93+
e
94+
);
8995
return;
9096
}
9197

9298
if let Err(e) = fs::rename(tmp_path, final_path).await {
93-
log::warn!("rename failed (will try remove+rename): {}", e);
94-
let _ = fs::remove_file(final_path).await;
99+
log::warn!(
100+
"Could not atomically replace development flow export path={} error={}; retrying after removing destination",
101+
final_path.display(),
102+
e
103+
);
104+
match fs::remove_file(final_path).await {
105+
Ok(()) => log::debug!(
106+
"Removed previous development flow export path={}",
107+
final_path.display()
108+
),
109+
Err(remove_error) if remove_error.kind() == std::io::ErrorKind::NotFound => {}
110+
Err(remove_error) => log::warn!(
111+
"Failed to remove previous development flow export path={} error={}",
112+
final_path.display(),
113+
remove_error
114+
),
115+
}
95116
if let Err(e2) = fs::rename(tmp_path, final_path).await {
96-
log::error!("Failed to move export into place: {}", e2);
97-
let _ = fs::remove_file(tmp_path).await;
117+
log::error!(
118+
"Failed to replace development flow export source_path={} destination_path={} initial_rename_error={} retry_error={}",
119+
tmp_path.display(),
120+
final_path.display(),
121+
e,
122+
e2
123+
);
124+
if let Err(cleanup_error) = fs::remove_file(tmp_path).await {
125+
log::warn!(
126+
"Failed to clean up temporary development flow export path={} error={}",
127+
tmp_path.display(),
128+
cleanup_error
129+
);
130+
}
131+
return;
98132
}
99133
}
100134

@@ -107,24 +141,21 @@ impl SagittariusFlowClient {
107141

108142
async fn handle_response(&mut self, response: FlowResponse) {
109143
let data = match response.data {
110-
Some(data) => {
111-
log::info!("Received a FlowResponse");
112-
data
113-
}
144+
Some(data) => data,
114145
None => {
115-
log::error!("Received a empty FlowResponse");
146+
log::warn!("Received empty Sagittarius flow response");
116147
return;
117148
}
118149
};
119150

120151
match data {
121152
Data::DeletedFlowId(id) => {
122-
log::info!("Deleting the Flow with the id: {}", id);
153+
log::debug!("Applying flow deletion flow_id={}", id);
123154
let mut keys = match self.store.keys().await {
124155
Ok(keys) => keys.boxed(),
125156
Err(err) => {
126157
log::error!(
127-
"Failed to list flows while deleting flow {}. Reason: {:?}",
158+
"Failed to list stored flows for deletion flow_id={} error={:?}",
128159
id,
129160
err
130161
);
@@ -141,7 +172,7 @@ impl SagittariusFlowClient {
141172
match self.store.delete(&key).await {
142173
Ok(_) => deleted_count += 1,
143174
Err(err) => log::error!(
144-
"Failed to delete flow {} with key {}. Reason: {:?}",
175+
"Failed to delete stored flow flow_id={} key={} error={:?}",
145176
id,
146177
key,
147178
err
@@ -150,7 +181,7 @@ impl SagittariusFlowClient {
150181
}
151182

152183
if deleted_count == 0 {
153-
log::warn!("No stored flow found with the id: {}", id);
184+
log::warn!("Flow deletion matched no stored keys flow_id={}", id);
154185
} else {
155186
log::info!(
156187
"Flow deleted successfully id={} deleted_keys={}",
@@ -160,23 +191,35 @@ impl SagittariusFlowClient {
160191
}
161192
}
162193
Data::UpdatedFlow(flow) => {
163-
log::info!("Updating the Flow with the id: {}", &flow.flow_id);
164194
let key = get_flow_identifier(&flow);
195+
let flow_id = flow.flow_id.clone();
165196
let bytes = flow.encode_to_vec();
166-
match self.store.put(key, bytes.into()).await {
167-
Ok(_) => log::info!("Flow updated successfully"),
168-
Err(err) => log::error!("Failed to update flow. Reason: {:?}", err),
197+
match self.store.put(key.clone(), bytes.into()).await {
198+
Ok(_) => log::info!("Stored flow update flow_id={} key={}", flow_id, key),
199+
Err(err) => log::error!(
200+
"Failed to store flow update flow_id={} key={} error={:?}",
201+
flow_id,
202+
key,
203+
err
204+
),
169205
};
170206
}
171207
Data::Flows(flows) => {
172-
log::info!("Dropping all Flows & inserting the new ones!");
208+
let received_count = flows.flows.len();
209+
log::info!(
210+
"Replacing stored flows from Sagittarius received_count={}",
211+
received_count
212+
);
173213

174214
self.export_flows_json_overwrite(flows.clone()).await;
175215

176216
let mut keys = match self.store.keys().await {
177217
Ok(keys) => keys.boxed(),
178218
Err(err) => {
179-
log::error!("Service wasn't able to get keys. Reason: {:?}", err);
219+
log::error!(
220+
"Failed to list stored flows before replacement error={:?}",
221+
err
222+
);
180223
return;
181224
}
182225
};
@@ -185,21 +228,34 @@ impl SagittariusFlowClient {
185228
while let Ok(Some(key)) = keys.try_next().await {
186229
match self.store.purge(&key).await {
187230
Ok(_) => purged_count += 1,
188-
Err(e) => log::error!("Failed to purge key {}: {}", key, e),
231+
Err(e) => {
232+
log::error!("Failed to purge stored flow key={} error={}", key, e)
233+
}
189234
}
190235
}
191236

192-
log::info!("Purged {} existing keys", purged_count);
193-
237+
let mut stored_count = 0;
194238
for flow in flows.flows {
195239
let key = get_flow_identifier(&flow);
196-
log::debug!("trying to insert: {}", key);
197240
let bytes = flow.encode_to_vec();
198-
match self.store.put(key, bytes.into()).await {
199-
Ok(_) => log::info!("Flow updated successfully"),
200-
Err(err) => log::error!("Failed to update flow. Reason: {:?}", err),
241+
match self.store.put(key.clone(), bytes.into()).await {
242+
Ok(_) => {
243+
stored_count += 1;
244+
log::debug!("Stored replacement flow key={}", key);
245+
}
246+
Err(err) => log::error!(
247+
"Failed to store replacement flow key={} error={:?}",
248+
key,
249+
err
250+
),
201251
};
202252
}
253+
log::info!(
254+
"Finished replacing stored flows received_count={} purged_count={} stored_count={}",
255+
received_count,
256+
purged_count,
257+
stored_count
258+
);
203259
}
204260
Data::ModuleConfigurations(action_configurations) => {
205261
let (project_count, config_count) = module_config_stats(&action_configurations);

‎src/sagittarius/module_service_client_impl.rs‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -60,8 +60,11 @@ impl SagittariusModuleServiceClient {
6060
}
6161
Err(err) => {
6262
log::error!(
63-
"Failed to update Modules via Sagittarius RPC transport: {:?}",
64-
err
63+
"Sagittarius module update RPC failed module_count={} code={} message={} timeout_ms={}",
64+
module_count,
65+
err.code(),
66+
err.message(),
67+
self.unary_rpc_timeout.as_millis()
6568
);
6669
tucana::aquila::ModuleUpdateResponse { success: false }
6770
}

‎src/sagittarius/retry.rs‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,11 @@ pub async fn create_channel_with_retry(
3030

3131
let channel = match Endpoint::from_shared(url.clone()) {
3232
Ok(c) => {
33-
log::debug!("Creating a new endpoint for the: {} Service", channel_name);
33+
log::debug!(
34+
"Created Sagittarius endpoint channel={} url={}",
35+
channel_name,
36+
url
37+
);
3438
c.connect_timeout(Duration::from_secs(2))
3539
}
3640
Err(err) => {
@@ -68,10 +72,11 @@ pub async fn create_channel_with_retry(
6872

6973
if retries >= MAX_RETRIES {
7074
log::error!(
71-
"Reached max retries channel={} url={} max_retries={}",
75+
"Sagittarius connection retries exhausted channel={} url={} attempts={} last_error={:?}",
7276
channel_name,
7377
url,
74-
MAX_RETRIES
78+
MAX_RETRIES,
79+
err
7580
);
7681
panic!("Reached max retries to url {}", url)
7782
}

‎src/sagittarius/runtime_status_service_client_impl.rs‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,12 @@ impl SagittariusRuntimeStatusServiceClient {
3939
response.into_inner()
4040
}
4141
Err(err) => {
42-
log::error!("Failed to update RuntimeStatus: {:?}", err);
42+
log::error!(
43+
"Sagittarius runtime status update RPC failed code={} message={} timeout_ms={}",
44+
err.code(),
45+
err.message(),
46+
self.unary_rpc_timeout.as_millis()
47+
);
4348
return tucana::aquila::RuntimeStatusUpdateResponse { success: false };
4449
}
4550
};

‎src/sagittarius/test_execution_client_impl.rs‎

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -289,16 +289,24 @@ impl SagittariusTestExecutionServiceClient {
289289
Some(flow)
290290
}
291291
Err(err) => {
292-
log::error!("Cannot decode ValidationFlow for {}: {:?}", flow_id, err);
292+
log::error!(
293+
"Failed to decode validation flow flow_id={} error={:?}",
294+
flow_id,
295+
err
296+
);
293297
None
294298
}
295299
},
296300
Ok(None) => {
297-
log::error!("No flow found with id: {}", flow_id);
301+
log::error!("Validation flow was not found flow_id={}", flow_id);
298302
None
299303
}
300304
Err(err) => {
301-
log::error!("Error fetching flow {}: {:?}", flow_id, err);
305+
log::error!(
306+
"Failed to fetch validation flow flow_id={} error={:?}",
307+
flow_id,
308+
err
309+
);
302310
None
303311
}
304312
}
@@ -314,7 +322,10 @@ impl SagittariusTestExecutionServiceClient {
314322

315323
log::debug!("Queueing Sagittarius execution stream logon before opening stream");
316324
if let Err(err) = tx.send(logon).await {
317-
log::error!("Failed to queue test execution logon: {:?}", err);
325+
log::error!(
326+
"Failed to queue Sagittarius execution stream logon reason=channel_closed error={:?}",
327+
err
328+
);
318329
self.response_sender.clear().await;
319330
return;
320331
}
@@ -335,8 +346,9 @@ impl SagittariusTestExecutionServiceClient {
335346
}
336347
Err(error) => {
337348
log::error!(
338-
"Failed to establish Sagittarius execution stream: {:?}",
339-
error
349+
"Failed to establish Sagittarius execution stream code={} message={}",
350+
error.code(),
351+
error.message()
340352
);
341353
self.response_sender.clear().await;
342354
return;

0 commit comments

Comments
 (0)