Tracing: root spans for background jobs, retire raw OTel spans (#142)

Background work has no request parent, so each job now opens a named
tracing root and carries it into its fan-out:

- watcher: `watcher.tick` (scan=quick|full) entered per tick on the
  watcher's std thread, `watcher.library` per library, and
  `watcher.playlist_cleanup` per orphaned-HLS pass. face_watch /
  clip_watch carry the ambient span into every per-candidate
  tokio::spawn and give each call its own `face_watch.detect` /
  `clip_watch.encode` span.
- thumbnails: `creating thumbnails` via #[instrument]; per-video spans on
  rayon workers name their parent explicitly.
- video actors: `playlistgenerator.generate_playlist` and
  `preview_clip.generate` are tracing roots that instrument their futures.
- reels: `reel.job` (created before the spawn, so on-demand jobs stay in
  the request trace), `reel.pregen_batch`, `reel.cache_sweep`.
- RefreshThumbnailsMessage and the dead-code GIF job move to tracing.

No raw OpenTelemetry span creation is left outside otel.rs;
`global_tracer` is removed.

tests/otel_background.rs pins the std-thread tick -> block_on -> spawn
shape and the rayon explicit-parent shape.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Cameron CordesandClaude Opus 5.5 committed 2026-09-29 17:02:40 -04:00
1 parent b6deca2a52
commit 26115b3378
11 files changed
+330 -179

No files matched your search

+13 -4
View File
@@ -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,
+24 -15
View File
@@ -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,
+2 -4
View File
@@ -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<RefreshThumbnailsMessage> 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
+3 -8
View File
@@ -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);
+41 -28
View File
@@ -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<AppState>) {
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<AppState>) {
// 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)");
}
+10 -18
View File
@@ -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),
+53 -43
View File
@@ -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<GeneratePlaylistMessage> 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<GeneratePlaylistMessage> for PlaylistGenerator {
span.set_status(Status::error(detail.clone()));
Err(std::io::Error::other(detail))
}
})
}
.instrument(fut_span))
}
}
@@ -818,39 +821,46 @@ impl Handler<GeneratePreviewClipMessage> 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),
);
})
}
}
+57 -57
View File
@@ -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<String> = 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<String> = 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")),
);
}
+16
View File
@@ -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<libraries::Library> =
@@ -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