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

fix: make all error strings more verbose

Browse Source
This commit is contained in:
Ruben Fiszel
2024-05-25 22:15:15 +02:00
parent 218ea8ce3e
commit 6a332bc30f
23 changed files with 111 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+3 -3
View File
@@ -357,7 +357,7 @@ Windmill Community Edition {GIT_VERSION}
base_internal_tx
.send(base_internal_url.clone())
.map_err(|e| {
anyhow::anyhow!("Could not send base_internal_url to agent: {e}")
anyhow::anyhow!("Could not send base_internal_url to agent: {e:#}")
})?;
}
Ok(()) as anyhow::Result<()>
@@ -457,7 +457,7 @@ Windmill Community Edition {GIT_VERSION}
},
DEFAULT_TAGS_PER_WORKSPACE_SETTING => {
if let Err(e) = load_tag_per_workspace_enabled(&db).await {
tracing::error!("Error loading default tag per workpsace: {e}");
tracing::error!("Error loading default tag per workpsace: {e:#}");
}
},
RETENTION_PERIOD_SECS_SETTING => {
@@ -561,7 +561,7 @@ Windmill Community Edition {GIT_VERSION}
});
if let Err(e) = h.await {
tracing::error!("Error waiting for monitor handle:{e}")
tracing::error!("Error waiting for monitor handle:{e:#}")
}
Ok(()) as anyhow::Result<()>
};
backend/src/monitor.rs
+6 -6
View File
@@ -104,15 +104,15 @@ pub async fn initial_load(
_is_agent: bool,
) {
if let Err(e) = load_metrics_enabled(db).await {
tracing::error!("Error loading expose metrics: {e}");
tracing::error!("Error loading expose metrics: {e:#}");
}
if let Err(e) = load_metrics_debug_enabled(db).await {
tracing::error!("Error loading expose debug metrics: {e}");
tracing::error!("Error loading expose debug metrics: {e:#}");
}
if let Err(e) = load_tag_per_workspace_enabled(db).await {
tracing::error!("Error loading default tag per workpsace: {e}");
tracing::error!("Error loading default tag per workpsace: {e:#}");
}
if server_mode {
@@ -273,7 +273,7 @@ pub async fn load_keep_job_dir(db: &DB) {
match value {
Ok(Some(serde_json::Value::Bool(t))) => KEEP_JOB_DIR.store(t, Ordering::Relaxed),
Err(e) => {
tracing::error!("Error loading keep job dir metrics: {e}");
tracing::error!("Error loading keep job dir metrics: {e:#}");
}
_ => (),
};
@@ -287,7 +287,7 @@ pub async fn load_require_preexisting_user(db: &DB) {
REQUIRE_PREEXISTING_USER_FOR_OAUTH.store(t, Ordering::Relaxed)
}
Err(e) => {
tracing::error!("Error loading keep job dir metrics: {e}");
tracing::error!("Error loading keep job dir metrics: {e:#}");
}
_ => (),
};
@@ -950,7 +950,7 @@ pub async fn reload_worker_config(
tracing::info!("Waiting 5 seconds to allow others workers to start potential jobs that depend on a potential shared cache volume");
tokio::time::sleep(Duration::from_secs(5)).await;
if let Err(e) = windmill_worker::common::clean_cache().await {
tracing::error!("Error cleaning the cache: {e}");
tracing::error!("Error cleaning the cache: {e:#}");
}
}
}
backend/windmill-api/src/apps.rs
+1 -1
View File
@@ -736,7 +736,7 @@ async fn delete_app(
.await
.map_err(|e| {
Error::InternalErr(format!(
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e}"
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e:#}"
))
})?;
backend/windmill-api/src/db.rs
+3 -3
View File
@@ -86,7 +86,7 @@ impl Migrate for CustomMigrator {
.fetch_one(&mut *self.inner)
.await
.map_err(|e| {
tracing::error!("Error acquiring lock: {e}");
tracing::error!("Error acquiring lock: {e:#}");
sqlx::migrate::MigrateError::Execute(e)
})?
.unwrap_or(false);
@@ -175,7 +175,7 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
Err(sqlx::migrate::MigrateError::VersionMissing(e)) => {
tracing::error!("Database had been applied more migrations than this container.
This usually mean than another container on a more recent version migrated the database and this one is on an earlier version.
Please update the container to latest. Not critical, but may cause issues if migration introduced a breaking change. Version missing: {e}");
Please update the container to latest. Not critical, but may cause issues if migration introduced a breaking change. Version missing: {e:#}");
custom_migrator.unlock().await?;
Ok(())
}
@@ -184,7 +184,7 @@ pub async fn migrate(db: &DB) -> Result<(), Error> {
#[cfg(feature = "enterprise")]
if let Err(e) = windmill_migrations(&mut custom_migrator, db).await {
tracing::error!("Could not apply windmill custom migrations: {e}")
tracing::error!("Could not apply windmill custom migrations: {e:#}")
}
Ok(())
backend/windmill-api/src/flows.rs
+6 -6
View File
@@ -504,7 +504,7 @@ async fn update_flow(
nf.visible_to_runner_only.unwrap_or(false),
)
.execute(&mut tx)
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to flow update: {e}")))?;
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to flow update: {e:#}")))?;
if nf.path != flow_path {
check_schedule_conflict(tx.transaction_mut(), &w_id, &nf.path).await?;
@@ -520,7 +520,7 @@ async fn update_flow(
.bind(&flow_path)
.bind(&w_id)
.fetch_all(&mut tx)
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to related schedules update: {e}")))?;
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to related schedules update: {e:#}")))?;
let schedule = sqlx::query_as::<_, Schedule>(
"UPDATE schedule SET path = $1, script_path = $1 WHERE path = $2 AND workspace_id = $3 AND is_flow IS true RETURNING *")
@@ -528,7 +528,7 @@ async fn update_flow(
.bind(&flow_path)
.bind(&w_id)
.fetch_optional(&mut tx)
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to related schedule update: {e}")))?;
.await.map_err(|e| error::Error::InternalErr(format!("Error updating flow due to related schedule update: {e:#}")))?;
if let Some(schedule) = schedule {
clear_schedule(tx.transaction_mut(), &flow_path, &w_id).await?;
@@ -621,7 +621,7 @@ async fn update_flow(
.await
.map_err(|e| {
error::Error::InternalErr(format!(
"Error updating flow due to updating dependency job field: {e}"
"Error updating flow due to updating dependency job field: {e:#}"
))
})?;
if let Some(old_dep_job) = old_dep_job {
@@ -633,7 +633,7 @@ async fn update_flow(
.await
.map_err(|e| {
error::Error::InternalErr(format!(
"Error updating flow due to cancelling dependency job: {e}"
"Error updating flow due to cancelling dependency job: {e:#}"
))
})?;
}
@@ -854,7 +854,7 @@ async fn delete_flow_by_path(
.await
.map_err(|e| {
Error::InternalErr(format!(
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e}"
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e:#}"
))
})?;
backend/windmill-api/src/jobs.rs
+2 -2
View File
@@ -484,7 +484,7 @@ pub async fn get_path_for_hash<'c>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"querying getting path for hash {hash} in {w_id}: {e}"
"querying getting path for hash {hash} in {w_id}: {e:#}"
))
})?;
Ok(path)
@@ -516,7 +516,7 @@ pub async fn get_path_tag_limits_cache_for_hash(
.await
.map_err(|e| {
Error::InternalErr(format!(
"querying getting path for hash {hash} in {w_id}: {e}"
"querying getting path for hash {hash} in {w_id}: {e:#}"
))
})?;
Ok((
backend/windmill-api/src/lib.rs
+1 -1
View File
@@ -130,7 +130,7 @@ pub async fn run_server(
let r =
rsmq_async::RsmqConnection::create_queue(&mut rsmq, &tag, None, None, None).await;
if let Err(e) = r {
tracing::info!("Redis queue {tag} could not be created: {e}");
tracing::info!("Redis queue {tag} could not be created: {e:#}");
} else {
tracing::info!("Redis queue {tag} created");
}
backend/windmill-api/src/openai.rs
+2 -2
View File
@@ -198,7 +198,7 @@ async fn proxy(
}
let config: OpenaiConfig = serde_json::from_value(resource.unwrap())
.map_err(|e| Error::InternalErr(format!("validating openai resource {e}")))?;
.map_err(|e| Error::InternalErr(format!("validating openai resource {e:#}")))?;
let mut user = None::<String>;
let mut resource = match config {
@@ -238,7 +238,7 @@ async fn proxy(
let azure_base_path = if let Some(azure_base_path) = azure_base_path {
Some(
serde_json::from_value::<String>(azure_base_path).map_err(|e| {
Error::InternalErr(format!("validating openai azure base path {e}"))
Error::InternalErr(format!("validating openai azure base path {e:#}"))
})?,
)
} else {
backend/windmill-api/src/schedule.rs
+2 -2
View File
@@ -185,7 +185,7 @@ async fn create_schedule(
.bind(&ns.tag)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("inserting schedule in {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("inserting schedule in {w_id}: {e:#}")))?;
handle_deployment_metadata(
&authed.email,
@@ -268,7 +268,7 @@ async fn edit_schedule(
.bind(&w_id)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("updating schedule in {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("updating schedule in {w_id}: {e:#}")))?;
handle_deployment_metadata(
&authed.email,
backend/windmill-api/src/scripts.rs
+5 -5
View File
@@ -1160,7 +1160,7 @@ async fn archive_script_by_path(
)
.fetch_one(&db)
.await
.map_err(|e| Error::InternalErr(format!("archiving script in {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("archiving script in {w_id}: {e:#}")))?;
audit_log(
&mut *tx,
&authed,
@@ -1211,7 +1211,7 @@ async fn archive_script_by_hash(
.bind(&hash.0)
.fetch_one(&mut *tx)
.await
.map_err(|e| Error::InternalErr(format!("archiving script in {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("archiving script in {w_id}: {e:#}")))?;
audit_log(
&mut *tx,
@@ -1251,7 +1251,7 @@ async fn delete_script_by_hash(
.bind(&w_id)
.fetch_one(&db)
.await
.map_err(|e| Error::InternalErr(format!("deleting script by hash {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("deleting script by hash {w_id}: {e:#}")))?;
audit_log(
&mut *tx,
@@ -1310,7 +1310,7 @@ async fn delete_script_by_path(
)
.fetch_one(&db)
.await
.map_err(|e| Error::InternalErr(format!("deleting script by path {w_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("deleting script by path {w_id}: {e:#}")))?;
sqlx::query!(
"DELETE FROM draft WHERE path = $1 AND workspace_id = $2 AND typ = 'script'",
@@ -1357,7 +1357,7 @@ async fn delete_script_by_path(
.await
.map_err(|e| {
Error::InternalErr(format!(
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e}"
"error deleting deployment metadata for script with path {path} in workspace {w_id}: {e:#}"
))
})?;
backend/windmill-api/src/settings.rs
+2 -2
View File
@@ -115,7 +115,7 @@ pub async fn test_s3_bucket(
.ok_or_else(|| {
error::Error::InternalErr("Failed to list files in blob storage".to_string())
})?
.map_err(|e| anyhow::anyhow!("error listing bucket: {e}"))?;
.map_err(|e| anyhow::anyhow!("error listing bucket: {e:#}"))?;
tracing::info!("Listed files: {:?}", first_file);
let path = object_store::path::Path::from(format!(
"/test-s3-bucket-{uuid}",
@@ -125,7 +125,7 @@ pub async fn test_s3_bucket(
client
.put(&path, Bytes::from_static(b"hello"))
.await
.map_err(|e| anyhow::anyhow!("error writing file to {path}: {e}"))?;
.map_err(|e| anyhow::anyhow!("error writing file to {path}: {e:#}"))?;
let content = client
.get(&path)
.await
backend/windmill-api/src/utils.rs
+2 -2
View File
@@ -21,7 +21,7 @@ pub async fn require_super_admin(db: &DB, email: &str) -> error::Result<()> {
let is_admin = sqlx::query_scalar!("SELECT super_admin FROM password WHERE email = $1", email)
.fetch_optional(db)
.await
.map_err(|e| Error::InternalErr(format!("fetching super admin: {e}")))?
.map_err(|e| Error::InternalErr(format!("fetching super admin: {e:#}")))?
.unwrap_or(false);
if !is_admin {
@@ -148,7 +148,7 @@ pub async fn get_instance_username_or_create_pending<'c>(
)
.execute(&mut **tx)
.await
.map_err(|e| Error::InternalErr(format!("creating pending user: {e}")))?;
.map_err(|e| Error::InternalErr(format!("creating pending user: {e:#}")))?;
Ok(username)
}
backend/windmill-api/src/workspaces.rs
+3 -3
View File
@@ -376,7 +376,7 @@ async fn get_settings(
)
.fetch_one(&mut *tx)
.await
.map_err(|e| Error::InternalErr(format!("getting settings: {e}")))?;
.map_err(|e| Error::InternalErr(format!("getting settings: {e:#}")))?;
tx.commit().await?;
Ok(Json(settings))
@@ -399,7 +399,7 @@ async fn get_deploy_to(
)
.fetch_one(&mut *tx)
.await
.map_err(|e| Error::InternalErr(format!("getting deploy_to: {e}")))?;
.map_err(|e| Error::InternalErr(format!("getting deploy_to: {e:#}")))?;
tx.commit().await?;
Ok(Json(settings))
@@ -919,7 +919,7 @@ async fn get_copilot_info(
)
.fetch_one(&mut *tx)
.await
.map_err(|e| Error::InternalErr(format!("getting openai_resource_path and code_completion_enabled: {e}")))?;
.map_err(|e| Error::InternalErr(format!("getting openai_resource_path and code_completion_enabled: {e:#}")))?;
tx.commit().await?;
Ok(Json(CopilotInfo {
backend/windmill-common/src/jobs.rs
+1 -1
View File
@@ -434,7 +434,7 @@ pub async fn script_hash_to_tag_and_limits<'c>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"querying getting tag for hash {script_hash}: {e}"
"querying getting tag for hash {script_hash}: {e:#}"
))
})?;
Ok((
backend/windmill-common/src/variables.rs
+1 -1
View File
@@ -88,7 +88,7 @@ pub async fn get_workspace_key<'c>(
)
.fetch_one(&mut **db)
.await
.map_err(|e| crate::Error::InternalErr(format!("fetching workspace key: {e}")))?;
.map_err(|e| crate::Error::InternalErr(format!("fetching workspace key: {e:#}")))?;
Ok(key)
}
backend/windmill-queue/src/jobs.rs
+16 -16
View File
@@ -608,7 +608,7 @@ pub async fn add_completed_job<
)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
// tracing::error!("2 {:?}", start.elapsed());
if !queued_job.is_flow_step {
@@ -793,7 +793,7 @@ pub async fn add_completed_job<
.await
.map_err(|e| {
Error::InternalErr(format!(
"Error updating to add ended_at timestamp concurrency_key={concurrency_key}: {e}"
"Error updating to add ended_at timestamp concurrency_key={concurrency_key}: {e:#}"
))
}) {
tracing::error!("Could not update concurrency_key: {}", e);
@@ -827,7 +827,7 @@ pub async fn add_completed_job<
sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", w_id)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("fetching if {w_id} is premium: {e}")))?;
.map_err(|e| Error::InternalErr(format!("fetching if {w_id} is premium: {e:#}")))?;
let _ = sqlx::query!(
"INSERT INTO usage (id, is_workspace, month_, usage)
VALUES ($1, TRUE, EXTRACT(YEAR FROM current_date) * 12 + EXTRACT(MONTH FROM current_date), $2)
@@ -836,7 +836,7 @@ pub async fn add_completed_job<
additional_usage as i32)
.execute(db)
.await
.map_err(|e| Error::InternalErr(format!("updating usage: {e}")));
.map_err(|e| Error::InternalErr(format!("updating usage: {e:#}")));
if !premium_workspace {
let _ = sqlx::query!(
@@ -847,7 +847,7 @@ pub async fn add_completed_job<
additional_usage as i32)
.execute(db)
.await
.map_err(|e| Error::InternalErr(format!("updating usage: {e}")));
.map_err(|e| Error::InternalErr(format!("updating usage: {e:#}")));
}
}
@@ -1708,7 +1708,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"Error getting concurrency count for script path {job_script_path}: {e}"
"Error getting concurrency count for script path {job_script_path}: {e:#}"
))
})?;
tracing::debug!("running_job: {}", running_job.unwrap_or(0));
@@ -1719,7 +1719,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
f64::from(job_custom_concurrency_time_window_s),
).fetch_one(&mut tx).await.map_err(|e| {
Error::InternalErr(format!(
"Error getting completed count for key {job_concurrency_key}: {e}"
"Error getting completed count for key {job_concurrency_key}: {e:#}"
))
})?;
@@ -1736,7 +1736,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"Error getting concurrency count for script path {job_script_path}: {e}"
"Error getting concurrency count for script path {job_script_path}: {e:#}"
))
})?;
@@ -1764,7 +1764,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"Error decreasing concurrency count for script path {job_script_path}: {e}"
"Error decreasing concurrency count for script path {job_script_path}: {e:#}"
))
})?;
@@ -1840,7 +1840,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
))
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e:#}")))?;
if let Some(ref mut rsmq) = tx.rsmq {
rsmq.send_message(
@@ -1863,7 +1863,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
)
.fetch_all(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Could not update and re-queue job {job_uuid}. The job will be marked as running but it is not running: {e:#}")))?;
tx.commit().await?
}
}
@@ -2806,7 +2806,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
.await
.map_err(|e| {
Error::InternalErr(format!(
"fetching if {workspace_id} is premium and overquota: {e}"
"fetching if {workspace_id} is premium and overquota: {e:#}"
))
})?;
@@ -2824,7 +2824,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
)
.fetch_one(_db)
.await
.map_err(|e| Error::InternalErr(format!("updating usage: {e}")))?;
.map_err(|e| Error::InternalErr(format!("updating usage: {e:#}")))?;
let user_usage = if !premium_workspace {
Some(sqlx::query_scalar!(
@@ -2836,7 +2836,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
)
.fetch_one(_db)
.await
.map_err(|e| Error::InternalErr(format!("updating usage: {e}")))?)
.map_err(|e| Error::InternalErr(format!("updating usage: {e:#}")))?)
} else {
None
};
@@ -3547,7 +3547,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
)
.execute(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Could not insert concurrency_key={concurrency_key} for job_id={job_id} script_path={script_path:?} workspace_id={workspace_id}: {e:#}")))?;
}
let uuid = sqlx::query_scalar!(
@@ -3592,7 +3592,7 @@ pub async fn push<'c, R: rsmq_async::RsmqConnection + Send + 'c>(
)
.fetch_one(&mut tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id} with tag {tag}, schedule_path {schedule_path:?}, script_path: {script_path:?}, email {email}, workspace_id {workspace_id}: {e:#}")))?;
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
#[cfg(feature = "prometheus")]
backend/windmill-worker/src/bun_executor.rs
+1 -1
View File
@@ -450,7 +450,7 @@ pub async fn generate_wrapper_mjs(
format!("{job_dir}/wrapper.mjs"),
)
.await
.map_err(|e| error::Error::InternalErr(format!("Could not move wrapper to mjs: {e}")))?;
.map_err(|e| error::Error::InternalErr(format!("Could not move wrapper to mjs: {e:#}")))?;
Ok(())
}
backend/windmill-worker/src/common.rs
+10 -10
View File
@@ -172,13 +172,13 @@ pub async fn transform_json<'a>(
let inner_vs = v.get();
if (*RE_RES_VAR).is_match(inner_vs) {
let value = serde_json::from_str(inner_vs).map_err(|e| {
error::Error::InternalErr(format!("Error while parsing inner arg: {e}"))
error::Error::InternalErr(format!("Error while parsing inner arg: {e:#}"))
})?;
let transformed =
transform_json_value(&k, &client.get_authed().await, workspace, value, job, db)
.await?;
let as_raw = serde_json::from_value(transformed).map_err(|e| {
error::Error::InternalErr(format!("Error while parsing inner arg: {e}"))
error::Error::InternalErr(format!("Error while parsing inner arg: {e:#}"))
})?;
r.insert(k.to_string(), as_raw);
} else {
@@ -200,13 +200,13 @@ pub async fn transform_json_as_values<'a>(
let inner_vs = v.get();
if (*RE_RES_VAR).is_match(inner_vs) {
let value = serde_json::from_str(inner_vs).map_err(|e| {
error::Error::InternalErr(format!("Error while parsing inner arg: {e}"))
error::Error::InternalErr(format!("Error while parsing inner arg: {e:#}"))
})?;
let transformed =
transform_json_value(&k, &client.get_authed().await, workspace, value, job, db)
.await?;
let as_raw = serde_json::from_value(transformed).map_err(|e| {
error::Error::InternalErr(format!("Error while parsing inner arg: {e}"))
error::Error::InternalErr(format!("Error while parsing inner arg: {e:#}"))
})?;
r.insert(k.to_string(), as_raw);
} else {
@@ -254,7 +254,7 @@ pub async fn transform_json_value(
.await
.map(|x| json!(x))
.map_err(|e| {
Error::NotFound(format!("Variable {path} not found for `{name}`: {e}"))
Error::NotFound(format!("Variable {path} not found for `{name}`: {e:#}"))
})
}
Value::String(y) if y.starts_with("$res:") => {
@@ -271,7 +271,7 @@ pub async fn transform_json_value(
)
.await
.map_err(|e| {
Error::NotFound(format!("Resource {path} not found for `{name}`: {e}"))
Error::NotFound(format!("Resource {path} not found for `{name}`: {e:#}"))
})
}
Value::String(y) if y.starts_with("$") => {
@@ -613,7 +613,7 @@ where
.fetch_optional(&db)
.await
.unwrap_or_else(|e| {
tracing::error!(%e, "error updating job {job_id}: {e}");
tracing::error!(%e, "error updating job {job_id}: {e:#}");
Some((false, None, None, false))
})
.unwrap_or_else(|| {
@@ -886,7 +886,7 @@ pub async fn handle_child(
if let Some(mut file) = File::create(format!("/proc/{pid}/oom_score_adj"))
.await
.map_err(|e| {
tracing::error!("Could not create oom_score_file to pid {pid}: {e}");
tracing::error!("Could not create oom_score_file to pid {pid}: {e:#}");
e
})
.ok()
@@ -1202,7 +1202,7 @@ pub async fn resolve_job_timeout(
.fetch_one(_db)
.await
.map_err(|e| {
tracing::error!(%e, "error getting premium workspace for job {_job_id}: {e}");
tracing::error!(%e, "error getting premium workspace for job {_job_id}: {e:#}");
})
.unwrap_or(false);
#[cfg(not(feature = "cloud"))]
@@ -1558,7 +1558,7 @@ pub async fn save_in_cache(
.execute(db)
.await
{
tracing::error!("Error creating cache resource {e}")
tracing::error!("Error creating cache resource {e:#}")
}
}
backend/windmill-worker/src/dedicated_worker.rs
+2 -2
View File
@@ -110,7 +110,7 @@ pub async fn handle_dedicated_process(
.await
.expect("child process encountered an error");
if let Err(e) = process_status(status) {
tracing::error!("child exit status was not success: {e}");
tracing::error!("child exit status was not success: {e:#}");
} else {
tracing::info!("child exist status was success");
}
@@ -203,7 +203,7 @@ pub async fn handle_dedicated_process(
child
.await
.map_err(|e| anyhow::anyhow!("child process encountered an error: {e}"))?;
.map_err(|e| anyhow::anyhow!("child process encountered an error: {e:#}"))?;
tracing::info!("dedicated worker child process exited successfully");
Ok(())
}
backend/windmill-worker/src/mysql_executor.rs
+1 -1
View File
@@ -59,7 +59,7 @@ pub async fn do_mysql(
.await?;
let as_raw = serde_json::from_value(val)
.map_err(|e| Error::InternalErr(format!("Error while parsing inline resource: {e}")))?;
.map_err(|e| Error::InternalErr(format!("Error while parsing inline resource: {e:#}")))?;
Some(as_raw)
} else {
backend/windmill-worker/src/pg_executor.rs
+1 -1
View File
@@ -132,7 +132,7 @@ pub async fn do_postgresql(
if !root_certificate_pem.is_empty() {
connector.add_root_certificate(
Certificate::from_pem(root_certificate_pem.as_bytes())
.map_err(|e| error::Error::BadConfig(format!("Invalid Certs: {e}")))?,
.map_err(|e| error::Error::BadConfig(format!("Invalid Certs: {e:#}")))?,
);
} else {
connector.danger_accept_invalid_certs(true);
backend/windmill-worker/src/worker.rs
+5 -5
View File
@@ -1242,7 +1242,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
)
.await
{
tracing::error!("Error updating flow status after job completion for {flow} on {worker_name2}: {e}");
tracing::error!("Error updating flow status after job completion for {flow} on {worker_name2}: {e:#}");
}
}
SendResult::Kill => {
@@ -1375,7 +1375,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
queue_init_bash_maybe(db, same_worker_tx.clone(), &worker_name, rsmq.clone()).await
{
killpill_tx.send(()).unwrap_or_default();
tracing::error!("Error queuing init bash script for worker {worker_name}: {e}");
tracing::error!("Error queuing init bash script for worker {worker_name}: {e:#}");
return;
}
}
@@ -2111,7 +2111,7 @@ async fn spawn_dedicated_worker(
.bind(&w_id)
.fetch_optional(&db)
.await
.map_err(|e| Error::InternalErr(format!("expected content and lock: {e}")))
.map_err(|e| Error::InternalErr(format!("expected content and lock: {e:#}")))
.map(|x| x.map(|y| (y.0, y.1, y.2, y.3, if y.4 { y.5.map(|z| z.to_string()) } else { None })))
};
if let Ok(q) = q {
@@ -2492,7 +2492,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
let (flow, job_status_to_update) = if let Some(parent_job_id) = job.parent_job {
if let Err(e) = update_job_future().await {
tracing::error!(
"error updating job future for job {} for handle_job_error: {e}",
"error updating job future for job {} for handle_job_error: {e:#}",
job.id
);
}
@@ -2727,7 +2727,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("fetching step flow status: {e}")))?
.map_err(|e| Error::InternalErr(format!("fetching step flow status: {e:#}")))?
.ok_or_else(|| Error::InternalErr(format!("Expected script_path")))?;
let step = step.unwrap_or(-1);
Some(format!(
backend/windmill-worker/src/worker_flow.rs
+35 -35
View File
@@ -117,7 +117,7 @@ pub async fn update_flow_status_after_job_completion<
{
Ok(j) => j,
Err(e) => {
tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e}", nrec.flow,&nrec.job_id_for_status);
tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", nrec.flow,&nrec.job_id_for_status);
update_flow_status_after_job_completion_internal(
db,
client,
@@ -203,7 +203,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.await
.map_err(|e| {
Error::InternalErr(format!(
"fetching flow status {flow} while reporting {success} {result:?}: {e}"
"fetching flow status {flow} while reporting {success} {result:?}: {e:#}"
))
})?;
@@ -269,7 +269,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.bind(flow)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e}")))?;
.map_err(|e| Error::InternalErr(format!("retrieval of stop_early_expr from state: {e:#}")))?;
let r = SkipIfStopped::from_row(&row)?;
let stop_early = success
@@ -331,7 +331,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.fetch_one(&mut tx)
.await.map_err(|e| {
Error::InternalErr(format!(
"error while fetching iterator index: {e}"
"error while fetching iterator index: {e:#}"
))
})?
.ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?;
@@ -355,7 +355,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.await
.map_err(|e| {
Error::InternalErr(format!(
"error while fetching branchall index: {e}"
"error while fetching branchall index: {e:#}"
))
})?
.ok_or_else(|| Error::InternalErr(format!("requiring an index in InProgress")))?;
@@ -375,7 +375,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.await
.map_err(|e| {
Error::InternalErr(format!(
"error while fetching sucess from completed_jobs: {e}"
"error while fetching sucess from completed_jobs: {e:#}"
))
})?
.into_iter()
@@ -405,7 +405,7 @@ pub async fn update_flow_status_after_job_completion_internal<
flow,
).fetch_optional(db).await.map_err(|e| {
Error::InternalErr(format!(
"error while deleting parallel_monitor_lock: {e}"
"error while deleting parallel_monitor_lock: {e:#}"
))
})?;
if r.is_some() {
@@ -431,7 +431,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.fetch_all(db)
.await
.map_err(|e| {
Error::InternalErr(format!("error while locking jobs to decrease parallelism of: {e}"))
Error::InternalErr(format!("error while locking jobs to decrease parallelism of: {e:#}"))
})?;
for id in ids {
sqlx::query!(
@@ -442,7 +442,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.await
.map_err(|e| {
Error::InternalErr(format!(
"error decreasing suspend for {id}: {e}"
"error decreasing suspend for {id}: {e:#}"
))
})?;
}
@@ -457,7 +457,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.execute(db)
.await
.map_err(|e| {
Error::InternalErr(format!("error while setting last ping to null: {e}"))
Error::InternalErr(format!("error while setting last ping to null: {e:#}"))
})?;
let r = sqlx::query_scalar!(
@@ -465,7 +465,7 @@ pub async fn update_flow_status_after_job_completion_internal<
flow,
job_id_for_status
).fetch_optional(db).await.map_err(|e| {
Error::InternalErr(format!("error while removing parallel_monitor_lock: {e}"))
Error::InternalErr(format!("error while removing parallel_monitor_lock: {e:#}"))
})?;
if r.is_some() {
tracing::info!(
@@ -536,7 +536,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.fetch_optional(&mut tx)
.await
.map_err(|e| {
Error::InternalErr(format!("error while getting retry from step: {e}"))
Error::InternalErr(format!("error while getting retry from step: {e:#}"))
})?
.flatten();
@@ -574,7 +574,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.execute(&mut tx)
.await
.map_err(|e| {
Error::InternalErr(format!("error while setting flow index for {flow}: {e}"))
Error::InternalErr(format!("error while setting flow index for {flow}: {e:#}"))
})?;
old_status.step + 1
} else {
@@ -595,7 +595,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.fetch_one(&mut tx)
.await.map_err(|e| {
Error::InternalErr(format!(
"error while fetching failure module: {e}"
"error while fetching failure module: {e:#}"
))
})?;
@@ -613,7 +613,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.await
.map_err(|e| {
Error::InternalErr(format!(
"error while setting flow status in failure step: {e}"
"error while setting flow status in failure step: {e:#}"
))
})?;
} else {
@@ -628,7 +628,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.execute(&mut tx)
.await
.map_err(|e| {
Error::InternalErr(format!("error while setting new flow status: {e}"))
Error::InternalErr(format!("error while setting new flow status: {e:#}"))
})?;
if let Some(job_result) = new_status.job_result() {
@@ -643,7 +643,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.execute(&mut tx)
.await.map_err(|e| {
Error::InternalErr(format!(
"error while setting leaf jobs: {e}"
"error while setting leaf jobs: {e:#}"
))
})?;
}
@@ -759,7 +759,7 @@ pub async fn update_flow_status_after_job_completion_internal<
.execute(db)
.await
.map_err(|e| {
Error::InternalErr(format!("error while cleaning up completed_job: {e}"))
Error::InternalErr(format!("error while cleaning up completed_job: {e:#}"))
})?;
}
}
@@ -821,7 +821,7 @@ pub async fn update_flow_status_after_job_completion_internal<
stop_early && skip_if_stop_early,
Json(
&serde_json::from_str::<Value>(nresult.get()).unwrap_or_else(
|e| json!({"error": format!("Impossible to serialize error: {e}")}),
|e| json!({"error": format!("Impossible to serialize error: {e:#}")}),
),
),
0,
@@ -966,7 +966,7 @@ async fn compute_skip_loop_failures_and_parallelism(
.fetch_one(db)
.await
.map(|(v, n)| (v,n))
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}")))
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e:#}")))
}
async fn compute_skip_branchall_failure<'c>(
@@ -986,7 +986,7 @@ async fn compute_skip_branchall_failure<'c>(
.fetch_one(db)
.await
.map(|(v,)| v)
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e}")))
.map_err(|e| Error::InternalErr(format!("error during retrieval of skip_loop_failures: {e:#}")))
}
async fn has_failure_module<'c>(
@@ -1001,7 +1001,7 @@ async fn has_failure_module<'c>(
.bind(flow)
.fetch_one(&mut **tx)
.await
.map_err(|e| Error::InternalErr(format!("error during retrieval of has_failure_module: {e}")))
.map_err(|e| Error::InternalErr(format!("error during retrieval of has_failure_module: {e:#}")))
.map(|v| v.unwrap_or(false))
}
@@ -1015,7 +1015,7 @@ async fn has_failure_module<'c>(
// )
// .fetch_one(db)
// .await
// .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e}")))?;
// .map_err(|e| Error::InternalErr(format!("error during retrieval of cleanup module: {e:#}")))?;
// raw_value
// .clone()
@@ -1117,7 +1117,7 @@ pub async fn get_step_of_flow_status(db: &DB, id: Uuid) -> error::Result<Step> {
)
.fetch_one(db)
.await
.map_err(|e| Error::InternalErr(format!("fetching step flow status: {e}")))?;
.map_err(|e| Error::InternalErr(format!("fetching step flow status: {e:#}")))?;
if r.step < r.len {
Ok(Step::Step(r.step.ok_or_else(|| {
Error::InternalErr("step is null".to_string())
@@ -1176,7 +1176,7 @@ async fn transform_input(
.await
.map_err(|e| {
Error::ExecutionErr(format!(
"Error during isolated evaluation of expression `{expr}`:\n{e}"
"Error during isolated evaluation of expression `{expr}`:\n{e:#}"
))
})?;
mapped.insert(key.to_string(), v);
@@ -1359,7 +1359,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
"error sending update flow message to job completed channel: {e:#}"
))
})?;
@@ -1406,7 +1406,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
"error sending update flow message to job completed channel: {e:#}"
))
})?;
@@ -1438,7 +1438,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
"error sending update flow message to job completed channel: {e:#}"
))
})?;
@@ -1547,7 +1547,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::ExecutionErr(format!(
"Error during isolated evaluation of expression `{expr}`:\n{e}"
"Error during isolated evaluation of expression `{expr}`:\n{e:#}"
))
})?
.get(),
@@ -1557,7 +1557,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
} else {
let e = eval_result.err().unwrap();
return Err(Error::ExecutionErr(format!(
"Result returned by input transform invalid `{e}`"
"Result returned by input transform invalid `{e:#}`"
)));
}
}
@@ -1686,7 +1686,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
"error sending update flow message to job completed channel: {e:#}"
))
})?;
@@ -1742,7 +1742,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await
.map_err(|e| {
Error::ExecutionErr(format!(
"Error during isolated evaluation of expression `{expr}`:\n{e}"
"Error during isolated evaluation of expression `{expr}`:\n{e:#}"
))
})?
.get(),
@@ -1881,7 +1881,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
if let Some(row) = row {
RawArgs::from_row(&row)
.map(|x| x.args.map(|x| x.0).unwrap_or_else(HashMap::new))
.map_err(|e| error::Error::InternalErr(format!("Impossible to build args: {e}")))
.map_err(|e| error::Error::InternalErr(format!("Impossible to build args: {e:#}")))
} else {
Ok(HashMap::new())
}
@@ -1913,7 +1913,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
})
.unwrap(),
)
.map_err(|e| error::Error::InternalErr(format!("identity: {e}"))),
.map_err(|e| error::Error::InternalErr(format!("identity: {e:#}"))),
Ok(_) => Ok(flow_job_args),
Err(e) => {
@@ -2152,7 +2152,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
if payload_tag.delete_after_use {
let uuid_singleton_json = serde_json::to_value(&[uuid])
.map_err(|e| error::Error::InternalErr(format!("Unable to serialize uuid: {e}")))?;
.map_err(|e| error::Error::InternalErr(format!("Unable to serialize uuid: {e:#}")))?;
sqlx::query(
"UPDATE queue