Fixed History Logger

Performance Optimization - History Logging was spawning a new tokio runtime every function call, it now uses the global run time
Documentation - Added a lot more span events for better logging

airlock_libs/Cargo.lock

@@ -26,7 +26,7 @@ dependencies = [
 
 [[package]]
 name = "airlock_libs"
-version = "7.1.0"
+version = "7.4.0"
 dependencies = [
  "chrono",
  "crossbeam",

airlock_libs/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "airlock_libs"
-version = "7.1.0"
+version = "7.4.0"
 edition = "2024"
 
 [dependencies]

airlock_libs/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "airlock_libs"
-version = "7.1.0"
+version = "7.4.0"
 description = "Airlock Digital API Wrapper"
 readme = "README.md"
 license = { text = "AGPL-3.0-only" }

airlock_libs/src/services.rs

@@ -1,5 +1,6 @@
 use crate::modules::datatypes::*;
 use crate::prelude::*;
+use opentelemetry::trace::SpanContext;
 
 #[pyfunction]
 pub fn pull_policy_exec_histories(
@@ -106,9 +107,10 @@ pub fn pull_policy_exec_histories(
         });
         let cutoff: chrono::NaiveDateTime =
             Local::now().naive_local() - chrono::Duration::days(days);
-        let (tx, rx) = unbounded::<Vec<Group>>();
+        let (tx, rx) = unbounded::<(SpanContext, Vec<Group>)>();
         let pb_clone = progress_bar.clone();
         thread::spawn(move || {
+            let tracer = global::tracer("loxide");
             let mut seen: HashMap<(String, String, String), Group> = if writeable_filepath.exists()
             {
                 let contents: String = fs::read_to_string(&writeable_filepath).unwrap_or_default();
@@ -137,7 +139,22 @@ pub fn pull_policy_exec_histories(
             } else {
                 HashMap::new()
             };
-            while let Ok(parsed_responses) = rx.recv() {
+            while let Ok((parent_spancontext, parsed_responses)) = rx.recv() {
+                let parent_ctx = Context::new().with_remote_span_context(parent_spancontext);
+                let span = tracer.build_with_context(
+                    tracer
+                        .span_builder("Deduplicate and Write")
+                        .with_kind(trace::SpanKind::Consumer),
+                    &parent_ctx,
+                );
+                let cx = Context::current_with_span(span);
+                cx.span().add_event(
+                    "Received Data from Producer",
+                    vec![KeyValue::new(
+                        "Items to Process",
+                        parsed_responses.len().to_string(),
+                    )],
+                );
                 for executions in parsed_responses {
                     if executions.checkpoint.is_empty() || executions.datetime.is_empty() {
                         continue;
@@ -166,11 +183,28 @@ pub fn pull_policy_exec_histories(
                 };
                 let data_write: String = serde_json::to_string_pretty(&final_response).unwrap();
                 match fs::write(&writeable_filepath, data_write) {
-                    Ok(_) => {}
+                    Ok(_) => {
+                        cx.span().add_event(
+                            "Writing Data to File",
+                            vec![KeyValue::new("Success", "Ok".to_string())],
+                        );
+                    }
                     Err(e) => {
-                        println!("Failed to write to: {:?}: {}", &writeable_filepath, e);
+                        cx.span().add_event(
+                            "Writing Data to File",
+                            vec![KeyValue::new("Failed", e.to_string())],
+                        );
+                        cx.span()
+                            .set_status(Status::error("Failed to Write to File"));
                     }
                 }
+                cx.span().add_event(
+                    "Finished Deduplicating Data",
+                    vec![KeyValue::new(
+                        "Items Successfully Processed",
+                        seen.len().to_string(),
+                    )],
+                );
             }
         });
         let mut first_date: Option<NaiveDate> = None;
@@ -193,13 +227,28 @@ pub fn pull_policy_exec_histories(
             ));
             loop {
                 let execution_histories = tracer.in_span(checkpoint_number.to_string(), |cx| {
-                    let results: ApiResponse = history_logging(
+                    cx.span().add_event(
+                        "Retrieving Responses from API",
+                        vec![KeyValue::new(
+                            "Checkpoint Number",
+                            checkpoint_number.to_string(),
+                        )],
+                    );
+                    let results: ApiResponse = rt.block_on(history_logging(
                         &base_url,
                         &exec_types,
                         &checkpoint_number,
                         &policy_names,
                         &client,
+                    ));
+                    cx.span().add_event(
+                        "Got Responses from API",
+                        vec![KeyValue::new(
+                            "Items in Response",
+                            results.response.exechistories.len().to_string(),
+                        )],
                     );
+                    cx.span().set_status(Status::Ok);
                     cx.span().set_attribute(KeyValue::new(
                         "items_in_response",
                         results.response.exechistories.len().to_string(),
@@ -210,7 +259,16 @@ pub fn pull_policy_exec_histories(
                 if parsed_responses.is_empty() {
                     break;
                 }
-                tx.send(parsed_responses.clone()).unwrap();
+                match tx.send((cx.span().span_context().clone(), parsed_responses.clone())) {
+                    Ok(_) => {}
+                    Err(e) => {
+                        cx.span().add_event(
+                            "Failed to Send Items to Processor",
+                            vec![KeyValue::new("Response from Processor", e.to_string())],
+                        );
+                        cx.span().set_status(Status::error("Processor Failed"))
+                    }
+                }
                 checkpoint_number = parsed_responses.last().unwrap().checkpoint.clone();
                 if let Some(last_item) = parsed_responses.last()
                     && let Ok(last_date) = NaiveDate::parse_from_str(
@@ -263,7 +321,7 @@ fn build_client(headers: HeaderMap) -> Result<reqwest::Client, reqwest::Error> {
         .build()
 }
 
-#[tokio::main]
+#[tracing::instrument(name = "history_logging")]
 async fn history_logging(
     base_url: &String,
     exec_types: &String,
@@ -292,7 +350,7 @@ async fn history_logging(
         Ok(res) => {
             let first_response: ApiResponse = serde_json::from_str(&res.text().await.unwrap())
                 .expect("Failed to retrieve response from API");
-            return first_response;
+            first_response
         }
         Err(_res) => {
             let failed_response: ApiResponse = ApiResponse {
@@ -301,7 +359,7 @@ async fn history_logging(
                     exechistories: vec![],
                 },
             };
-            return failed_response;
+            failed_response
         }
     }
 }

requirements.txt

@@ -23,4 +23,4 @@ pyperclip==1.11.0
 
 # Custom/Private packages
 --extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/
-airlock_libs==7.1.0
+airlock_libs==7.4.0