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

only register prometheus metrics if they are enabled

Browse Source
This commit is contained in:
Ruben Fiszel
2023-04-09 15:39:25 +02:00
parent ff22759a43
commit bbb6a15da7
6 changed files with 66 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+2 -10
View File
@@ -10,7 +10,7 @@ use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use git_version::git_version;
use sqlx::{Pool, Postgres};
use windmill_common::utils::rd_string;
use windmill_common::{utils::rd_string, METRICS_ADDR};
const GIT_VERSION: &str = git_version!(args = ["--tag", "--always"], fallback = "unknown-version");
const DEFAULT_NUM_WORKERS: usize = 3;
@@ -30,15 +30,7 @@ async fn main() -> anyhow::Result<()> {
.and_then(|x| x.parse::<i32>().ok())
.unwrap_or(DEFAULT_NUM_WORKERS as i32);
let metrics_addr: Option<SocketAddr> = std::env::var("METRICS_ADDR")
.ok()
.map(|s| {
s.parse::<bool>()
.map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001))))
.or_else(|_| s.parse::<SocketAddr>().map(Some))
})
.transpose()?
.flatten();
let metrics_addr: Option<SocketAddr> = *METRICS_ADDR;
let server_bind_address: IpAddr = std::env::var("SERVER_BIND_ADDR")
.ok()
backend/windmill-api/src/webhook_util.rs
+3 -2
View File
@@ -2,6 +2,7 @@ use std::time::Duration;
use serde::Serialize;
use tokio::{select, sync::mpsc, time::interval};
use windmill_common::METRICS_ENABLED;
use crate::db::DB;
@@ -89,9 +90,9 @@ impl WebhookShared {
};
let webook_opt = url_guard.value();
if let Some(url) = webook_opt {
let timer = WEBHOOK_REQUEST_COUNT.start_timer();
let timer = if *METRICS_ENABLED { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
let _ = client.post(url).json(&message).send().await;
timer.stop_and_record();
timer.map(|x| x.stop_and_record());
drop(url_guard);
}
},
backend/windmill-common/src/lib.rs
+11
View File
@@ -30,6 +30,17 @@ pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 3;
lazy_static::lazy_static! {
pub static ref METRICS_ADDR: Option<SocketAddr> = std::env::var("METRICS_ADDR")
.ok()
.map(|s| {
s.parse::<bool>()
.map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001))))
.or_else(|_| s.parse::<SocketAddr>().map(Some))
})
.transpose().ok()
.flatten()
.flatten();
pub static ref METRICS_ENABLED: bool = METRICS_ADDR.is_some();
pub static ref BASE_URL: String = std::env::var("BASE_URL").unwrap_or_else(|_| "http://localhost".to_string());
pub static ref IS_READY: Arc<std::sync::atomic::AtomicBool> = Arc::new(std::sync::atomic::AtomicBool::new(false));
}
backend/windmill-queue/src/jobs.rs
+8 -3
View File
@@ -22,6 +22,7 @@ use windmill_common::{
flows::{FlowModule, FlowModuleValue, FlowValue},
scripts::{get_full_hub_script_by_path, HubScript, ScriptHash, ScriptLang},
utils::StripPath,
METRICS_ENABLED,
};
lazy_static::lazy_static! {
@@ -146,7 +147,7 @@ pub async fn pull(
.fetch_optional(db)
.await?;
if job.is_some() {
if job.is_some() && *METRICS_ENABLED {
QUEUE_PULL_COUNT.inc();
}
@@ -211,7 +212,9 @@ pub async fn delete_job(
w_id: &str,
job_id: Uuid,
) -> windmill_common::error::Result<()> {
QUEUE_DELETE_COUNT.inc();
if *METRICS_ENABLED {
QUEUE_DELETE_COUNT.inc();
}
let job_removed = sqlx::query_scalar!(
"DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
w_id,
@@ -542,7 +545,9 @@ pub async fn push<'c>(
.await
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_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.
QUEUE_PUSH_COUNT.inc();
if *METRICS_ENABLED {
QUEUE_PUSH_COUNT.inc();
}
{
let uuid_string = job_id.to_string();
backend/windmill-worker/src/jobs.rs
+6 -2
View File
@@ -9,7 +9,9 @@
use sqlx::{Pool, Postgres, Transaction};
use tracing::instrument;
use uuid::Uuid;
use windmill_common::{error::Error, flow_status::FlowStatusModule, schedule::Schedule};
use windmill_common::{
error::Error, flow_status::FlowStatusModule, schedule::Schedule, METRICS_ENABLED,
};
use windmill_queue::{delete_job, schedule::get_schedule_opt, JobKind, QueuedJob, CLOUD_HOSTED};
#[instrument(level = "trace", skip_all)]
@@ -20,7 +22,9 @@ pub async fn add_completed_job_error(
e: serde_json::Value,
metrics: Option<crate::worker::Metrics>,
) -> Result<serde_json::Value, Error> {
metrics.map(|m| m.worker_execution_failed.inc());
if *METRICS_ENABLED {
metrics.map(|m| m.worker_execution_failed.inc());
}
let result = serde_json::json!({ "error": e });
let _ = add_completed_job(db, &queued_job, false, false, result.clone(), logs).await?;
Ok(result)
backend/windmill-worker/src/worker.rs
+36 -25
View File
@@ -27,7 +27,7 @@ use windmill_common::{
flows::{FlowModuleValue, FlowValue},
scripts::{ScriptHash, ScriptLang},
utils::{rd_string, calculate_hash},
variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY,
variables, BASE_URL, users::SUPERADMIN_SECRET_EMAIL, IS_READY, METRICS_ENABLED,
};
use windmill_queue::{canceled_job_to_result, get_queued_job, pull, JobKind, QueuedJob, CLOUD_HOSTED};
@@ -610,8 +610,9 @@ pub async fn run_worker(
let mut jobs_executed = 0;
WORKER_STARTED.inc();
if *METRICS_ENABLED {
WORKER_STARTED.inc();
}
let (_copy_bucket_tx, mut _copy_bucket_rx) = mpsc::channel::<()>(2);
@@ -644,13 +645,14 @@ pub async fn run_worker(
loop {
worker_busy.set(0);
uptime_metric.inc_by(
(((Instant::now() - start_time).as_millis() as f64)/1000.0 - uptime_metric.get())
.try_into()
.unwrap(),
);
if *METRICS_ENABLED {
worker_busy.set(0);
uptime_metric.inc_by(
(((Instant::now() - start_time).as_millis() as f64)/1000.0 - uptime_metric.get())
.try_into()
.unwrap(),
);
}
let do_break = async {
@@ -707,10 +709,12 @@ pub async fn run_worker(
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())))
},
(job, timer) = {
let timer = worker_pull_duration.start_timer();
let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None };
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer)) } => {
let duration_pull_s = timer.stop_and_record();
worker_pull_duration_counter.inc_by(duration_pull_s);
timer.map(|timer| {
let duration_pull_s = timer.stop_and_record();
worker_pull_duration_counter.inc_by(duration_pull_s);
});
(false, job)
},
}
@@ -718,8 +722,9 @@ pub async fn run_worker(
if do_break {
return true;
}
worker_busy.set(1);
if *METRICS_ENABLED {
worker_busy.set(1);
}
match next_job {
Ok(Some(job)) => {
let label_values = [
@@ -732,9 +737,11 @@ pub async fn run_worker(
.start_timer();
jobs_executed += 1;
worker_execution_count
.with_label_values(label_values.as_slice())
.inc();
if *METRICS_ENABLED {
worker_execution_count
.with_label_values(label_values.as_slice())
.inc();
}
let metrics = Metrics {
worker_execution_failed: worker_execution_failed
@@ -825,12 +832,12 @@ pub async fn run_worker(
}
}
Ok(None) => {
let _timer = worker_sleep_duration
.start_timer();
let _timer = if *METRICS_ENABLED { Some(worker_sleep_duration.start_timer()) } else { None };
tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await;
let duration = _timer.stop_and_record();
worker_sleep_duration_counter.inc_by(duration);
_timer.map(|timer| {
let duration = timer.stop_and_record();
worker_sleep_duration_counter.inc_by(duration);
});
}
Err(err) => {
tracing::error!(worker = %worker_name, "run_worker: pulling jobs: {}", err);
@@ -2915,7 +2922,9 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str) {
.ok()
.unwrap_or_else(|| vec![]);
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
if *METRICS_ENABLED {
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
}
for r in restarted {
tracing::info!(
"restarted zombie job {} {} {}",
@@ -2940,7 +2949,9 @@ async fn handle_zombie_jobs(db: &Pool<Postgres>, base_internal_url: &str) {
.ok()
.unwrap_or_else(|| vec![]);
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
if *METRICS_ENABLED {
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
}
for job in timeouts {
tracing::info!(
"timedout zombie job {} {}",