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

add more metrics

Browse Source
This commit is contained in:
Ruben Fiszel
2023-10-28 15:35:45 +02:00
parent 0a2648f560
commit 2cec6a0971
3 changed files with 63 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+1 -1
View File
@@ -543,7 +543,7 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
for i in 1..(num_workers + 1) {
let db1 = db.clone();
let instance_name = instance_name.clone();
let worker_name = format!("wk-{}-{}", &instance_name, rd_string(5));
let worker_name = format!("wk-{}-{}-{}", *WORKER_GROUP, &instance_name, rd_string(5));
let ip = ip.clone();
let rx = rx.resubscribe();
let tx = tx.clone();
backend/windmill-worker/src/worker.rs
+58 -13
View File
@@ -601,7 +601,22 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
"worker_pull_duration",
"Duration pulling next job",
)
.const_label("name", &worker_name),)
.const_label("name", &worker_name)
.const_label("has_job", "true"),)
.expect("register prometheus metric"),
)
} else {
None
};
let worker_pull_duration_empty = if METRICS_ENABLED.load(Ordering::Relaxed) {
Some(
prometheus::register_histogram!(prometheus::HistogramOpts::new(
"worker_pull_duration",
"Duration pulling next job",
)
.const_label("name", &worker_name)
.const_label("has_job", "false"),)
.expect("register prometheus metric"),
)
} else {
@@ -698,6 +713,21 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
None
};
let worker_pull_duration_counter_empty =
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
Some(
prometheus::register_counter!(prometheus::opts!(
"worker_pull_duration_counter",
"Total number of seconds spent pulling jobs (if growing large the db is undersized)"
)
.const_label("name", &worker_name)
.const_label("has_job", "false"))
.expect("register prometheus metric"),
)
} else {
None
};
let worker_pull_duration_counter = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed)
{
Some(
@@ -705,13 +735,13 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
"worker_pull_duration_counter",
"Total number of seconds spent pulling jobs (if growing large the db is undersized)"
)
.const_label("name", &worker_name))
.const_label("name", &worker_name)
.const_label("has_job", "true"))
.expect("register prometheus metric"),
)
} else {
None
};
let worker_busy: Option<prometheus::IntGauge> =
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
Some(
@@ -1287,24 +1317,39 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.await
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()))
},
(job, timer) = {
let timer = worker_pull_duration.as_ref().map(|x| x.start_timer());
(job, timer) = async {
let pull_time = Instant::now();
let suspend_first = if last_checked_suspended.elapsed().as_secs() > 3 {
last_checked_suspended = Instant::now();
true
} else { false };
pull(&db, rsmq.clone(), suspend_first).map(|x| (x, timer))
pull(&db, rsmq.clone(), suspend_first).map(|x| (x, pull_time)).await
} => {
add_time!(timing, loop_start, "post pull");
timer.map(|timer| {
let duration_pull_s = timer.stop_and_record();
if let Some(wp) = worker_pull_duration_counter.as_ref() {
wp.inc_by(duration_pull_s);
let duration_pull_s = timer.elapsed().as_secs_f64();
if duration_pull_s > 0.5 {
tracing::error!("pull took more than 0.5s, this is a sign that the database is VERY undersized for this load")
} else if duration_pull_s > 0.1 {
tracing::error!("pull took more than 0.5s, this is a sign that the database is undersized for this load")
}
if let Ok(j) = job.as_ref() {
if j.is_some() {
if let Some(wp) = worker_pull_duration_counter.as_ref() {
wp.inc_by(duration_pull_s);
}
if let Some(wp) = worker_pull_duration.as_ref() {
wp.observe(duration_pull_s);
}
} else {
if let Some(wp) = worker_pull_duration_counter_empty.as_ref() {
wp.inc_by(duration_pull_s);
}
if let Some(wp) = worker_pull_duration_empty.as_ref() {
wp.observe(duration_pull_s);
}
}
});
}
job
},
}
};
benchmarks/pulumi/index.ts
+4 -9
View File
@@ -242,7 +242,7 @@ const lb = new awsx.lb.ApplicationLoadBalancer("lb2", {
// desiredCount: 2,
// taskDefinitionArgs: {
// container: {
// image: "nginx:latest",
// image: "nginx:main",
// cpu: 512,
// memory: 128,
// essential: true,
@@ -261,7 +261,7 @@ db.address.apply((address) => {
containerDefinitions: JSON.stringify([
{
name: "windmill-worker",
image: "ghcr.io/windmill-labs/windmill-ee:latest",
image: "ghcr.io/windmill-labs/windmill-ee:main",
cpu: 1024,
memory: 1800,
essential: true,
@@ -319,7 +319,7 @@ db.address.apply((address) => {
containerDefinitions: JSON.stringify([
{
name: "windmill-worker",
image: "ghcr.io/windmill-labs/windmill-ee:latest",
image: "ghcr.io/windmill-labs/windmill-ee:main",
cpu: 1024,
memory: 1800,
essential: true,
@@ -378,7 +378,7 @@ db.address.apply((address) => {
containerDefinitions: JSON.stringify([
{
name: "windmill-server",
image: "ghcr.io/windmill-labs/windmill-ee:latest",
image: "ghcr.io/windmill-labs/windmill-ee:main",
cpu: 1024,
memory: 1024,
essential: true,
@@ -447,11 +447,6 @@ db.address.apply((address) => {
field: "cpu",
},
],
serviceRegistries: {
registryArn: "aws.servicediscovery.Service",
containerName: "windmill-worker",
containerPort: 8001
}
});
const service_worker2 = new aws.ecs.Service("service-worker-2", {