mirrors/background-jobs
1
0
Fork 0
mirror of https://git.asonix.dog/asonix/background-jobs.git synced 2024-10-23 02:44:19 +00:00
Code Issues Projects Releases Packages Wiki Activity

Don't have long-running spans

Browse source
This commit is contained in:
Aode (Lion) 2021-09-21 12:02:58 -05:00
parent 1d30a9472c
commit cce5a97306
2 changed files with 45 additions and 33 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
jobs-actix/Cargo.toml
View file

@ -1,7 +1,7 @@
[package]
name = "background-jobs-actix"
description = "in-process jobs processor based on Actix"
version = "0.10.0"
version = "0.10.1"
license = "AGPL-3.0"
authors = ["asonix <asonix@asonix.dog>"]
repository = "https://git.asonix.dog/Aardwolf/background-jobs"

80
jobs-actix/src/worker.rs
View file

@ -2,7 +2,7 @@ use crate::Server;
use actix_rt::spawn;
use background_jobs_core::{CachedProcessorMap, JobInfo};
use tokio::sync::mpsc::{channel, Sender};
use tracing::{debug, error, info, warn, Span};
use tracing::{error, info, warn, Span};
use tracing_futures::Instrument;
use uuid::Uuid;
@ -22,6 +22,20 @@ pub(crate) struct LocalWorkerHandle {
queue: String,
}
impl LocalWorkerHandle {
fn span(&self, operation: &str) -> Span {
tracing::info_span!(
"Worker",
worker.id = tracing::field::display(&self.id),
worker.queue = tracing::field::display(&self.queue),
worker.operation.id = tracing::field::display(&Uuid::new_v4()),
worker.operation.name = tracing::field::display(operation),
exception.message = tracing::field::Empty,
exception.details = tracing::field::Empty,
)
}
}
#[async_trait::async_trait]
impl Worker for LocalWorkerHandle {
async fn process(&self, job: JobInfo) -> Result<(), JobInfo> {
@ -50,50 +64,48 @@ pub(crate) fn local_worker<State>(
) where
State: Clone + 'static,
{
spawn(async move {
let id = Uuid::new_v4();
let span = tracing::info_span!(
"Worker",
worker.id = tracing::field::display(&id),
worker.queue = tracing::field::display(&queue),
exception.message = tracing::field::Empty,
exception.details = tracing::field::Empty,
);
spawn(
async move {
let (tx, mut rx) = channel(16);
let handle = LocalWorkerHandle { tx, id, queue };
let span = Span::current();
debug!("Beginning worker loop for {}", id);
if let Err(e) = server.request_job(Box::new(handle.clone())).await {
loop {
let span = handle.span("request");
if let Err(e) = server
.request_job(Box::new(handle.clone()))
.instrument(span.clone())
.await
{
let display = format!("{}", e);
let debug = format!("{:?}", e);
span.record("exception.message", &tracing::field::display(&display));
span.record("exception.details", &tracing::field::display(&debug));
error!("Failed to notify server of new worker, {}", e);
return;
}
while let Some(job) = rx.recv().await {
let return_job = processors.process(job).await;
if let Err(e) = server.return_job(return_job).await {
warn!("Failed to return completed job, {}", e);
}
if let Err(e) = server.request_job(Box::new(handle.clone())).await {
let display = format!("{}", e);
let debug = format!("{:?}", e);
span.record("exception.message", &tracing::field::display(&display));
span.record("exception.details", &tracing::field::display(&debug));
error!("Failed to notify server of ready worker, {}", e);
span.in_scope(|| error!("Failed to notify server of ready worker, {}", e));
break;
}
drop(span);
if let Some(job) = rx.recv().await {
let return_job = processors
.process(job)
.instrument(handle.span("process"))
.await;
let span = handle.span("return");
if let Err(e) = server.return_job(return_job).instrument(span.clone()).await {
let display = format!("{}", e);
let debug = format!("{:?}", e);
span.record("exception.message", &tracing::field::display(&display));
span.record("exception.details", &tracing::field::display(&debug));
span.in_scope(|| warn!("Failed to return completed job, {}", e));
}
info!("Worker closing");
continue;
}
.instrument(span),
);
break;
}
handle.span("closing").in_scope(|| info!("Worker closing"));
});
}