starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Instrument logs (#3566)

Browse Source 
* Add spans to show worker_name and job_id

Always show worker_name and job_id when relevant, by creating spans that
live as long as the worker and the job.

* Remove redundant worker names from fn run_worker

Remove the now redundant worker_name tracing, as the information now
lives at the span level

* Use span when spawning new thread

* Instrument more pertinent functions

Move spans to more pertinent functions. Remove more redundant info, and
change worker_name to just worker

* Change tracing subscriber to log path:line_no

Change from logging target (module name usually) to file path and line
number
This commit is contained in:
wendrul
2024-04-18 01:11:14 +02:00
committed by GitHub
parent da3ded6196
commit c3325c2cbf
4 changed files with 34 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/tracing_init.rs
+7 -1
View File
@@ -53,7 +53,13 @@ pub fn initialize_tracing() {
match json_fmt {
true => ts_base.with(json_layer().flatten_event(true)).init(),
false => ts_base
.with(compact_layer().with_ansi(style.to_lowercase() != "never"))
.with(
compact_layer()
.with_ansi(style.to_lowercase() != "never")
.with_file(true)
.with_line_number(true)
.with_target(false),
)
.init(),
}
}
backend/windmill-worker/src/python_executor.rs
+1 -9
View File
@@ -391,8 +391,6 @@ mount {{
}
tracing::info!(
worker_name = %worker_name,
job_id = %job.id,
workspace_id = %job.workspace_id,
"started python code execution {}",
job.id
@@ -614,7 +612,7 @@ async fn replace_pip_secret(
}
let secret = get_secret_value_as_admin(db, w_id, variable).await?;
tracing::info!(
worker_name = %worker_name,
worker = %worker_name,
job_id = %job_id,
workspace_id = %w_id,
"found secret variable in pip requirements: {}",
@@ -878,16 +876,12 @@ pub async fn handle_python_reqs(
append_logs(job_id.clone(), w_id.to_string(), logs1, db).await;
tracing::info!(
worker_name = %worker_name,
job_id = %job_id,
workspace_id = %w_id,
"started setup python dependencies"
);
let child = if !*DISABLE_NSJAIL {
tracing::info!(
worker_name = %worker_name,
job_id = %job_id,
workspace_id = %w_id,
"starting nsjail"
);
@@ -992,8 +986,6 @@ pub async fn handle_python_reqs(
)
.await;
tracing::info!(
worker_name = %worker_name,
job_id = %job_id,
workspace_id = %w_id,
is_ok = child.is_ok(),
"finished setting up python dependencies {}",
backend/windmill-worker/src/worker.rs
+24 -24
View File
@@ -13,6 +13,7 @@ use prometheus::{
core::{AtomicI64, GenericGauge},
IntCounter,
};
use tracing::Instrument;
#[cfg(feature = "prometheus")]
use windmill_common::METRICS_DEBUG_ENABLED;
#[cfg(feature = "prometheus")]
@@ -584,6 +585,7 @@ impl JobCompletedSender {
}
}
#[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name))]
pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
db: &Pool<Postgres>,
worker_instance: &str,
@@ -608,7 +610,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let start_time = Instant::now();
let worker_dir = format!("{TMP_DIR}/{worker_name}");
tracing::debug!(worker_dir = %worker_dir, worker_name = %worker_name, "Creating worker dir");
tracing::debug!(worker_dir = %worker_dir, "Creating worker dir");
if let Some(ref netrc) = *NETRC {
tracing::info!("Writing netrc at {}/.netrc", HOME_ENV.as_str());
@@ -1022,7 +1024,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let worker_name2 = worker_name.clone();
let killpill_tx2 = killpill_tx.clone();
let job_completed_sender = job_completed_tx.0.clone();
let send_result = tokio::spawn(async move {
let send_result = tokio::spawn((async move {
while let Some(sr) = job_completed_rx.recv().await {
match sr {
SendResult::JobCompleted(jc) => {
@@ -1200,7 +1202,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
tokio::time::sleep(Duration::from_millis(50)).await;
}
tracing::info!("finished processing all completed jobs");
});
}).instrument(tracing::Span::current()));
let mut last_executed_job: Option<Instant> = None;
let mut last_checked_suspended = Instant::now();
@@ -1214,7 +1216,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let vacuum_shift = rand::thread_rng().gen_range(0..VACUUM_PERIOD);
IS_READY.store(true, Ordering::Relaxed);
tracing::info!(worker = %worker_name, "listening for jobs, WORKER_GROUP: {}, config: {:?}", *WORKER_GROUP, WORKER_CONFIG.read().await);
tracing::info!("listening for jobs, WORKER_GROUP: {}, config: {:?}", *WORKER_GROUP, WORKER_CONFIG.read().await);
// (dedi_path, dedicated_worker_tx, dedicated_worker_handle)
// Option<Sender<Arc<QueuedJob>>>,
@@ -1385,17 +1387,17 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if (jobs_executed as u32 + vacuum_shift) % VACUUM_PERIOD == 0 {
let db2 = db.clone();
let worker_name2 = worker_name.clone();
tokio::task::spawn(async move {
tracing::info!(worker = %worker_name2, "vacuuming queue and completed_job");
let current_span = tracing::Span::current();
tokio::task::spawn((async move {
tracing::info!("vacuuming queue and completed_job");
if let Err(e) = sqlx::query!("VACUUM (skip_locked) queue")
.execute(&db2)
.await
{
tracing::error!(worker = %worker_name2, "failed to vacuum queue: {}", e);
tracing::error!("failed to vacuum queue: {}", e);
}
tracing::info!(worker = %worker_name2, "vacuumed queue and completed_job");
});
tracing::info!("vacuumed queue and completed_job");
}).instrument(current_span));
jobs_executed += 1;
}
@@ -1499,7 +1501,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
last_executed_job = None;
jobs_executed += 1;
tracing::debug!(worker = %worker_name, "started handling of job {}", job.id);
tracing::debug!("started handling of job {}", job.id);
if matches!(job.job_kind, JobKind::Script | JobKind::Preview) {
if !dedicated_workers.is_empty() {
@@ -1611,10 +1613,12 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.unwrap_or_else(|| "none".to_string());
if job.id == Uuid::nil() {
tracing::info!(worker = %worker_name, "running warmup job");
tracing::info!("running warmup job");
} else {
tracing::info!(worker = %worker_name, workspace_id = %job.workspace_id, id = %job.id, root_id = %job_root, "fetched job {}, root job: {}", job.id, job_root);
}
tracing::info!(workspace_id = %job.workspace_id, job_id = %job.id, root_id = %job_root, "fetched job {}, root job: {}", job.id, job_root);
} // Here we can't remove the job id, but maybe with the
// fields macro we can make a job id that only appears when
// the job is defined?
let job_dir = format!("{worker_dir}/{}", job.id);
@@ -1739,7 +1743,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if let Some(secs) = *EXIT_AFTER_NO_JOB_FOR_SECS {
if let Some(lj) = last_executed_job {
if lj.elapsed().as_secs() > secs {
tracing::info!(worker = %worker_name, "no job for {} seconds, exiting", secs);
tracing::info!("no job for {} seconds, exiting", secs);
break;
}
} else {
@@ -1768,7 +1772,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
});
}
Err(err) => {
tracing::error!(worker = %worker_name, "Failed to pull jobs: {}", err);
tracing::error!("Failed to pull jobs: {}", err);
}
};
}
@@ -1797,7 +1801,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
send_result.await.expect("send result failed");
tracing::info!("worker {} exited", worker_name);
println!("worker {} exited", worker_name);
}
type DedicatedWorker = (String, Sender<Arc<QueuedJob>>, Option<JoinHandle<()>>);
@@ -2218,6 +2221,7 @@ async fn queue_init_bash_maybe<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
// logs: String,
// ) -> error::Result<()> {
#[tracing::instrument(name = "completed_job", level = "info", skip_all, fields(job_id = %job.id))]
pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, .. }: JobCompleted,
client: &AuthedClient,
@@ -2359,6 +2363,8 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
// *barrier = None;
// tracing::debug!("leader worker done waiting for");
// }
#[tracing::instrument(name = "job_error", level = "info", skip_all, fields(job_id = %job.id))]
pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
db: &Pool<Postgres>,
client: &AuthedClient,
@@ -2554,7 +2560,7 @@ pub struct PreviousResult<'a> {
pub previous_result: Option<&'a RawValue>,
}
#[tracing::instrument(level = "trace", skip_all)]
#[tracing::instrument(name = "job", level = "info", skip_all, fields(job_id = %job.id))]
async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
job: Arc<QueuedJob>,
db: &DB,
@@ -2731,8 +2737,6 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
}
tracing::debug!(
worker = %worker_name,
job_id = %job.id,
workspace_id = %job.workspace_id,
"handling job {}",
job.id
@@ -3210,8 +3214,6 @@ async fn handle_code_execution_job(
.unwrap_or_else(|| "NO_LANG".to_string());
tracing::debug!(
worker_name = %worker_name,
job_id = %job.id,
workspace_id = %job.workspace_id,
"started {} job {}",
&lang_str,
@@ -3346,8 +3348,6 @@ mount {{
_ => panic!("unreachable, language is not supported: {language:#?}"),
};
tracing::info!(
worker_name = %worker_name,
job_id = %job.id,
workspace_id = %job.workspace_id,
is_ok = result.is_ok(),
"finished {} job {}",
backend/windmill-worker/src/worker_flow.rs
+2 -2
View File
@@ -671,7 +671,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.root_job
.map(|x| x.to_string())
.unwrap_or_else(|| "none".to_string());
tracing::info!(id = %flow_job.id, root_id = %job_root, worker_name = %worker_name, "update flow status");
tracing::info!(id = %flow_job.id, root_id = %job_root, "update flow status");
let module = get_module(&flow_job, module_index);
@@ -869,7 +869,7 @@ pub async fn update_flow_status_after_job_completion_internal<
}
if let Some(parent_job) = flow_job.parent_job {
tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, worker_name = %worker_name, "subflow is finished, updating parent flow status");
tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status");
return Ok(Some(RecUpdateFlowStatusAfterJobCompletion {
flow: parent_job,