diff --git a/src/clip_watch.rs b/src/clip_watch.rs index bd38833..d0b1802 100644 --- a/src/clip_watch.rs +++ b/src/clip_watch.rs @@ -29,6 +29,7 @@ use log::{debug, info, warn}; use std::path::Path; use std::sync::{Arc, Mutex}; use tokio::sync::Semaphore; +use tracing::Instrument; /// One file the watcher would like to CLIP-encode. Built from the DAO /// `list_clip_unencoded_candidates` result — needs the `content_hash` @@ -90,6 +91,9 @@ pub fn run_clip_encoding_pass( let library_id = library.id; let library_root = library.root_path.clone(); + // The detect tasks run on this runtime's worker threads; carry the + // watcher's span into each so their spans stay in the tick's trace. + let parent = tracing::Span::current(); rt.block_on(async move { let sem = Arc::new(Semaphore::new(concurrency)); let mut handles = Vec::with_capacity(filtered.len()); @@ -98,10 +102,14 @@ pub fn run_clip_encoding_pass( let clip_client = clip_client.clone(); let exif_dao = exif_dao.clone(); let library_root = library_root.clone(); - handles.push(tokio::spawn(async move { - let _permit = permit_sem.acquire().await.expect("clip semaphore"); - process_one(library_id, &library_root, cand, &clip_client, exif_dao).await; - })); + let task_span = parent.clone(); + handles.push(tokio::spawn( + async move { + let _permit = permit_sem.acquire().await.expect("clip semaphore"); + process_one(library_id, &library_root, cand, &clip_client, exif_dao).await; + } + .instrument(task_span), + )); } for h in handles { let _ = h.await; @@ -109,6 +117,7 @@ pub fn run_clip_encoding_pass( }); } +#[tracing::instrument(name = "clip_watch.encode", skip_all, fields(rel_path = %cand.rel_path))] async fn process_one( library_id: i32, library_root: &str, diff --git a/src/face_watch.rs b/src/face_watch.rs index 8dabc99..9c7b11e 100644 --- a/src/face_watch.rs +++ b/src/face_watch.rs @@ -25,6 +25,7 @@ use log::{debug, info, warn}; use std::path::Path; use std::sync::{Arc, Mutex}; use tokio::sync::Semaphore; +use tracing::Instrument; /// One file the watcher would like to face-scan. Built by the caller from /// the EXIF batch (we need `content_hash` to key everything against). @@ -89,6 +90,9 @@ pub fn run_face_detection_pass( let library_id = library.id; let library_root = library.root_path.clone(); + // The detect tasks run on this runtime's worker threads; carry the + // watcher's span into each so their spans stay in the tick's trace. + let parent = tracing::Span::current(); rt.block_on(async move { let sem = Arc::new(Semaphore::new(concurrency)); let mut handles = Vec::with_capacity(filtered.len()); @@ -98,21 +102,25 @@ pub fn run_face_detection_pass( let face_dao = face_dao.clone(); let tag_dao = tag_dao.clone(); let library_root = library_root.clone(); - handles.push(tokio::spawn(async move { - // acquire_owned would let us drop the permit explicitly - // before await points; for a one-shot call into Apollo - // the simpler bounded acquire is enough. - let _permit = permit_sem.acquire().await.expect("face semaphore"); - process_one( - library_id, - &library_root, - cand, - &face_client, - face_dao, - tag_dao, - ) - .await; - })); + let task_span = parent.clone(); + handles.push(tokio::spawn( + async move { + // acquire_owned would let us drop the permit explicitly + // before await points; for a one-shot call into Apollo + // the simpler bounded acquire is enough. + let _permit = permit_sem.acquire().await.expect("face semaphore"); + process_one( + library_id, + &library_root, + cand, + &face_client, + face_dao, + tag_dao, + ) + .await; + } + .instrument(task_span), + )); } for h in handles { // join; per-task panics are logged inside process_one before @@ -122,6 +130,7 @@ pub fn run_face_detection_pass( }); } +#[tracing::instrument(name = "face_watch.detect", skip_all, fields(rel_path = %cand.rel_path))] async fn process_one( library_id: i32, library_root: &str, diff --git a/src/files.rs b/src/files.rs index 70b4067..5b375cf 100644 --- a/src/files.rs +++ b/src/files.rs @@ -30,11 +30,10 @@ use actix_web::{ use chrono::{DateTime, Utc}; use log::{debug, error, info, trace, warn}; use opentelemetry::KeyValue; -use opentelemetry::trace::{Status, Tracer}; +use opentelemetry::trace::Status; use crate::data::SortType::NameAsc; use crate::error::IntoHttpError; -use crate::otel::global_tracer; use crate::tags::{FileWithTagCount, TagDao}; use crate::video::actors::StreamActor; use path_absolutize::*; @@ -1373,9 +1372,8 @@ impl Message for RefreshThumbnailsMessage { impl Handler for StreamActor { type Result = (); + #[tracing::instrument(name = "RefreshThumbnailsMessage", skip_all)] fn handle(&mut self, _msg: RefreshThumbnailsMessage, _ctx: &mut Self::Context) -> Self::Result { - let tracer = global_tracer(); - let _ = tracer.start("RefreshThumbnailsMessage"); info!("Refreshing thumbnails after upload"); // The stub in lib.rs is a no-op; the real generation is driven by // the file watcher tick in main.rs, which has access to the diff --git a/src/otel.rs b/src/otel.rs index c511571..ee9235e 100644 --- a/src/otel.rs +++ b/src/otel.rs @@ -1,4 +1,3 @@ -use opentelemetry::global::BoxedTracer; use opentelemetry::propagation::TextMapPropagator; use opentelemetry::trace::{Status, TraceContextExt, TracerProvider}; use opentelemetry::{Context, KeyValue, global}; @@ -10,10 +9,6 @@ use opentelemetry_sdk::propagation::TraceContextPropagator; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; -pub fn global_tracer() -> BoxedTracer { - global::tracer("image-server") -} - /// Install the process-wide `tracing` subscriber plus the OTel span pipeline. /// /// Called in BOTH debug and release builds (the old split made spans a no-op @@ -54,9 +49,9 @@ pub fn init_tracing() { } let tracer_provider = provider_builder.build(); - // The rest of the codebase resolves spans through the global provider - // (`global_tracer()`, `global::tracer("db")`), so install it there; the - // subscriber's OTel layer uses the same provider's tracer. + // Install the provider globally too (flush/shutdown and any direct + // `global::tracer` users); the subscriber's OTel layer uses the same + // provider's tracer. let tracer = tracer_provider.tracer("image-server"); global::set_tracer_provider(tracer_provider); diff --git a/src/reels/mod.rs b/src/reels/mod.rs index 1cffbb3..b6871ae 100644 --- a/src/reels/mod.rs +++ b/src/reels/mod.rs @@ -28,6 +28,7 @@ use anyhow::{Context, anyhow}; use chrono::{DateTime, Datelike, Timelike}; use serde::{Deserialize, Serialize}; use serde_json::json; +use tracing::Instrument; use uuid::Uuid; use crate::data::Claims; @@ -557,34 +558,41 @@ pub async fn create_reel_handler( let state = app_state.clone(); let insight_dao = insight_dao.clone(); let exif_dao = exif_dao.clone(); - let handle = tokio::spawn(async move { - match run_reel_job( - &state, - &insight_dao, - &exif_dao, - job_id, - planned, - meta, - voice, - &key, - ) - .await - { - Ok((title, path)) => { - finish_job(job_id, ReelJobStatus::Done, Some(title), Some(path), None) - } - Err(e) => { - log::error!("reel job {job_id} failed: {e:?}"); - finish_job( - job_id, - ReelJobStatus::Error, - None, - None, - Some(format!("{e}")), - ) + // Created before the spawn so the job parents on the dispatching request + // (or runs as its own root from the pre-gen scheduler), then carried into + // the task — tracing context doesn't cross `tokio::spawn` on its own. + let job_span = tracing::info_span!("reel.job", job_id = %job_id); + let handle = tokio::spawn( + async move { + match run_reel_job( + &state, + &insight_dao, + &exif_dao, + job_id, + planned, + meta, + voice, + &key, + ) + .await + { + Ok((title, path)) => { + finish_job(job_id, ReelJobStatus::Done, Some(title), Some(path), None) + } + Err(e) => { + log::error!("reel job {job_id} failed: {e:?}"); + finish_job( + job_id, + ReelJobStatus::Error, + None, + None, + Some(format!("{e}")), + ) + } } } - }); + .instrument(job_span), + ); with_job(job_id, |job| job.abort = Some(handle.abort_handle())); // Capture params for passive prefs mirror (best-effort, never fails). @@ -1056,7 +1064,11 @@ pub(crate) async fn spawn_pregen_scheduler(app_state: web::Data) { log::debug!("Next pre-gen run in {}s", sleep_secs); tokio::time::sleep(std::time::Duration::from_secs(sleep_secs)).await; - if let Err(e) = run_pregen_batch(&app_state).await { + // One root span per nightly batch. + if let Err(e) = run_pregen_batch(&app_state) + .instrument(tracing::info_span!("reel.pregen_batch")) + .await + { log::error!("Reel pre-generation batch failed: {}", e); } } @@ -1267,7 +1279,8 @@ pub(crate) async fn spawn_reel_cache_sweeper(app_state: web::Data) { // Settle after startup, then sweep on a fixed cadence. tokio::time::sleep(Duration::from_secs(300)).await; loop { - let removed = sweep_reel_cache(&app_state); + let removed = + tracing::info_span!("reel.cache_sweep").in_scope(|| sweep_reel_cache(&app_state)); if removed > 0 { log::info!("Reel cache sweep removed {removed} stale file(s)"); } diff --git a/src/thumbnails.rs b/src/thumbnails.rs index c5ade67..11e4f5c 100644 --- a/src/thumbnails.rs +++ b/src/thumbnails.rs @@ -14,19 +14,15 @@ use image::GenericImageView; use image::codecs::jpeg::JpegEncoder; use lazy_static::lazy_static; use log::{debug, error, info, warn}; -use opentelemetry::{ - KeyValue, - trace::{Span, TraceContextExt, Tracer}, -}; use prometheus::IntGauge; use rayon::prelude::*; +use tracing_opentelemetry::OpenTelemetrySpanExt; use walkdir::DirEntry; use crate::content_hash; use crate::exif; use crate::file_types; use crate::libraries; -use crate::otel::global_tracer; use crate::video::actors::{generate_image_thumbnail_ffmpeg, generate_video_thumbnail}; /// Maximum long-edge size (px) for the large preview tier. Tuned to look @@ -327,9 +323,9 @@ fn generate_xlarge_preview_ffmpeg( Ok(()) } +#[tracing::instrument(name = "creating thumbnails", skip_all)] pub fn create_thumbnails(libs: &[libraries::Library], excluded_dirs: &[String]) { - let tracer = global_tracer(); - let span = tracer.start("creating thumbnails"); + let span = tracing::Span::current(); let thumbs = &dotenv::var("THUMBNAILS").expect("THUMBNAILS not defined"); let thumbnail_directory: &Path = Path::new(thumbs); @@ -387,16 +383,12 @@ pub fn create_thumbnails(libs: &[libraries::Library], excluded_dirs: &[String]) } if is_video(&entry) { - let mut video_span = tracer.start_with_context( - "generate_video_thumbnail", - &opentelemetry::Context::new() - .with_remote_span_context(span.span_context().clone()), - ); - video_span.set_attributes(vec![ - KeyValue::new("type", "video"), - KeyValue::new("file-name", thumb_path.display().to_string()), - KeyValue::new("library", lib.name.clone()), - ]); + // Explicit parent: this closure runs on rayon worker + // threads, where the ambient span isn't current. + let video_span = tracing::info_span!(parent: &span, "generate_video_thumbnail"); + video_span.set_attribute("type", "video"); + video_span.set_attribute("file-name", thumb_path.display().to_string()); + video_span.set_attribute("library", lib.name.clone()); debug!("Generating video thumbnail: {:?}", thumb_path); if let Err(e) = generate_video_thumbnail(src, &thumb_path) { @@ -409,7 +401,7 @@ pub fn create_thumbnails(libs: &[libraries::Library], excluded_dirs: &[String]) warn!("Failed to write sentinel {:?}: {}", sentinel, se); } } - video_span.end(); + drop(video_span); } else if is_image(&entry) { match generate_image_thumbnail(src, &thumb_path) { Ok(_) => info!("Saved thumbnail: {:?}", thumb_path), diff --git a/src/video/actors.rs b/src/video/actors.rs index b5feb1c..6f1520a 100644 --- a/src/video/actors.rs +++ b/src/video/actors.rs @@ -1,18 +1,19 @@ use crate::content_hash; use crate::database::PreviewDao; use crate::libraries::Library; -use crate::otel::global_tracer; use crate::video::ffmpeg::{generate_preview_clip, get_duration_seconds_blocking}; use crate::video::hls_paths; use actix::prelude::*; use log::{debug, error, info, warn}; use opentelemetry::KeyValue; -use opentelemetry::trace::{Span, Status, Tracer}; +use opentelemetry::trace::Status; use std::io::Result; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::sync::{Arc, Mutex}; use tokio::sync::Semaphore; +use tracing::Instrument; +use tracing_opentelemetry::OpenTelemetrySpanExt; // ffmpeg -i test.mp4 -c:v h264 -flags +cgop -g 30 -hls_time 3 out.m3u8 // ffmpeg -i "filename.mp4" -preset veryfast -c:v libx264 -f hls -hls_list_size 100 -hls_time 2 -crf 24 -vf scale=1080:-2,setsar=1:1 attempt/vid_out.m3u8 @@ -448,15 +449,16 @@ impl Handler for PlaylistGenerator { let playlist_file = playlist_path.to_string_lossy().to_string(); let segment_pattern = segment_template.to_string_lossy().to_string(); - let tracer = global_tracer(); - let mut span = tracer - .span_builder("playlistgenerator.generate_playlist") - .with_attributes(vec![ - KeyValue::new("video_file", video_file.clone()), - KeyValue::new("content_hash", content_hash_str.clone()), - KeyValue::new("playlist_file", playlist_file.clone()), - ]) - .start(&tracer); + // Job root: actix messages don't carry the sender's span, so each + // playlist generation is its own trace. The future is instrumented + // so any DB / child spans inside nest under it. + let span = tracing::info_span!( + "playlistgenerator.generate_playlist", + video_file = %video_file, + content_hash = %content_hash_str, + playlist_file = %playlist_file, + ); + let fut_span = span.clone(); Box::pin(async move { let wait_start = std::time::Instant::now(); @@ -691,7 +693,8 @@ impl Handler for PlaylistGenerator { span.set_status(Status::error(detail.clone())); Err(std::io::Error::other(detail)) } - }) + } + .instrument(fut_span)) } } @@ -818,39 +821,46 @@ impl Handler for PreviewClipGenerator { let video_path_owned = video_path.clone(); let relative_path_owned = relative_path.clone(); - tokio::spawn(async move { - match generate_preview_clip(&video_path_owned, &output_str).await { - Ok((duration, size)) => { - info!( - "Preview clip complete for '{}' ({:.1}s, {} bytes)", - relative_path_owned, duration, size - ); - let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); - let _ = dao.update_status( - &relative_path_owned, - "complete", - Some(duration as f32), - Some(size as i32), - None, - ); - } - Err(e) => { - error!( - "Failed to generate preview clip for '{}': {}", - relative_path_owned, e - ); - let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); - let _ = dao.update_status( - &relative_path_owned, - "failed", - None, - None, - Some(&e.to_string()), - ); + let clip_span = tracing::info_span!( + "preview_clip.generate", + rel_path = %relative_path_owned, + ); + tokio::spawn( + async move { + match generate_preview_clip(&video_path_owned, &output_str).await { + Ok((duration, size)) => { + info!( + "Preview clip complete for '{}' ({:.1}s, {} bytes)", + relative_path_owned, duration, size + ); + let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); + let _ = dao.update_status( + &relative_path_owned, + "complete", + Some(duration as f32), + Some(size as i32), + None, + ); + } + Err(e) => { + error!( + "Failed to generate preview clip for '{}': {}", + relative_path_owned, e + ); + let mut dao = preview_dao.lock().expect("Unable to lock PreviewDao"); + let _ = dao.update_status( + &relative_path_owned, + "failed", + None, + None, + Some(&e.to_string()), + ); + } } + drop(permit); } - drop(permit); - }); + .instrument(clip_span), + ); }) } } diff --git a/src/video/mod.rs b/src/video/mod.rs index f28d302..0a4a072 100644 --- a/src/video/mod.rs +++ b/src/video/mod.rs @@ -1,10 +1,9 @@ -use crate::otel::global_tracer; use crate::thumbnails::{is_video, update_media_counts}; use crate::video::ffmpeg::{Ffmpeg, GifType}; use log::info; -use opentelemetry::trace::Tracer; use std::fs; use std::path::{Path, PathBuf}; +use tracing::Instrument; use walkdir::WalkDir; pub mod actors; @@ -14,67 +13,68 @@ pub mod legacy_migration; #[allow(dead_code)] pub async fn generate_video_gifs() { - tokio::spawn(async { - info!("Starting to make video gifs"); + tokio::spawn( + async { + info!("Starting to make video gifs"); - let start = std::time::Instant::now(); - let tracer = global_tracer(); - tracer.start("creating video gifs"); + let start = std::time::Instant::now(); - let gif_base_path = &dotenv::var("GIFS_DIRECTORY").unwrap_or(String::from("gifs")); - let gif_directory: &Path = Path::new(gif_base_path); - fs::create_dir_all(gif_base_path).expect("There was an issue creating directory"); + let gif_base_path = &dotenv::var("GIFS_DIRECTORY").unwrap_or(String::from("gifs")); + let gif_directory: &Path = Path::new(gif_base_path); + fs::create_dir_all(gif_base_path).expect("There was an issue creating directory"); - let files = PathBuf::from(dotenv::var("BASE_PATH").unwrap()); - // EXCLUDED_DIRS read here for the update_media_counts call below. - // The WalkDir walk itself is left raw — this function is currently - // dead code (`#[allow(dead_code)]`) and not reachable from main. - // If revived, swap to file_scan::walk_library_files (dual lib/bin - // module path makes that non-trivial here). - let excluded_dirs: Vec = std::env::var("EXCLUDED_DIRS") - .unwrap_or_default() - .split(',') - .filter(|s| !s.trim().is_empty()) - .map(|s| s.trim().to_string()) - .collect(); + let files = PathBuf::from(dotenv::var("BASE_PATH").unwrap()); + // EXCLUDED_DIRS read here for the update_media_counts call below. + // The WalkDir walk itself is left raw — this function is currently + // dead code (`#[allow(dead_code)]`) and not reachable from main. + // If revived, swap to file_scan::walk_library_files (dual lib/bin + // module path makes that non-trivial here). + let excluded_dirs: Vec = std::env::var("EXCLUDED_DIRS") + .unwrap_or_default() + .split(',') + .filter(|s| !s.trim().is_empty()) + .map(|s| s.trim().to_string()) + .collect(); - let ffmpeg = Ffmpeg; - for file in WalkDir::new(&files) - .into_iter() - .filter_map(|entry| entry.ok()) - .filter(|entry| entry.file_type().is_file()) - .filter(is_video) - .filter(|entry| { - let path = entry.path(); + let ffmpeg = Ffmpeg; + for file in WalkDir::new(&files) + .into_iter() + .filter_map(|entry| entry.ok()) + .filter(|entry| entry.file_type().is_file()) + .filter(is_video) + .filter(|entry| { + let path = entry.path(); + let relative_path = &path.strip_prefix(&files).unwrap(); + let thumb_path = Path::new(gif_directory).join(relative_path); + let gif_path = thumb_path.with_extension("gif"); + !gif_path.exists() + }) + { + let path = file.path(); let relative_path = &path.strip_prefix(&files).unwrap(); - let thumb_path = Path::new(gif_directory).join(relative_path); - let gif_path = thumb_path.with_extension("gif"); - !gif_path.exists() - }) - { - let path = file.path(); - let relative_path = &path.strip_prefix(&files).unwrap(); - let gif_path = Path::new(gif_directory).join(relative_path); - let gif_path = gif_path.with_extension("gif"); - if let Some(parent_dir) = gif_path.parent() { - fs::create_dir_all(parent_dir).unwrap_or_else(|_| { - panic!("There was an issue creating gif directory {:?}", gif_path) - }); + let gif_path = Path::new(gif_directory).join(relative_path); + let gif_path = gif_path.with_extension("gif"); + if let Some(parent_dir) = gif_path.parent() { + fs::create_dir_all(parent_dir).unwrap_or_else(|_| { + panic!("There was an issue creating gif directory {:?}", gif_path) + }); + } + info!("Generating gif for {:?}", path); + + ffmpeg + .generate_video_gif( + path.to_str().unwrap(), + gif_path.to_str().unwrap(), + GifType::Overview, + ) + .await + .expect("There was an issue generating the gif"); } - info!("Generating gif for {:?}", path); - ffmpeg - .generate_video_gif( - path.to_str().unwrap(), - gif_path.to_str().unwrap(), - GifType::Overview, - ) - .await - .expect("There was an issue generating the gif"); + info!("Finished making video gifs in {:?}", start.elapsed()); + + update_media_counts(&files, &excluded_dirs); } - - info!("Finished making video gifs in {:?}", start.elapsed()); - - update_media_counts(&files, &excluded_dirs); - }); + .instrument(tracing::info_span!("video.create_gifs")), + ); } diff --git a/src/watcher.rs b/src/watcher.rs index 98fa20e..f10f243 100644 --- a/src/watcher.rs +++ b/src/watcher.rs @@ -125,6 +125,7 @@ pub fn cleanup_orphaned_playlists( } info!("Running orphaned HLS cleanup"); + let _pass = tracing::info_span!("watcher.playlist_cleanup").entered(); let start = std::time::Instant::now(); // Snapshot every live content_hash currently in image_exif. @@ -418,6 +419,15 @@ pub fn watch_files( let is_full_scan = since_last_full.as_secs() >= full_interval_secs; + // Root span per tick: this loop runs on a plain std thread, so a + // synchronous `entered()` guard is the right shape. Everything the + // tick does (probes, drains, ingest, maintenance) nests under it. + let _tick = tracing::info_span!( + "watcher.tick", + scan = if is_full_scan { "full" } else { "quick" }, + ) + .entered(); + // Fresh snapshot per tick — picks up PATCH /libraries/{id} // mutations to `enabled` / `excluded_dirs` without restart. let libs: Vec = @@ -436,6 +446,12 @@ pub fn watch_files( ); continue; } + let _lib_span = tracing::info_span!( + "watcher.library", + library = %lib.name, + library_id = lib.id, + ) + .entered(); // Availability probe: every tick checks that the // library's mount is reachable, is a directory, is diff --git a/tests/otel_background.rs b/tests/otel_background.rs new file mode 100644 index 0000000..2419c1d --- /dev/null +++ b/tests/otel_background.rs @@ -0,0 +1,106 @@ +//! Verifies background-job root spans (#142 / map #135). +//! +//! Background work has no request parent, so each job opens its own root +//! span and must carry it into whatever concurrency it fans out to. Two +//! production shapes are pinned here: +//! +//! 1. **Watcher tick → per-pass runtime.** `watcher::watch_files` runs on a +//! plain std thread and enters a `watcher.tick` span; `face_watch` / +//! `clip_watch` then build a multi-thread runtime, `block_on` it, and +//! `tokio::spawn` one task per candidate. Each task must carry the +//! tick's span via `.instrument(parent.clone())`. +//! 2. **Rayon fan-out.** `thumbnails::create_thumbnails` walks files on +//! rayon workers, where the ambient span isn't current, so child spans +//! name their parent explicitly (`info_span!(parent: &span, ...)`). +//! +//! One process-global subscriber per test binary, so this file holds a +//! single test. + +use opentelemetry::trace::{SpanId, TracerProvider}; +use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider, SpanData}; +use rayon::prelude::*; +use tracing::Instrument; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::registry::Registry; +use tracing_subscriber::util::SubscriberInitExt; + +fn find<'a>(spans: &'a [SpanData], name: &str) -> &'a SpanData { + spans + .iter() + .find(|s| s.name == name) + .unwrap_or_else(|| panic!("span {name} not exported")) +} + +#[tracing::instrument(name = "test.detect_one", skip_all)] +async fn detect_one() { + tokio::task::yield_now().await; +} + +#[test] +fn background_work_nests_under_job_root() { + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + Registry::default() + .with(tracing_opentelemetry::OpenTelemetryLayer::new( + provider.tracer("background-test"), + )) + .try_init() + .expect("global subscriber not yet set"); + + // Shape 1: std-thread tick with an entered root, per-pass runtime, spawns. + std::thread::spawn(|| { + let _tick = tracing::info_span!("test.tick").entered(); + let rt = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let parent = tracing::Span::current(); + rt.block_on(async move { + let handles: Vec<_> = (0..3) + .map(|_| tokio::spawn(detect_one().instrument(parent.clone()))) + .collect(); + for h in handles { + h.await.unwrap(); + } + }); + }) + .join() + .unwrap(); + + // Shape 2: rayon fan-out with an explicit parent. + { + let root = tracing::info_span!("test.thumbnails"); + (0..4).into_par_iter().for_each(|_| { + let _child = tracing::info_span!(parent: &root, "test.thumbnail_one"); + }); + } + let _ = provider.force_flush(); + + let spans = exporter.get_finished_spans().unwrap(); + + let tick = find(&spans, "test.tick"); + assert_eq!(tick.parent_span_id, SpanId::INVALID, "tick is a root span"); + let detects: Vec<_> = spans + .iter() + .filter(|s| s.name == "test.detect_one") + .collect(); + assert_eq!(detects.len(), 3); + for d in detects { + assert_eq!(d.span_context.trace_id(), tick.span_context.trace_id()); + assert_eq!(d.parent_span_id, tick.span_context.span_id()); + } + + let root = find(&spans, "test.thumbnails"); + let children: Vec<_> = spans + .iter() + .filter(|s| s.name == "test.thumbnail_one") + .collect(); + assert_eq!(children.len(), 4); + for c in children { + assert_eq!(c.span_context.trace_id(), root.span_context.trace_id()); + assert_eq!(c.parent_span_id, root.span_context.span_id()); + } +} diff --git a/tests/otel_db_spans.rs b/tests/otel_db_spans.rs index f7c336d..ead0850 100644 --- a/tests/otel_db_spans.rs +++ b/tests/otel_db_spans.rs @@ -14,8 +14,8 @@ //! One process-global subscriber per test binary, so this file holds a //! single test. -use opentelemetry::trace::{Status, TracerProvider}; use opentelemetry::Value; +use opentelemetry::trace::{Status, TracerProvider}; use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider, SpanData}; use tracing::Instrument; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -85,7 +85,10 @@ fn db_spans_nest_under_ambient_span() { let del = find(&spans, "db.delete.remove_tag"); for (span, qt, op) in [(get, "query", "get_tags"), (del, "delete", "remove_tag")] { - assert_eq!(span.span_context.trace_id(), handler.span_context.trace_id()); + assert_eq!( + span.span_context.trace_id(), + handler.span_context.trace_id() + ); assert_eq!( span.parent_span_id, handler.span_context.span_id(),