kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update metrics

Browse Source
This commit is contained in:
Ruben Fiszel
2023-02-27 11:56:55 +01:00
parent 1414030afe
commit 020e5f7b32
5 changed files with 1211 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/migrations/20230227100232_update_included_scripts_with_deno_changes.down.sql Normal file
View File

@@ -0,0 +1 @@
-- Add down migration script here

1182
backend/migrations/20230227100232_update_included_scripts_with_deno_changes.up.sql Normal file
View File

File diff suppressed because it is too large Load Diff

13
backend/windmill-queue/src/jobs.rs
View File

@@ -351,7 +351,10 @@ pub async fn push<'c>(
.unwrap_or(false);
if !is_super_admin {
if usage > MAX_FREE_EXECS {
if usage > MAX_FREE_EXECS
&& !matches!(job_payload, JobPayload::Dependencies { .. })
&& !matches!(job_payload, JobPayload::FlowDependencies { .. })
{
return Err(error::Error::BadRequest(format!(
"User {email} has exceeded the free usage limit of {MAX_FREE_EXECS} that applies outside of premium workspaces."
)));
@@ -466,10 +469,10 @@ pub async fn push<'c>(
}
JobPayload::Flow(flow) => {
let value_json = sqlx::query_scalar!(
"SELECT value FROM flow WHERE path = $1 AND workspace_id = $2",
flow,
workspace_id
)
"SELECT value FROM flow WHERE path = $1 AND workspace_id = $2",
flow,
workspace_id
)
.fetch_optional(&mut tx)
.await?
.ok_or_else(|| Error::InternalErr(format!("not found flow at path {:?}", flow)))?;

40
backend/windmill-worker/src/worker.rs
View File

@@ -383,7 +383,7 @@ lazy_static::lazy_static! {
.unwrap();
static ref WORKER_UPTIME_OPTS: prometheus::Opts = prometheus::opts!(
"worker_uptime",
"Total number of milliseconds since the worker has started"
"Total number of seconds since the worker has started"
);
static ref TIMEOUT: u16 = std::env::var("TIMEOUT")
@@ -445,15 +445,11 @@ pub async fn run_worker(
insert_initial_ping(worker_instance, &worker_name, ip, db).await;
let uptime_metric = prometheus::register_int_counter!(WORKER_UPTIME_OPTS
let uptime_metric = prometheus::register_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 worker_execution_duration = prometheus::register_histogram_vec!(
prometheus::HistogramOpts::new(
@@ -465,9 +461,9 @@ pub async fn run_worker(
)
.expect("register prometheus metric");
let worker_execution_duration_counter = prometheus::register_int_counter!(prometheus::opts!(
let worker_execution_duration_counter = prometheus::register_counter!(prometheus::opts!(
"worker_execution_duration_counter",
"Total number of milliseconds spent executing jobs"
"Total number of seconds spent executing jobs"
)
.const_label("name", &worker_name))
.expect("register prometheus metric");
@@ -481,9 +477,9 @@ pub async fn run_worker(
.expect("register prometheus metric");
let worker_sleep_duration_counter = prometheus::register_int_counter!(prometheus::opts!(
let worker_sleep_duration_counter = prometheus::register_counter!(prometheus::opts!(
"worker_execution_sleep_counter",
"Total number of milliseconds spent sleeping between pulling jobs from the queue"
"Total number of seconds spent sleeping between pulling jobs from the queue"
)
.const_label("name", &worker_name))
.expect("register prometheus metric");
@@ -496,9 +492,9 @@ pub async fn run_worker(
.const_label("name", &worker_name),)
.expect("register prometheus metric");
let worker_pull_duration_counter = prometheus::register_int_counter!(prometheus::opts!(
let worker_pull_duration_counter = prometheus::register_counter!(prometheus::opts!(
"worker_pull_sleep_counter",
"Total number of milliseconds spent pulling jobs (if growing large the db is undersized)"
"Total number of seconds spent pulling jobs (if growing large the db is undersized)"
)
.const_label("name", &worker_name))
.expect("register prometheus metric");
@@ -550,11 +546,12 @@ pub async fn run_worker(
worker_busy.set(0);
uptime_metric.inc_by(
((Instant::now() - start_time).as_millis() - uptime_metric.get() as u128)
(((Instant::now() - start_time).as_millis() as f64)/1000.0 - uptime_metric.get())
.try_into()
.unwrap(),
);
let do_break = async {
if last_ping.elapsed().as_secs() > NUM_SECS_ENV_CHECK {
sqlx::query!(
@@ -595,12 +592,11 @@ pub async fn run_worker(
.await
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string())))
},
(job, timer, instant) = {
(job, timer) = {
let timer = worker_pull_duration.start_timer();
let instant = Instant::now().to_owned();
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer, instant)) } => {
worker_pull_duration_counter.inc_by((instant.elapsed().as_millis() as i64).try_into().unwrap());
drop(timer);
pull(&db, WHITELIST_WORKSPACES.clone(), BLACKLIST_WORKSPACES.clone()).map(|x| (x, timer)) } => {
let duration_pull_s = timer.stop_and_record();
worker_pull_duration_counter.inc_by(duration_pull_s);
(false, job)
},
}
@@ -706,6 +702,8 @@ pub async fn run_worker(
.await;
};
let duration = _timer.stop_and_record();
worker_execution_duration_counter.inc_by(duration);
if !*KEEP_JOB_DIR && !(is_flow && same_worker) {
let _ = tokio::fs::remove_dir_all(job_dir).await;
@@ -715,9 +713,9 @@ pub async fn run_worker(
let _timer = worker_sleep_duration
.start_timer();
tokio::time::sleep(Duration::from_millis(*SLEEP_QUEUE)).await;
let duration = _timer.stop_and_record();
worker_sleep_duration_counter.inc_by(duration);
}
Err(err) => {
tracing::error!(worker = %worker_name, "run_worker: pulling jobs: {}", err);

2
cli/login.ts
View File

@@ -54,7 +54,7 @@ export async function browserLogin(
await open(url)
console.log("Opened browser for you");
} catch {
console.error(`Failed to open browser, please navgiate to ${url}`)
console.error(`Failed to open browser, please navigate to ${url}`)
}
const firstConnection = await server.accept();
const httpFirstConnection = Deno.serveHttp(firstConnection);