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

feat: worker vcpus/memory limits + mem usage (#3828)

Browse Source
This commit is contained in:
HugoCasa
2024-05-27 18:43:02 +02:00
committed by GitHub
parent 1e877c0ea1
commit 5de79aace3
9 changed files with 77 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-54fef88cc6b9e8db7c07fccbaa845edfefa339153a32e468faad9f063008863d.json → backend/.sqlx/query-177661a6487cfef198c2ceca4c81e546230c2fd7756d13197a7c19a78eb4b3d6.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, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $4",
"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",
"describe": {
"columns": [],
"parameters": {
@@ -8,10 +8,11 @@
"Int4",
"TextArray",
"Float4",
"Int8",
"Text"
]
},
"nullable": []
},
"hash": "54fef88cc6b9e8db7c07fccbaa845edfefa339153a32e468faad9f063008863d"
"hash": "177661a6487cfef198c2ceca4c81e546230c2fd7756d13197a7c19a78eb4b3d6"
}
backend/.sqlx/query-e00171cc3fc8f32922562d4fc4d3db56955fbc6ee9872d4b2e99157ebed131e0.json → backend/.sqlx/query-99d79e2a3792296b0582bbaf8aa1d39855d2109578d0a62d7f2cf1444e44550e.json
Generated
+20 -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\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\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": [
{
@@ -62,6 +62,21 @@
"ordinal": 11,
"name": "occupancy_rate",
"type_info": "Float4"
},
{
"ordinal": 12,
"name": "memory",
"type_info": "Int8"
},
{
"ordinal": 13,
"name": "vcpus",
"type_info": "Int8"
},
{
"ordinal": 14,
"name": "memory_usage",
"type_info": "Int8"
}
],
"parameters": {
@@ -84,8 +99,11 @@
true,
false,
false,
true,
true,
true,
true
]
},
"hash": "e00171cc3fc8f32922562d4fc4d3db56955fbc6ee9872d4b2e99157ebed131e0"
"hash": "99d79e2a3792296b0582bbaf8aa1d39855d2109578d0a62d7f2cf1444e44550e"
}
backend/migrations/20240527151853_add_worker_memory_usage.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20240527151853_add_worker_memory_usage.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TABLE worker_ping ADD COLUMN memory_usage BIGINT;
backend/windmill-api/openapi.yaml
+6
View File
@@ -9907,6 +9907,12 @@ components:
type: string
occupancy_rate:
type: number
memory:
type: number
vcpus:
type: number
memory_usage:
type: number
required:
- worker
- worker_instance
backend/windmill-api/src/workers.rs
+8 -1
View File
@@ -51,7 +51,14 @@ struct WorkerPing {
custom_tags: Option<Vec<String>>,
worker_group: String,
wm_version: String,
#[serde(skip_serializing_if = "Option::is_none")]
occupancy_rate: Option<f32>,
#[serde(skip_serializing_if = "Option::is_none")]
memory: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
vcpus: Option<i64>,
#[serde(skip_serializing_if = "Option::is_none")]
memory_usage: Option<i64>,
}
#[derive(Serialize, Deserialize)]
@@ -79,7 +86,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
"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
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/src/worker.rs
+18
View File
@@ -132,6 +132,24 @@ 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"])
.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.usage_in_bytes"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).to_string())
}
memory.map(|x| x.parse::<i64>().ok()).flatten()
}
pub async fn update_ping(worker_instance: &str, worker_name: &str, ip: &str, db: &DB) {
let (tags, dw) = {
let wc = WORKER_CONFIG.read().await.clone();
backend/windmill-worker/src/worker.rs
+5 -2
View File
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use windmill_common::worker::TMP_DIR;
use windmill_common::worker::{get_worker_memory_usage, TMP_DIR};
use anyhow::Result;
use const_format::concatcp;
@@ -1426,11 +1426,14 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
if last_ping.elapsed().as_secs() > NUM_SECS_PING {
let tags = WORKER_CONFIG.read().await.worker_tags.clone();
let memory_usage = get_worker_memory_usage();
if let Err(e) = sqlx::query!(
"UPDATE worker_ping SET ping_at = now(), jobs_executed = $1, custom_tags = $2, occupancy_rate = $3, current_job_id = NULL, current_job_workspace_id = NULL WHERE worker = $4",
"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",
jobs_executed,
tags.as_slice(),
worker_code_execution_metric / start_time.elapsed().as_secs_f32(),
memory_usage,
&worker_name
).execute(db).await {
tracing::error!("failed to update worker ping, exiting: {}", e);
frontend/src/routes/(root)/(logged)/workers/+page.svelte
+14 -7
View File
@@ -345,6 +345,8 @@
<Cell head>Current job</Cell>
<Cell head>Occupancy rate</Cell>
{/if}
<Cell head>Memory usage</Cell>
<Cell head>vCPUs/memory limits</Cell>
<Cell head>Version</Cell>
<Cell head last>Liveness</Cell>
</tr>
@@ -354,7 +356,7 @@
<tr class="border-t">
<Cell
first
colspan={(!config || config?.dedicated_worker == undefined) && $superadmin ? 9 : 7}
colspan={(!config || config?.dedicated_worker == undefined) && $superadmin ? 11 : 9}
scope="colgroup"
class="bg-surface-secondary/60 py-2 border-b"
>
@@ -367,7 +369,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 }}
{#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 }}
<tr>
<Cell first>{worker}</Cell>
<Cell>
@@ -394,11 +396,16 @@
{Math.ceil(occupancy_rate ?? 0 * 100)}%
</Cell>
{/if}
<Cell
><div class="!text-2xs"
>{wm_version.split('-')[0]}<Tooltip>{wm_version}</Tooltip></div
></Cell
>
<Cell>{memory_usage ? Math.round(memory_usage / 1000000) + 'MB' : '--'}</Cell>
<Cell>
{vcpus ? (vcpus / 100000).toFixed(1) + ' vCPUs' : '--'}
/ {memory ? Math.round(memory / 1000000) + 'MB' : '--'}
</Cell>
<Cell>
<div class="!text-2xs">
{wm_version.split('-')[0]}<Tooltip>{wm_version}</Tooltip>
</div>
</Cell>
<Cell last>
<Badge
color={last_ping != undefined ? (last_ping < 60 ? 'green' : 'red') : 'gray'}