diff --git a/backend/.sqlx/query-0af0e0a1dddeee2021ba060e390e1b60caa3752669636e9fb0817a68121a9451.json b/backend/.sqlx/query-0af0e0a1dddeee2021ba060e390e1b60caa3752669636e9fb0817a68121a9451.json new file mode 100644 index 0000000000..afc74d8e7a --- /dev/null +++ b/backend/.sqlx/query-0af0e0a1dddeee2021ba060e390e1b60caa3752669636e9fb0817a68121a9451.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "0af0e0a1dddeee2021ba060e390e1b60caa3752669636e9fb0817a68121a9451" +} diff --git a/backend/.sqlx/query-1db82007445ff5f644bb607aa28f5747cb50d193475fff5fcfdde37d1bc74636.json b/backend/.sqlx/query-1db82007445ff5f644bb607aa28f5747cb50d193475fff5fcfdde37d1bc74636.json deleted file mode 100644 index aaa1ae945b..0000000000 --- a/backend/.sqlx/query-1db82007445ff5f644bb607aa28f5747cb50d193475fff5fcfdde37d1bc74636.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Text", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "1db82007445ff5f644bb607aa28f5747cb50d193475fff5fcfdde37d1bc74636" -} diff --git a/backend/.sqlx/query-a837494a58ab58bfa18c0385350a861076a5fc4f7eefceb1c0de5cf55293f327.json b/backend/.sqlx/query-a837494a58ab58bfa18c0385350a861076a5fc4f7eefceb1c0de5cf55293f327.json new file mode 100644 index 0000000000..dc77dbc402 --- /dev/null +++ b/backend/.sqlx/query-a837494a58ab58bfa18c0385350a861076a5fc4f7eefceb1c0de5cf55293f327.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Uuid", + "Text", + "Float4" + ] + }, + "nullable": [] + }, + "hash": "a837494a58ab58bfa18c0385350a861076a5fc4f7eefceb1c0de5cf55293f327" +} diff --git a/backend/.sqlx/query-d44c37882150532383d1058639f4d3af288a346c50f29809dbc55a34088a0abc.json b/backend/.sqlx/query-d44c37882150532383d1058639f4d3af288a346c50f29809dbc55a34088a0abc.json deleted file mode 100644 index c071272ff9..0000000000 --- a/backend/.sqlx/query-d44c37882150532383d1058639f4d3af288a346c50f29809dbc55a34088a0abc.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Text", - "Uuid", - "Text", - "Float4" - ] - }, - "nullable": [] - }, - "hash": "d44c37882150532383d1058639f4d3af288a346c50f29809dbc55a34088a0abc" -} diff --git a/backend/migrations/20260311000000_optimize_job_stats_timestamps.down.sql b/backend/migrations/20260311000000_optimize_job_stats_timestamps.down.sql new file mode 100644 index 0000000000..e526b89b11 --- /dev/null +++ b/backend/migrations/20260311000000_optimize_job_stats_timestamps.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE job_stats DROP COLUMN IF EXISTS timeseries_start; +ALTER TABLE job_stats DROP COLUMN IF EXISTS offsets_cs; diff --git a/backend/migrations/20260311000000_optimize_job_stats_timestamps.up.sql b/backend/migrations/20260311000000_optimize_job_stats_timestamps.up.sql new file mode 100644 index 0000000000..f851c7887e --- /dev/null +++ b/backend/migrations/20260311000000_optimize_job_stats_timestamps.up.sql @@ -0,0 +1,5 @@ +-- Store timeseries timestamps as a start time + integer centisecond offsets +-- instead of full TIMESTAMPTZ[] arrays. Saves ~4 bytes per data point. +-- i32 centiseconds gives ~248 days of range with 10ms precision. +ALTER TABLE job_stats ADD COLUMN IF NOT EXISTS timeseries_start TIMESTAMPTZ; +ALTER TABLE job_stats ADD COLUMN IF NOT EXISTS offsets_cs INTEGER[]; diff --git a/backend/windmill-api-jobs/src/job_metrics.rs b/backend/windmill-api-jobs/src/job_metrics.rs index aba8a5f25c..ad316150c0 100644 --- a/backend/windmill-api-jobs/src/job_metrics.rs +++ b/backend/windmill-api-jobs/src/job_metrics.rs @@ -79,7 +79,7 @@ async fn get_job_metrics( >, ) -> error::JsonResult { let records = sqlx::query_as::<_, JobStatsRecord>( - "SELECT * FROM job_stats where workspace_id = $1 and job_id = $2", + "SELECT workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float, timeseries_start, offsets_cs FROM job_stats WHERE workspace_id = $1 AND job_id = $2", ) .bind(w_id) .bind(job_id) @@ -91,7 +91,7 @@ async fn get_job_metrics( let mut timeseries_metrics: Vec = vec![]; for record in records { - let metric_id = record.metric_id; + let metric_id = record.metric_id.clone(); match record.metric_kind { MetricKind::ScalarInt => { let value = record.scalar_int.unwrap_or_default() as f64; @@ -102,47 +102,43 @@ async fn get_job_metrics( scalar_metrics.push(ScalarMetric { metric_id: metric_id.clone(), value }); } MetricKind::TimeseriesInt => { - if record.timestamps.clone().unwrap_or_default().len() - != record.timeseries_int.clone().unwrap_or_default().len() - { - tracing::warn!("Timeseries metric {} has an invalid shape. It doesn't have one timestamp per measurement. (timestamps: {:?}, measurements: {:?})", metric_id, record.timestamps, record.timeseries_int) + let timestamps = resolve_timestamps(&record); + let timeseries_int = record.timeseries_int.unwrap_or_default(); + if timestamps.len() != timeseries_int.len() { + tracing::warn!("Timeseries metric {} has an invalid shape. timestamps: {}, measurements: {}", metric_id, timestamps.len(), timeseries_int.len()); } let (timestamps, timeseries_int) = timeseries_sample( from_timestamp, to_timestamp, timeseries_max_datapoints, - record.timestamps.unwrap_or_default(), - record.timeseries_int.unwrap_or_default(), + timestamps, + timeseries_int, ); - let mut values: Vec = vec![]; - for (idx, value) in timeseries_int.iter().enumerate() { - values.push(DataPoint { - timestamp: timestamps[idx], - value: value.to_owned() as f64, - }); - } + let values: Vec = timestamps + .iter() + .zip(timeseries_int.iter()) + .map(|(ts, v)| DataPoint { timestamp: *ts, value: *v as f64 }) + .collect(); timeseries_metrics.push(TimeseriesMetric { metric_id: metric_id.clone(), values }); } MetricKind::TimeseriesFloat => { - if record.timestamps.clone().unwrap_or_default().len() - != record.timeseries_int.clone().unwrap_or_default().len() - { - tracing::warn!("Timeseries metric {} has an invalid shape. It doesn't have one timestamp per measurement. (timestamps: {:?}, measurements: {:?})", metric_id, record.timestamps, record.timeseries_float) + let timestamps = resolve_timestamps(&record); + let timeseries_float = record.timeseries_float.unwrap_or_default(); + if timestamps.len() != timeseries_float.len() { + tracing::warn!("Timeseries metric {} has an invalid shape. timestamps: {}, measurements: {}", metric_id, timestamps.len(), timeseries_float.len()); } let (timestamps, timeseries_float) = timeseries_sample( from_timestamp, to_timestamp, timeseries_max_datapoints, - record.timestamps.unwrap_or_default(), - record.timeseries_float.unwrap_or_default(), + timestamps, + timeseries_float, ); - let mut values: Vec = vec![]; - for (idx, value) in timeseries_float.iter().enumerate() { - values.push(DataPoint { - timestamp: timestamps[idx], - value: value.to_owned() as f64, - }); - } + let values: Vec = timestamps + .iter() + .zip(timeseries_float.iter()) + .map(|(ts, v)| DataPoint { timestamp: *ts, value: *v as f64 }) + .collect(); timeseries_metrics.push(TimeseriesMetric { metric_id: metric_id.clone(), values }); } }; @@ -152,6 +148,21 @@ async fn get_job_metrics( let response = JobStatsResponse { metrics_metadata, scalar_metrics, timeseries_metrics }; Ok(Json(response)) } + +/// Reconstruct full timestamps from `timeseries_start` + `offsets_cs` if available, +/// otherwise fall back to legacy `timestamps` column. +fn resolve_timestamps(record: &JobStatsRecord) -> Vec> { + if let (Some(start), Some(offsets)) = (record.timeseries_start, &record.offsets_cs) { + if !offsets.is_empty() { + return offsets + .iter() + .map(|&cs| start + chrono::Duration::milliseconds(cs as i64 * 10)) + .collect(); + } + } + // Legacy fallback: use the full timestamps column + record.timestamps.clone().unwrap_or_default() +} #[derive(Deserialize)] struct JobProgressSetRequest { percent: i32, diff --git a/backend/windmill-common/src/job_metrics.rs b/backend/windmill-common/src/job_metrics.rs index d416e4d44b..c49185de72 100644 --- a/backend/windmill-common/src/job_metrics.rs +++ b/backend/windmill-common/src/job_metrics.rs @@ -15,9 +15,11 @@ pub struct JobStatsRecord { pub timestamps: Option>>, pub timeseries_int: Option>, pub timeseries_float: Option>, + pub timeseries_start: Option>, + pub offsets_cs: Option>, } -#[derive(sqlx::Type, Debug, PartialEq, Deserialize, Serialize)] +#[derive(sqlx::Type, Debug, Clone, PartialEq, Deserialize, Serialize)] #[sqlx(type_name = "METRIC_KIND", rename_all = "snake_case")] pub enum MetricKind { ScalarInt, @@ -52,29 +54,21 @@ pub async fn register_metric_for_job( return Ok(metric_id); } - let (scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float) = match metric_kind - { + let is_timeseries = matches!( + metric_kind, + MetricKind::TimeseriesInt | MetricKind::TimeseriesFloat + ); + + let (scalar_int, scalar_float, timeseries_int, timeseries_float) = match metric_kind { MetricKind::ScalarInt | MetricKind::ScalarFloat => { - (None as Option, None as Option, None, None, None) + (None as Option, None as Option, None, None) } - MetricKind::TimeseriesInt => ( - None, - None, - Some(&[] as &[chrono::DateTime]), - Some(&[] as &[i32]), - None, - ), - MetricKind::TimeseriesFloat => ( - None, - None, - Some(&[] as &[chrono::DateTime]), - None, - Some(&[] as &[f32]), - ), + MetricKind::TimeseriesInt => (None, None, Some(&[] as &[i32]), None), + MetricKind::TimeseriesFloat => (None, None, None, Some(&[] as &[f32])), }; sqlx::query( - "INSERT INTO job_stats (workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timestamps, timeseries_int, timeseries_float) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", + "INSERT INTO job_stats (workspace_id, job_id, metric_id, metric_name, metric_kind, scalar_int, scalar_float, timeseries_int, timeseries_float, timeseries_start, offsets_cs) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, CASE WHEN $10 THEN now() ELSE NULL END, CASE WHEN $10 THEN ARRAY[]::int[] ELSE NULL END)", ) .bind(workspace_id) .bind(job_id) @@ -83,9 +77,9 @@ pub async fn register_metric_for_job( .bind(metric_kind) .bind(scalar_int) .bind(scalar_float) - .bind(timestamps) .bind(timeseries_int) .bind(timeseries_float) + .bind(is_timeseries) .execute(db) .warn_after_seconds(1) .await?; @@ -117,6 +111,30 @@ pub async fn record_metric( } let metric_kind = metric_kind_opt.unwrap(); + record_metric_impl(db, workspace_id, job_id, metric_id, value, metric_kind).await +} + +/// Record a timeseries metric value without the extra SELECT to look up metric_kind. +/// Use this when the caller already knows the metric kind (e.g. the worker that registered it). +pub async fn record_timeseries_value( + db: &DB, + workspace_id: String, + job_id: Uuid, + metric_id: String, + value: MetricNumericValue, + metric_kind: MetricKind, +) -> error::Result<()> { + record_metric_impl(db, workspace_id, job_id, metric_id, value, metric_kind).await +} + +async fn record_metric_impl( + db: &DB, + workspace_id: String, + job_id: Uuid, + metric_id: String, + value: MetricNumericValue, + metric_kind: MetricKind, +) -> error::Result<()> { let (value_int, value_float) = match value { MetricNumericValue::Integer(val) => { if metric_kind != MetricKind::TimeseriesInt && metric_kind != MetricKind::ScalarInt { @@ -160,7 +178,7 @@ pub async fn record_metric( } MetricKind::TimeseriesInt => { sqlx::query!( - "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_int = array_append(timeseries_int, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", &workspace_id, &job_id, &metric_id, @@ -169,7 +187,7 @@ pub async fn record_metric( } MetricKind::TimeseriesFloat => { sqlx::query!( - "UPDATE job_stats SET timestamps = array_append(timestamps, now()), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", + "UPDATE job_stats SET offsets_cs = array_append(offsets_cs, (EXTRACT(EPOCH FROM (now() - timeseries_start)) * 100)::int), timeseries_float = array_append(timeseries_float, $4) WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3", &workspace_id, &job_id, &metric_id, diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index 1a107990dc..205da6700a 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/backend/windmill-worker/src/handle_child.rs @@ -737,21 +737,24 @@ where let update_job_row = i == 2 || (!*SLOW_LOGS && (i < 20 || (i < 120 && i % 5 == 0) || i % 10 == 0)) || i % 20 == 0; if update_job_row && job_id != Uuid::nil() { if let Connection::Sql(ref db) = conn { - // tracking metric starting at i >= 2 b/c first point it useless and we don't want to track metric for super fast jobs - if i == 2 { - memory_metric_id = job_metrics::register_metric_for_job( - &db, - w_id.to_string(), - job_id, - "memory_kb".to_string(), - job_metrics::MetricKind::TimeseriesInt, - Some("Job Memory Footprint (kB)".to_string()), - ) - .await; - } - if let Ok(ref metric_id) = memory_metric_id { - if let Err(err) = job_metrics::record_metric(&db, w_id.to_string(), job_id, metric_id.to_owned(), job_metrics::MetricNumericValue::Integer(current_mem)).await { - tracing::error!("Unable to save memory stat for job {} in workspace {}. Error was: {:?}", job_id, w_id, err); + // Only track memory when it's non-zero (avoids storing all-zero timeseries for jobs that don't report memory) + if current_mem > 0 { + // Register on first non-zero reading (deferred from i==2 to avoid metric for jobs with no memory reporting) + if memory_metric_id.is_err() { + memory_metric_id = job_metrics::register_metric_for_job( + &db, + w_id.to_string(), + job_id, + "memory_kb".to_string(), + job_metrics::MetricKind::TimeseriesInt, + Some("Job Memory Footprint (kB)".to_string()), + ) + .await; + } + if let Ok(ref metric_id) = memory_metric_id { + if let Err(err) = job_metrics::record_timeseries_value(&db, w_id.to_string(), job_id, metric_id.to_owned(), job_metrics::MetricNumericValue::Integer(current_mem), job_metrics::MetricKind::TimeseriesInt).await { + tracing::error!("Unable to save memory stat for job {} in workspace {}. Error was: {:?}", job_id, w_id, err); + } } } }