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

fix: worker metrics (#3837)

Browse Source 
* fix: worker metrics

* chore: sqlx

* chore: update ee ref

* fix: remove jemalloc stats unwrap
This commit is contained in:
HugoCasa
2024-05-29 01:12:00 +02:00
committed by GitHub
parent df4c3eb977
commit bba0091bff
16 changed files with 176 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-177661a6487cfef198c2ceca4c81e546230c2fd7756d13197a7c19a78eb4b3d6.json → backend/.sqlx/query-4398bc6f701dcba647a259ec2d091a70938a3ecd7ff8a89277cc35d119e2bad4.json
Generated
+3 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, memory_usage = $4, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $5",
"query": "UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $6",
"describe": {
"columns": [],
"parameters": {
@@ -9,10 +9,11 @@
"TextArray",
"Float4",
"Int8",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "177661a6487cfef198c2ceca4c81e546230c2fd7756d13197a7c19a78eb4b3d6"
"hash": "4398bc6f701dcba647a259ec2d091a70938a3ecd7ff8a89277cc35d119e2bad4"
}
backend/.sqlx/query-99d79e2a3792296b0582bbaf8aa1d39855d2109578d0a62d7f2cf1444e44550e.json → backend/.sqlx/query-55d6c6f887f1cceb311d9ab034d3485b8ccb2d07d723f184cef2696760d3ccf9.json
Generated
+8 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate, memory, vcpus, memory_usage\n FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3",
"query": "SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate, memory, vcpus, memory_usage, wm_memory_usage\n FROM worker_ping\n WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)\n ORDER BY ping_at desc LIMIT $2 OFFSET $3",
"describe": {
"columns": [
{
@@ -77,6 +77,11 @@
"ordinal": 14,
"name": "memory_usage",
"type_info": "Int8"
},
{
"ordinal": 15,
"name": "wm_memory_usage",
"type_info": "Int8"
}
],
"parameters": {
@@ -102,8 +107,9 @@
true,
true,
true,
true,
true
]
},
"hash": "99d79e2a3792296b0582bbaf8aa1d39855d2109578d0a62d7f2cf1444e44550e"
"hash": "55d6c6f887f1cceb311d9ab034d3485b8ccb2d07d723f184cef2696760d3ccf9"
}
backend/.sqlx/query-8c62e3bb264c7336b3b3a68677993ba3b729c7358cc3da14513d8b447ce65aaf.json → backend/.sqlx/query-95cb1fe8658f98fb736d899fa21cd7378b0c9d3b5f3d6bd6cafcba273f8277d4.json
Generated
+4 -2
View File
@@ -1,16 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2 WHERE worker = $3",
"query": "UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2, memory_usage = $3, wm_memory_usage = $4 WHERE worker = $5",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Varchar",
"Int8",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "8c62e3bb264c7336b3b3a68677993ba3b729c7358cc3da14513d8b447ce65aaf"
"hash": "95cb1fe8658f98fb736d899fa21cd7378b0c9d3b5f3d6bd6cafcba273f8277d4"
}
backend/Cargo.lock
Generated
+3
View File
@@ -8313,6 +8313,7 @@ dependencies = [
"libc",
"ntapi",
"once_cell",
"rayon",
"windows",
]
@@ -9872,7 +9873,9 @@ dependencies = [
"serde_json",
"sha2 0.10.8",
"sqlx",
"sysinfo",
"thiserror",
"tikv-jemalloc-ctl",
"tokio",
"tracing",
"tracing-flame",
backend/Cargo.toml
+10 -4
View File
@@ -51,7 +51,7 @@ prometheus = ["windmill-common/prometheus", "windmill-api/prometheus", "windmill
flow_testing = ["windmill-worker/flow_testing"]
openidconnect = ["windmill-api/openidconnect"]
cloud = ["windmill-queue/cloud"]
jemalloc = ["dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"]
jemalloc = ["windmill-common/jemalloc", "dep:tikv-jemallocator", "dep:tikv-jemalloc-sys", "dep:tikv-jemalloc-ctl"]
[dependencies]
anyhow.workspace = true
@@ -84,9 +84,9 @@ deno_core.workspace = true
pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true, default-features = false, features = ['rt_tokio']}
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = { optional = true, version = "0.5", features = ["profiling"] }
tikv-jemalloc-sys = { optional = true, version = "^0.5", features = ["profiling"] }
tikv-jemalloc-ctl = { optional = true, version = "^0.5" }
tikv-jemallocator = { optional = true, workspace = true }
tikv-jemalloc-sys = { optional = true, workspace = true }
tikv-jemalloc-ctl = { optional = true, workspace = true }
[dev-dependencies]
serde_json.workspace = true
@@ -248,3 +248,9 @@ crc = "^3"
tar = "^0"
http = "^1"
async-stream = "^0"
sysinfo = "0.30.12"
tikv-jemallocator = { version = "0.5", features = ["profiling"] }
tikv-jemalloc-sys = { version = "^0.5", features = ["profiling"] }
tikv-jemalloc-ctl = { version = "^0.5" }
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
cea722235664845af25689abf98c5fd2678ddf76
553aa1d75498c9c6c07a1495a154478fb0bc064e
backend/migrations/20240528135719_add_windmill_memory_usage.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20240528135719_add_windmill_memory_usage.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE worker_ping ADD COLUMN wm_memory_usage BIGINT;
backend/src/monitor.rs
+33 -9
View File
@@ -250,19 +250,43 @@ pub async fn monitor_mem() {
// Obtain a MIB for the `epoch`, `stats.allocated`, and
// `atats.resident` keys:
let e = epoch::mib().unwrap();
let allocated = stats::allocated::mib().unwrap();
let resident = stats::resident::mib().unwrap();
let e = match epoch::mib() {
Ok(mib) => mib,
Err(e) => {
tracing::error!("Error getting jemalloc epoch mib: {:?}", e);
return;
}
};
let allocated = match stats::allocated::mib() {
Ok(mib) => mib,
Err(e) => {
tracing::error!("Error getting jemalloc allocated mib: {:?}", e);
return;
}
};
let resident = match stats::resident::mib() {
Ok(mib) => mib,
Err(e) => {
tracing::error!("Error getting jemalloc resident mib: {:?}", e);
return;
}
};
loop {
// Many statistics are cached and only updated
// when the epoch is advanced:
e.advance().unwrap();
// Read statistics using MIB key:
let allocated = allocated.read().unwrap();
let resident = resident.read().unwrap();
tracing::info!("{} mb allocated/{} mb resident", bytes_to_mb(allocated as u64), bytes_to_mb(resident as u64));
match e.advance() {
Ok(_) => {
// Read statistics using MIB key:
let allocated = allocated.read().unwrap_or_default();
let resident = resident.read().unwrap_or_default();
tracing::info!("{} mb allocated/{} mb resident", bytes_to_mb(allocated as u64), bytes_to_mb(resident as u64));
},
Err(e) => {
tracing::error!("Error advancing jemalloc epoch: {:?}", e);
}
}
tokio::time::sleep(Duration::from_secs(10)).await;
}
});
backend/windmill-api/openapi.yaml
+2
View File
@@ -9913,6 +9913,8 @@ components:
type: number
memory_usage:
type: number
wm_memory_usage:
type: number
required:
- worker
- worker_instance
backend/windmill-api/src/workers.rs
+3 -1
View File
@@ -59,6 +59,8 @@ struct WorkerPing {
vcpus: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
memory_usage: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
wm_memory_usage: Option<i64>,
}
#[derive(Serialize, Deserialize)]
@@ -86,7 +88,7 @@ async fn list_worker_pings(
let rows = sqlx::query_as!(
WorkerPing,
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate, memory, vcpus, memory_usage
"SELECT worker, worker_instance, EXTRACT(EPOCH FROM (now() - ping_at))::integer as last_ping, started_at, ip, jobs_executed, CASE WHEN $4 IS TRUE THEN current_job_id ELSE NULL END as current_job_id, CASE WHEN $4 IS TRUE THEN current_job_workspace_id ELSE NULL END as current_job_workspace_id, custom_tags, worker_group, wm_version, occupancy_rate, memory, vcpus, memory_usage, wm_memory_usage
FROM worker_ping
WHERE ($1::integer IS NULL AND ping_at > now() - interval '5 minute') OR (ping_at > now() - ($1 || ' seconds')::interval)
ORDER BY ping_at desc LIMIT $2 OFFSET $3",
backend/windmill-common/Cargo.toml
+5
View File
@@ -7,6 +7,7 @@ edition.workspace = true
[features]
default = []
enterprise = []
jemalloc = ["dep:tikv-jemalloc-ctl"]
prometheus = ["dep:prometheus"]
flamegraph = ["dep:tracing-flame"]
loki = ["dep:tracing-loki"]
@@ -49,3 +50,7 @@ aws-sdk-sts = { workspace = true, optional = true }
indexmap.workspace = true
bytes = { workspace = true, optional = true }
mail-send.workspace = true
sysinfo.workspace = true
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemalloc-ctl = { optional = true, workspace = true }
backend/windmill-common/src/worker.rs
+73 -46
View File
@@ -7,6 +7,7 @@ use std::{
collections::{HashMap, HashSet},
sync::{atomic::AtomicBool, Arc},
};
use sysinfo::{MemoryRefreshKind, System};
use tokio::sync::RwLock;
use crate::{error, global_settings::CUSTOM_TAGS_SETTING, server::ServerConfig, DB};
@@ -132,22 +133,81 @@ fn process_custom_tags(tags: Vec<String>) -> (Vec<String>, HashMap<String, Vec<S
(global, specific)
}
pub fn get_worker_memory_usage() -> Option<i64> {
let mut memory = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/memory.current"])
pub fn get_vcpus() -> Option<i64> {
let mut vcpus = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/cpu.max"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string());
.map(|o| {
String::from_utf8_lossy(&o.stdout)
.to_string()
.split(" ")
.map(|s| s.to_string())
.collect::<Vec<String>>()
.get(0)
.map(|s| s.to_string().trim().parse::<i64>().ok())
.flatten()
})
.flatten();
if memory.is_none() {
memory = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/memory/memory.usage_in_bytes"])
if vcpus.is_none() {
vcpus = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/cpu/cpu.cfs_quota_us"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string())
.map(|o| {
String::from_utf8_lossy(&o.stdout)
.to_string()
.trim()
.parse::<i64>()
.ok()
})
.flatten()
}
memory.map(|x| x.parse::<i64>().ok()).flatten()
vcpus
}
pub fn get_memory() -> Option<i64> {
let mut sys = System::new();
sys.refresh_memory();
let limits = sys.cgroup_limits();
if let Some(limits) = limits {
i64::try_from(limits.total_memory).ok()
} else {
None
}
}
pub fn get_worker_memory_usage() -> Option<i64> {
let mut sys = System::new();
sys.refresh_memory_specifics(MemoryRefreshKind::new().with_ram());
i64::try_from(sys.used_memory()).ok()
}
pub fn get_windmill_memory_usage() -> Option<i64> {
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
{
match tikv_jemalloc_ctl::epoch::advance() {
Ok(_) => match tikv_jemalloc_ctl::stats::resident::read() {
Ok(resident) => i64::try_from(resident).ok(),
Err(e) => {
tracing::error!("jemalloc resident memory read failed: {:?}", e);
None
}
},
Err(e) => {
tracing::error!("jemalloc epoch advance failed: {:?}", e);
None
}
}
}
#[cfg(any(target_env = "msvc", not(feature = "jemalloc")))]
{
None
}
}
pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
@@ -160,42 +220,9 @@ pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db:
.map(|x| format!("{}:{}", x.workspace_id, x.path)),
)
};
let mut vcpus = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/cpu.max"])
.output()
.ok()
.map(|o| {
String::from_utf8_lossy(&o.stdout)
.to_string()
.split(" ")
.map(|s| s.to_string())
.collect::<Vec<String>>()
.get(0)
.map(|s| s.to_string())
})
.flatten();
if vcpus.is_none() {
vcpus = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/cpu/cpu.cfs_quota_us"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string())
}
let mut memory = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/memory.max"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string());
if memory.is_none() {
memory = std::process::Command::new("cat")
.args(["/sys/fs/cgroup/memory/memory.limit_in_bytes"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string())
}
let vcpus = get_vcpus();
let memory = get_memory();
sqlx::query!(
"INSERT INTO worker_ping (worker_instance, worker, ip, custom_tags, worker_group, dedicated_worker, wm_version, vcpus, memory) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (worker) DO UPDATE set ip = $3, custom_tags = $4, worker_group = $5",
@@ -206,8 +233,8 @@ pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db:
*WORKER_GROUP,
dw,
crate::utils::GIT_VERSION,
vcpus.map(|x| x.parse::<i64>().ok()).flatten(),
memory.map(|x| x.parse::<i64>().ok()).flatten()
vcpus,
memory
)
.execute(db)
.await
backend/windmill-worker/src/common.rs
+8 -2
View File
@@ -27,7 +27,9 @@ use windmill_common::s3_helpers::{
get_etag_or_empty, LargeFileStorage, ObjectStoreResource, S3Object,
};
use windmill_common::variables::{build_crypt_with_key_suffix, decrypt_value_with_mc};
use windmill_common::worker::{CLOUD_HOSTED, TMP_DIR, WORKER_CONFIG};
use windmill_common::worker::{
get_windmill_memory_usage, get_worker_memory_usage, CLOUD_HOSTED, TMP_DIR, WORKER_CONFIG,
};
use windmill_common::{
error::{self, Error},
jobs::QueuedJob,
@@ -582,10 +584,14 @@ where
// update the last_ping column every 5 seconds
i+=1;
if i % 10 == 0 {
let memory_usage = get_worker_memory_usage();
let wm_memory_usage = get_windmill_memory_usage();
sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2 WHERE worker = $3",
"UPDATE worker_ping SET ping_at = now(), current_job_id = $1, current_job_workspace_id = $2, memory_usage = $3, wm_memory_usage = $4 WHERE worker = $5",
&job_id,
&w_id,
memory_usage,
wm_memory_usage,
&worker_name
)
.execute(&db)
backend/windmill-worker/src/worker.rs
+4 -3
View File
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use windmill_common::worker::{get_worker_memory_usage, TMP_DIR};
use windmill_common::worker::{get_windmill_memory_usage, get_worker_memory_usage, TMP_DIR};
use anyhow::Result;
use const_format::concatcp;
@@ -1427,13 +1427,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let tags = WORKER_CONFIG.read().await.worker_tags.clone();
let memory_usage = get_worker_memory_usage();
let wm_memory_usage = get_windmill_memory_usage();
if let Err(e) = sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, memory_usage = $4, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $5",
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, memory_usage = $4, wm_memory_usage = $5, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $6",
jobs_executed,
tags.as_slice(),
worker_code_execution_metric / start_time.elapsed().as_secs_f32(),
memory_usage,
wm_memory_usage,
&worker_name
).execute(db).await {
tracing::error!("failed to update worker ping, exiting: {}", e);
frontend/src/routes/(root)/(logged)/workers/+page.svelte
+16 -9
View File
@@ -311,6 +311,9 @@
>
{#each groupedWorkers as worker_group (worker_group[0])}
{@const config = (workerGroups ?? {})[worker_group[0]]}
{@const activeWorkers = worker_group?.[1].flatMap((x) =>
x[1]?.filter((y) => (y.last_ping ?? 0) < 15)
)}
<WorkspaceGroup
{customTags}
name={worker_group[0]}
@@ -318,9 +321,7 @@
on:reload={() => {
loadWorkerGroups()
}}
activeWorkers={worker_group?.[1].flatMap((x) =>
x[1]?.filter((y) => (y.last_ping ?? 0) < 15)
)?.length ?? 0}
activeWorkers={activeWorkers?.length ?? 0}
{defaultTagPerWorkspace}
/>
@@ -340,13 +341,13 @@
</Cell>
<Cell head>Last ping</Cell>
<Cell head>Worker start</Cell>
<Cell head>Nb of jobs executed</Cell>
<Cell head>Jobs ran</Cell>
{#if (!config || config?.dedicated_worker == undefined) && $superadmin}
<Cell head>Current job</Cell>
<Cell head>Occupancy rate</Cell>
{/if}
<Cell head>Memory usage</Cell>
<Cell head>vCPUs/memory limits</Cell>
<Cell head>Memory usage<br />(Windmill usage)</Cell>
<Cell head>Limits</Cell>
<Cell head>Version</Cell>
<Cell head last>Liveness</Cell>
</tr>
@@ -369,7 +370,7 @@
</tr>
{#if workers}
{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, current_job_id, current_job_workspace_id, occupancy_rate, wm_version, vcpus, memory, memory_usage }}
{#each workers as { worker, custom_tags, last_ping, started_at, jobs_executed, current_job_id, current_job_workspace_id, occupancy_rate, wm_version, vcpus, memory, memory_usage, wm_memory_usage }}
<tr>
<Cell first>{worker}</Cell>
<Cell>
@@ -389,6 +390,7 @@
<a href={`/run/${current_job_id}?workspace=${current_job_workspace_id}`}>
View job
</a>
<br />
(workspace {current_job_workspace_id})
{/if}
</Cell>
@@ -396,10 +398,15 @@
{Math.ceil(occupancy_rate ?? 0 * 100)}%
</Cell>
{/if}
<Cell>{memory_usage ? Math.round(memory_usage / 1000000) + 'MB' : '--'}</Cell>
<Cell
>{memory_usage ? Math.round(memory_usage / 1024 / 1024) + 'MB' : '--'}<br
/>({wm_memory_usage
? Math.round(wm_memory_usage / 1024 / 1024) + 'MB'
: '--'})</Cell
>
<Cell>
{vcpus ? (vcpus / 100000).toFixed(1) + ' vCPUs' : '--'}
/ {memory ? Math.round(memory / 1000000) + 'MB' : '--'}
/ {memory ? Math.round(memory / 1024 / 1024) + 'MB' : '--'}
</Cell>
<Cell>
<div class="!text-2xs">