a way to store arbitrarily complex data with tracing

Author: Byron

Cargo.lock

@@ -23,7 +23,7 @@ estimate-hours = ["dep:itertools", "dep:fs-err", "dep:crossbeam-channel", "dep:s
 query = ["dep:rusqlite"]
 ## Run algorithms on a corpus of repositories and store their results for later comparison and intelligence gathering.
 ## *Note that* `organize` we need for finding git repositories fast.
-corpus = ["dep:rusqlite", "dep:sysinfo", "organize", "dep:crossbeam-channel"]
+corpus = ["dep:rusqlite", "dep:sysinfo", "organize", "dep:crossbeam-channel", "dep:tracing-forest", "dep:serde_json", "dep:tracing-subscriber", "dep:tracing"]
 
 #! ### Mutually Exclusive Networking
 #! If both are set, _blocking-client_ will take precedence, allowing `--all-features` to be used.
@@ -36,7 +36,7 @@ async-client = ["gix/async-network-client-async-std", "gix-transport-configurati
 
 #! ### Other
 ## Data structures implement `serde::Serialize` and `serde::Deserialize`.
-serde = ["gix/serde", "serde_json", "dep:serde", "bytesize/serde"]
+serde = ["gix/serde", "dep:serde_json", "dep:serde", "bytesize/serde"]
 
 
 [dependencies]
@@ -48,7 +48,6 @@ serde = { version = "1.0.114", optional = true, default-features = false, featur
 anyhow = "1.0.42"
 thiserror = "1.0.34"
 bytesize = "1.0.1"
-serde_json = { version = "1.0.65", optional = true }
 tempfile = "3.1.0"
 
 # for async-client
@@ -74,6 +73,10 @@ rusqlite = { version = "0.29.0", optional = true, features = ["bundled"] }
 
 # for 'corpus'
 sysinfo = { version = "0.29.2", optional = true, default-features = false }
+tracing-forest = { version = "0.1.5", features = ["serde"], optional = true }
+serde_json = { version = "1.0.65", optional = true }
+tracing-subscriber = { version = "0.3.17", optional = true }
+tracing = { version = "0.1.37", optional = true }
 
 # for svg graph output
 layout-rs = "0.1.1"

gitoxide-core/Cargo.toml

@@ -131,6 +131,7 @@ pub fn create(path: impl AsRef<std::path::Path>) -> anyhow::Result<rusqlite::Con
         insertion_time integer NOT NULL, -- in seconds since UNIX epoch
         duration real, -- in seconds or NULL if not yet finished (either successfull or with failure)
         error text, -- or NULL if there was no error
+        spans_json text, -- all spans collecteted while performing the run
         FOREIGN KEY (repository) REFERENCES repository (id),
         FOREIGN KEY (task) REFERENCES task (id),
         FOREIGN KEY (runner) REFERENCES runner (id),

gitoxide-core/src/corpus/db.rs

@@ -58,13 +58,16 @@ impl Engine {
         task_progress.set_name("run");
         task_progress.init(Some(tasks.len()), gix::progress::count("tasks"));
         let threads = gix::parallel::num_threads(threads);
+        let db_path = self.con.path().expect("opened from path on disk").to_owned();
         for (task_id, task) in tasks {
             let task_start = Instant::now();
             let mut repo_progress = task_progress.add_child(format!("run '{}'", task.short_name));
             repo_progress.init(Some(repos.len()), gix::progress::count("repos"));
 
             if task.execute_exclusive || threads == 1 {
                 let mut run_progress = repo_progress.add_child("set later");
+                let (_guard, current_id) = corpus::trace::override_thread_subscriber(db_path.as_str())?;
+
                 for repo in &repos {
                     if gix::interrupt::is_triggered() {
                         bail!("interrupted by user");
@@ -76,14 +79,19 @@ impl Engine {
                             .expect("corpus contains repo")
                             .display()
                     ));
+
+                    // TODO: wait for new release to be able to provide run_id via span attributes
                     let mut run = Self::insert_run(&self.con, gitoxide_id, runner_id, *task_id, repo.id)?;
-                    task.perform(
-                        &mut run,
-                        &repo.path,
-                        &mut run_progress,
-                        Some(threads),
-                        &gix::interrupt::IS_INTERRUPTED,
-                    );
+                    current_id.store(run.id, Ordering::SeqCst);
+                    tracing::info_span!("run", run_id = run.id).in_scope(|| {
+                        task.perform(
+                            &mut run,
+                            &repo.path,
+                            &mut run_progress,
+                            Some(threads),
+                            &gix::interrupt::IS_INTERRUPTED,
+                        );
+                    });
                     Self::update_run(&self.con, run)?;
                     repo_progress.inc();
                 }
@@ -96,22 +104,31 @@ impl Engine {
                     Some(threads),
                     {
                         let shared_repo_progress = repo_progress.clone();
-                        let path = self.con.path().expect("opened from path on disk").to_owned();
+                        let db_path = db_path.clone();
                         move |tid| {
                             (
+                                corpus::trace::override_thread_subscriber(db_path.as_str()),
                                 gix::threading::lock(&shared_repo_progress).add_child(format!("{tid}")),
-                                rusqlite::Connection::open(&path),
+                                rusqlite::Connection::open(&db_path),
                             )
                         }
                     },
-                    |repo, (progress, con), _threads_left, should_interrupt| -> anyhow::Result<()> {
+                    |repo, (subscriber, progress, con), _threads_left, should_interrupt| -> anyhow::Result<()> {
                         progress.set_name(format!(
                             "{}",
                             repo.path
                                 .strip_prefix(corpus_path)
                                 .expect("corpus contains repo")
                                 .display()
                         ));
+                        let current_id = match subscriber {
+                            Ok((_guard, current_id)) => current_id,
+                            Err(err) => {
+                                progress.fail(format!("{err:#?}"));
+                                should_interrupt.store(true, Ordering::SeqCst);
+                                return Ok(());
+                            }
+                        };
                         let con = match con {
                             Ok(con) => con,
                             Err(err) => {
@@ -121,7 +138,10 @@ impl Engine {
                             }
                         };
                         let mut run = Self::insert_run(con, gitoxide_id, runner_id, *task_id, repo.id)?;
-                        task.perform(&mut run, &repo.path, progress, Some(1), should_interrupt);
+                        current_id.store(run.id, Ordering::SeqCst);
+                        tracing::info_span!("run", run_id = run.id).in_scope(|| {
+                            task.perform(&mut run, &repo.path, progress, Some(1), should_interrupt);
+                        });
                         Self::update_run(con, run)?;
                         if let Some(counter) = counter.as_ref() {
                             counter.fetch_add(1, Ordering::SeqCst);

gitoxide-core/src/corpus/engine.rs

@@ -38,6 +38,46 @@ pub(crate) struct Run {
     error: Option<String>,
 }
 
+pub(crate) mod trace {
+    use rusqlite::params;
+    use std::path::Path;
+    use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::{Arc, Mutex};
+    use tracing_forest::tree::Tree;
+    use tracing_subscriber::layer::SubscriberExt;
+
+    pub fn override_thread_subscriber(
+        db_path: impl AsRef<Path>,
+    ) -> anyhow::Result<(tracing::subscriber::DefaultGuard, Arc<AtomicU32>)> {
+        let current_id = Arc::new(AtomicU32::default());
+        let processor = tracing_forest::Printer::new().formatter(StoreTreeToDb {
+            con: Arc::new(Mutex::new(rusqlite::Connection::open(&db_path)?)),
+            run_id: current_id.clone(),
+        });
+        let subscriber = tracing_subscriber::Registry::default().with(tracing_forest::ForestLayer::from(processor));
+        let guard = tracing::subscriber::set_default(subscriber);
+        Ok((guard, current_id))
+    }
+
+    pub struct StoreTreeToDb {
+        pub con: Arc<Mutex<rusqlite::Connection>>,
+        pub run_id: Arc<AtomicU32>,
+    }
+    impl tracing_forest::printer::Formatter for StoreTreeToDb {
+        type Error = rusqlite::Error;
+
+        fn fmt(&self, tree: &Tree) -> Result<String, Self::Error> {
+            let json = serde_json::to_string_pretty(&tree).expect("serialization to string always works");
+            let run_id = self.run_id.load(Ordering::SeqCst);
+            self.con
+                .lock()
+                .unwrap()
+                .execute("UPDATE run SET spans_json = ?1 WHERE id = ?2", params![json, run_id])?;
+            Ok(String::new())
+        }
+    }
+}
+
 pub(crate) mod run {
     use crate::corpus;
     use crate::corpus::{Run, Task};

gitoxide-core/src/corpus/mod.rs

@@ -166,7 +166,7 @@ pub fn statistics(
                 cancelled,
             ),
             thread_limit,
-            move |_| (repo.objects.clone().into_inner(), counter.clone()),
+            move |_| (repo.objects.clone().into_inner(), counter),
             |ids, (handle, counter)| {
                 let ids = ids?;
                 if let Some(counter) = counter {