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

More Metrics (#735)

Browse Source 
* Expose Worker Tokio Metrics

* Add metrics tracking the job queue

This can be used to (approximate) the queue length
This can be used to (approximate) the running jobs

* Remove testing code

* Rename metrics to exclude TOTAL_

* Sleep in worker metrics loop

* Add jobs_executed

* Add zombie job metrics

* Add uptime metric

* Fix metric naming

* Remove poll stats

* unify worker execution metrics

* Rename jobs_* metrics

* Rename metrics to match exposed name

* Remove leftover import

* Rename variables to match further

* Fix Merge error

Co-authored-by: Ruben Fiszel <ruben@rubenfiszel.com>
This commit is contained in:
Kai Jellinghaus
2022-10-19 10:15:42 +02:00
committed by GitHub
co-authored by Ruben Fiszel
parent 2350a39335
commit 55d3f38b8b
5 changed files with 152 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+13
View File
@@ -3546,6 +3546,17 @@ dependencies = [
"syn",
]
[[package]]
name = "tokio-metrics"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bcb585a0069b53171684e22d5255984ec30d1c7304fd0a4a9a603ffd8c765cdd"
dependencies = [
"futures-util",
"pin-project-lite",
"tokio",
]
[[package]]
name = "tokio-native-tls"
version = "0.3.0"
@@ -4255,6 +4266,7 @@ dependencies = [
"hyper",
"itertools 0.10.5",
"json-pointer",
"lazy_static",
"lettre",
"magic-crypt",
"mime_guess",
@@ -4280,6 +4292,7 @@ dependencies = [
"thiserror",
"time 0.3.15",
"tokio",
"tokio-metrics",
"tokio-tar",
"tokio-util",
"tower",
backend/Cargo.toml
+2
View File
@@ -66,3 +66,5 @@ sqlx = { version = "^0", features = ["offline", "macros", "migrate", "uuid", "js
dotenv = "^0"
ulid = { version = "^1", features = ["uuid"] }
futures = "^0"
tokio-metrics = "0.1.0"
lazy_static = "1.4.0"
backend/src/jobs.rs
+28 -1
View File
@@ -1315,6 +1315,25 @@ pub enum JobPayload {
RawFlow { value: FlowValue, path: Option<String> },
}
lazy_static::lazy_static! {
// TODO: these aren't synced, they should be moved into the queue abstraction once/if that happens.
static ref QUEUE_PUSH_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_push_count",
"Total number of jobs pushed to the queue."
)
.unwrap();
static ref QUEUE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_delete_count",
"Total number of jobs deleted from the queue."
)
.unwrap();
static ref QUEUE_PULL_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_pull_count",
"Total number of jobs pulled from the queue."
)
.unwrap();
}
#[instrument(level = "trace", skip_all)]
pub async fn push<'c>(
mut tx: Transaction<'c, Postgres>,
@@ -1533,6 +1552,8 @@ pub async fn push<'c>(
.fetch_one(&mut tx)
.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();
{
let uuid_string = job_id.to_string();
@@ -1575,7 +1596,7 @@ pub async fn add_completed_job_error<E: ToString + std::fmt::Debug>(
e: E,
metrics: Option<worker::Metrics>,
) -> Result<(Uuid, Map<String, Value>), Error> {
metrics.map(|m| m.jobs_failed.inc());
metrics.map(|m| m.worker_execution_failed.inc());
let mut output_map = serde_json::Map::new();
output_map.insert(
"error".to_string(),
@@ -1734,11 +1755,17 @@ pub async fn pull(db: &DB) -> Result<Option<QueuedJob>, crate::Error> {
)
.fetch_optional(db)
.await?;
if job.is_some() {
QUEUE_PULL_COUNT.inc();
}
Ok(job)
}
#[instrument(level = "trace", skip_all)]
pub async fn delete_job(db: &DB, w_id: &str, job_id: Uuid) -> Result<(), crate::Error> {
QUEUE_DELETE_COUNT.inc();
let job_removed = sqlx::query_scalar!(
"DELETE FROM queue WHERE workspace_id = $1 AND id = $2 RETURNING 1",
w_id,
backend/src/lib.rs
+5 -3
View File
@@ -219,13 +219,14 @@ pub async fn run_workers(
rx: tokio::sync::broadcast::Receiver<()>,
) -> anyhow::Result<()> {
let instance_name = rd_string(5);
let monitor = tokio_metrics::TaskMonitor::new();
let ip = external_ip::get_ip().await.unwrap_or_else(|e| {
tracing::warn!(error = e.to_string(), "failed to get external IP");
"unretrievable IP".to_string()
});
let mut handles = Vec::new();
let mut handles = Vec::with_capacity(num_workers as usize);
for i in 1..(num_workers + 1) {
let db1 = db.clone();
@@ -234,7 +235,7 @@ pub async fn run_workers(
let ip = ip.clone();
let rx = rx.resubscribe();
let worker_config = worker_config.clone();
handles.push(tokio::spawn(async move {
handles.push(tokio::spawn(monitor.instrument(async move {
tracing::info!(addr = %addr.to_string(), worker = %worker_name, "starting worker");
worker::run_worker(
&db1,
@@ -249,8 +250,9 @@ pub async fn run_workers(
rx,
)
.await
}));
})));
}
futures::future::try_join_all(handles).await?;
Ok(())
}
backend/src/worker.rs
+104 -60
View File
@@ -86,7 +86,7 @@ const GO_REQ_SPLITTER: &str = "//go.sum";
#[derive(Clone)]
pub struct Metrics {
pub jobs_failed: prometheus::IntCounter,
pub worker_execution_failed: prometheus::IntCounter,
}
#[derive(Clone)]
@@ -98,6 +98,28 @@ pub struct WorkerConfig {
pub keep_job_dir: bool,
}
lazy_static::lazy_static! {
static ref WORKER_STARTED: prometheus::IntGauge = prometheus::register_int_gauge!(
"worker_started",
"Total number of workers started."
)
.unwrap();
static ref QUEUE_ZOMBIE_RESTART_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_zombie_restart_count",
"Total number of jobs restarted due to ping timeout."
)
.unwrap();
static ref QUEUE_ZOMBIE_DELETE_COUNT: prometheus::IntCounter = prometheus::register_int_counter!(
"queue_zombie_delete_count",
"Total number of jobs deleted due to their ping timing out in an unrecoverable state."
)
.unwrap();
static ref WORKER_UPTIME_OPTS: prometheus::Opts = prometheus::opts!(
"worker_uptime",
"Total number of milliseconds since the worker has started"
);
}
pub async fn run_worker(
db: &DB,
timeout: i32,
@@ -110,6 +132,8 @@ pub async fn run_worker(
worker_config: WorkerConfig,
mut rx: tokio::sync::broadcast::Receiver<()>,
) {
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");
@@ -138,23 +162,19 @@ pub async fn run_worker(
insert_initial_ping(worker_instance, &worker_name, ip, db).await;
prometheus::register_int_gauge!(prometheus::Opts::new(
"start_time_seconds",
"Start time of worker as seconds since unix epoch",
)
.const_label("name", &worker_name))
.expect("register prometheus metric")
.set(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs() as i64)
.unwrap_or(0),
let uptime_metric = prometheus::register_int_counter!(WORKER_UPTIME_OPTS
.clone()
.const_label("name", &worker_name))
.unwrap();
uptime_metric.inc_by(
((Instant::now() - start_time).as_millis() - uptime_metric.get() as u128)
.try_into()
.unwrap(),
);
let job_duration_seconds = prometheus::register_histogram_vec!(
let worker_execution_duration = prometheus::register_histogram_vec!(
prometheus::HistogramOpts::new(
"job_duration_seconds",
"worker_execution_duration",
"Duration between receiving a job and completing it",
)
.const_label("name", &worker_name),
@@ -162,8 +182,15 @@ pub async fn run_worker(
)
.expect("register prometheus metric");
let jobs_failed = prometheus::register_int_counter_vec!(
prometheus::Opts::new("jobs_failed", "Number of failed jobs",)
let worker_execution_failed = prometheus::register_int_counter_vec!(
prometheus::Opts::new("worker_execution_failed", "Number of failed jobs",)
.const_label("name", &worker_name),
&["workspace_id", "language"],
)
.expect("register prometheus metric");
let worker_execution_count = prometheus::register_int_counter_vec!(
prometheus::Opts::new("worker_execution_count", "Number of executed jobs",)
.const_label("name", &worker_name),
&["workspace_id", "language"],
)
@@ -198,10 +225,16 @@ pub async fn run_worker(
pip_extra_index_url,
pip_trusted_host,
};
WORKER_STARTED.inc();
let (same_worker_tx, mut same_worker_rx) = mpsc::channel::<Uuid>(5);
loop {
uptime_metric.inc_by(
((Instant::now() - start_time).as_millis() - uptime_metric.get() as u128)
.try_into()
.unwrap(),
);
if last_ping.elapsed().as_secs() > NUM_SECS_ENV_CHECK {
sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1 WHERE worker = $2",
@@ -238,14 +271,19 @@ pub async fn run_worker(
job.language.as_ref().map(|l| l.as_str()).unwrap_or(""),
];
let _timer = job_duration_seconds
let _timer = worker_execution_duration
.with_label_values(label_values.as_slice())
.start_timer();
jobs_executed += 1;
worker_execution_count
.with_label_values(label_values.as_slice())
.inc();
let metrics =
Metrics { jobs_failed: jobs_failed.with_label_values(label_values.as_slice()) };
let metrics = Metrics {
worker_execution_failed: worker_execution_failed
.with_label_values(label_values.as_slice()),
};
tracing::info!(worker = %worker_name, id = %job.id, "fetched job {}", job.id);
@@ -1885,7 +1923,20 @@ pub async fn handle_zombie_jobs_periodically(
mut rx: tokio::sync::broadcast::Receiver<()>,
) {
loop {
let restarted = sqlx::query!(
handle_zombie_jobs(db, timeout).await;
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(60)) => (),
_ = rx.recv() => {
println!("received killpill for monitor job");
break;
}
}
}
}
async fn handle_zombie_jobs(db: &DB, timeout: i32) {
let restarted = sqlx::query!(
"UPDATE queue SET running = false WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = false RETURNING id, workspace_id, last_ping",
(timeout * 5).to_string(),
JobKind::Flow: JobKind,
@@ -1895,16 +1946,17 @@ pub async fn handle_zombie_jobs_periodically(
.ok()
.unwrap_or_else(|| vec![]);
for r in restarted {
tracing::info!(
"restarted zombie job {} {} {}",
r.id,
r.workspace_id,
r.last_ping
);
}
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
for r in restarted {
tracing::info!(
"restarted zombie job {} {} {}",
r.id,
r.workspace_id,
r.last_ping
);
}
let timeouts = sqlx::query_as::<_, QueuedJob>(
let timeouts = sqlx::query_as::<_, QueuedJob>(
"SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND same_worker = true",
)
.bind((timeout * 5).to_string())
@@ -1914,38 +1966,30 @@ pub async fn handle_zombie_jobs_periodically(
.ok()
.unwrap_or_else(|| vec![]);
for job in timeouts {
tracing::info!(
"timedouts zombie same_worker job {} {}",
job.id,
job.workspace_id,
);
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
for job in timeouts {
tracing::info!(
"timedouts zombie same_worker job {} {}",
job.id,
job.workspace_id,
);
// since the job is unrecoverable, the same worker queue should never be sent anything
let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::<Uuid>(1);
// since the job is unrecoverable, the same worker queue should never be sent anything
let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::<Uuid>(1);
let _ = handle_job_error(
db,
job,
error::Error::ExecutionErr("Same worker job timed out".to_string()),
None,
true,
same_worker_tx_never_used,
"",
true,
&std::env::var("BASE_INTERNAL_URL")
.unwrap_or_else(|_| "http://localhost:8000".to_string()),
)
.await;
}
tokio::select! {
_ = tokio::time::sleep(Duration::from_secs(60)) => (),
_ = rx.recv() => {
println!("received killpill for monitor job");
break;
}
}
let _ = handle_job_error(
db,
job,
error::Error::ExecutionErr("Same worker job timed out".to_string()),
None,
true,
same_worker_tx_never_used,
"",
true,
&std::env::var("BASE_INTERNAL_URL")
.unwrap_or_else(|_| "http://localhost:8000".to_string()),
)
.await;
}
}