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: prometheus metrics are an instance settings

Browse Source
This commit is contained in:
Ruben Fiszel
2023-10-25 23:32:39 +02:00
parent 814f384751
commit 57dc62b298
9 changed files with 103 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/main.rs
+24 -9
View File
@@ -23,14 +23,14 @@ use tokio::{
use windmill_api::HTTP_CLIENT;
use windmill_common::{
global_settings::{
BASE_URL_SETTING, CUSTOM_TAGS_SETTING, DISABLE_STATS_SETTING, ENV_SETTINGS,
BASE_URL_SETTING, CUSTOM_TAGS_SETTING, DISABLE_STATS_SETTING, ENV_SETTINGS, EXPOSE_METRICS,
EXTRA_PIP_INDEX_URL_SETTING, LICENSE_KEY_SETTING, NPM_CONFIG_REGISTRY_SETTING,
OAUTH_SETTING, REQUEST_SIZE_LIMIT_SETTING, RETENTION_PERIOD_SECS_SETTING,
},
stats::schedule_stats,
utils::rd_string,
worker::{reload_custom_tags_setting, WORKER_GROUP},
DB, METRICS_ADDR,
DB, METRICS_ADDR, METRICS_ENABLED,
};
use windmill_worker::{
BUN_CACHE_DIR, BUN_TMP_CACHE_DIR, DENO_CACHE_DIR, DENO_CACHE_DIR_DEPS, DENO_CACHE_DIR_NPM,
@@ -65,6 +65,10 @@ pub enum Mode {
async fn main() -> anyhow::Result<()> {
dotenv::dotenv().ok();
if std::env::var("RUST_LOG").is_err() {
std::env::set_var("RUST_LOG", "info")
}
#[cfg(not(feature = "flamegraph"))]
windmill_common::tracing_init::initialize_tracing();
@@ -124,7 +128,6 @@ async fn main() -> anyhow::Result<()> {
"We STRONGLY recommend using at most 1 worker per container, use at your own risks"
);
}
let metrics_addr: Option<SocketAddr> = *METRICS_ADDR;
let server_mode = !std::env::var("DISABLE_SERVER")
.ok()
@@ -338,15 +341,26 @@ Windmill Community Edition {GIT_VERSION}
NPM_CONFIG_REGISTRY_SETTING => {
reload_npm_config_registry_setting(&db).await
},
REQUEST_SIZE_LIMIT_SETTING => {
tracing::info!("Request limit size change detected, killing server expecting to be restarted");
// we wait a bit randomly to avoid having all servers shutdown at same time
EXPOSE_METRICS => {
tracing::info!("Metrics setting changed, restarting");
// we wait a bit randomly to avoid having all serverss and workers shutdown at same time
let rd_delay = rand::thread_rng().gen_range(0..4);
tokio::time::sleep(Duration::from_secs(rd_delay)).await;
if let Err(e) = tx.send(()) {
tracing::error!(error = %e, "Could not send killpill to server");
}
},
REQUEST_SIZE_LIMIT_SETTING => {
if server_mode {
tracing::info!("Request limit size change detected, killing server expecting to be restarted");
// we wait a bit randomly to avoid having all servers shutdown at same time
let rd_delay = rand::thread_rng().gen_range(0..4);
tokio::time::sleep(Duration::from_secs(rd_delay)).await;
if let Err(e) = tx.send(()) {
tracing::error!(error = %e, "Could not send killpill to server");
}
}
},
DISABLE_STATS_SETTING => {},
a @_ => {
tracing::info!("Unrecognized Global Setting Change Payload: {:?}", a);
@@ -381,12 +395,13 @@ Windmill Community Edition {GIT_VERSION}
};
let metrics_f = async {
if let Some(_addr) = metrics_addr {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
#[cfg(not(feature = "enterprise"))]
panic!("Metrics are only available in the Enterprise Edition");
tracing::error!("Metrics are only available in the EE, ignoring...");
#[cfg(feature = "enterprise")]
windmill_common::serve_metrics(_addr, rx.resubscribe(), num_workers > 0).await;
windmill_common::serve_metrics(*METRICS_ADDR, rx.resubscribe(), num_workers > 0)
.await;
}
Ok(()) as anyhow::Result<()>
};
backend/src/monitor.rs
+29 -5
View File
@@ -1,4 +1,11 @@
use std::{collections::HashMap, fmt::Display, ops::Mul, str::FromStr, sync::Arc, time::Duration};
use std::{
collections::HashMap,
fmt::Display,
ops::Mul,
str::FromStr,
sync::{atomic::Ordering, Arc},
time::Duration,
};
use serde::de::DeserializeOwned;
use sqlx::{Pool, Postgres};
@@ -14,7 +21,7 @@ use windmill_api::{
use windmill_common::{
error,
global_settings::{
BASE_URL_SETTING, EXTRA_PIP_INDEX_URL_SETTING, LICENSE_KEY_SETTING,
BASE_URL_SETTING, EXPOSE_METRICS, EXTRA_PIP_INDEX_URL_SETTING, LICENSE_KEY_SETTING,
NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING, REQUEST_SIZE_LIMIT_SETTING,
RETENTION_PERIOD_SECS_SETTING,
},
@@ -76,6 +83,9 @@ pub async fn initial_load(
worker_mode: bool,
server_mode: bool,
) {
if let Err(e) = load_metrics_enabled(db).await {
tracing::error!("Error reloading loading metrics: {e}");
}
let reload_worker_config_f = async {
if worker_mode {
reload_worker_config(&db, tx, false).await;
@@ -144,6 +154,20 @@ pub async fn initial_load(
);
}
pub async fn load_metrics_enabled(db: &DB) -> error::Result<()> {
let metrics_enabled = sqlx::query_scalar!(
"SELECT value FROM global_settings WHERE name = $1",
EXPOSE_METRICS
)
.fetch_optional(db)
.await;
match metrics_enabled {
Ok(Some(serde_json::Value::Bool(t))) => METRICS_ENABLED.store(t, Ordering::Relaxed),
_ => (),
};
Ok(())
}
pub async fn delete_expired_items(db: &DB) -> () {
let tokens_deleted_r: std::result::Result<Vec<String>, _> = sqlx::query_scalar(
"DELETE FROM token WHERE expiration <= now()
@@ -413,7 +437,7 @@ pub async fn monitor_db<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
};
let expose_queue_metrics_f = async {
if *METRICS_ENABLED && server_mode {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) && server_mode {
expose_queue_metrics(&db).await;
}
};
@@ -587,7 +611,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.ok()
.unwrap_or_else(|| vec![]);
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_ZOMBIE_RESTART_COUNT.inc_by(restarted.len() as _);
}
for r in restarted {
@@ -613,7 +637,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.ok()
.unwrap_or_else(|| vec![]);
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_ZOMBIE_DELETE_COUNT.inc_by(timeouts.len() as _);
}
backend/windmill-api/src/webhook_util.rs
+2 -2
View File
@@ -109,13 +109,13 @@ impl WebhookShared {
}
};
if let Some(url) = webhook_opt {
let timer = if *METRICS_ENABLED { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
let timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
let _ = client.post(url).json(&message).send().await;
timer.map(|x| x.stop_and_record());
}
},
Some(WebhookPayload::InstanceEvent(event)) => {
if *METRICS_ENABLED { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(WEBHOOK_REQUEST_COUNT.start_timer()) } else { None };
let r = client.post(INSTANCE_EVENTS_WEBHOOK.as_ref().unwrap()).json(&event).send().await;
if let Err(e) = r {
tracing::error!("Error sending instance event: {}", e);
backend/windmill-common/src/global_settings.rs
+1
View File
@@ -9,6 +9,7 @@ pub const NPM_CONFIG_REGISTRY_SETTING: &str = "npm_config_registry";
pub const EXTRA_PIP_INDEX_URL_SETTING: &str = "pip_extra_index_url";
pub const UNIQUE_ID_SETTING: &str = "uid";
pub const DISABLE_STATS_SETTING: &str = "disable_stats";
pub const EXPOSE_METRICS: &str = "expose_metrics";
pub const ENV_SETTINGS: [&str; 54] = [
"DISABLE_NSJAIL",
backend/windmill-common/src/lib.rs
+27 -7
View File
@@ -6,7 +6,10 @@
* LICENSE-AGPL for a copy of the license.
*/
use std::{net::SocketAddr, sync::Arc};
use std::{
net::SocketAddr,
sync::{atomic::AtomicBool, Arc},
};
use error::Error;
use scripts::ScriptLang;
@@ -38,17 +41,24 @@ pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;
lazy_static::lazy_static! {
pub static ref METRICS_ADDR: Option<SocketAddr> = std::env::var("METRICS_ADDR")
pub static ref METRICS_PORT: u16 = std::env::var("METRICS_PORT")
.ok()
.and_then(|s| s.parse::<u16>().ok())
.unwrap_or(8001);
pub static ref METRICS_ADDR: SocketAddr = std::env::var("METRICS_ADDR")
.ok()
.map(|s| {
s.parse::<bool>()
.map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], 8001))))
.map(|b| b.then(|| SocketAddr::from(([0, 0, 0, 0], *METRICS_PORT))))
.or_else(|_| s.parse::<SocketAddr>().map(Some))
})
.transpose().ok()
.flatten()
.flatten();
pub static ref METRICS_ENABLED: bool = METRICS_ADDR.is_some();
.flatten()
.unwrap_or_else(|| SocketAddr::from(([0, 0, 0, 0], *METRICS_PORT)));
pub static ref METRICS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("METRICS_PORT").is_ok() || std::env::var("METRICS_ADDR").is_ok());
pub static ref BASE_URL: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
pub static ref IS_READY: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
}
@@ -92,9 +102,14 @@ pub async fn serve_metrics(
) -> JoinHandle<()> {
use std::sync::atomic::Ordering;
use axum::{routing::get, Router};
use axum::{
routing::{get, post},
Router,
};
use hyper::StatusCode;
let router = Router::new().route("/metrics", get(metrics));
let router = Router::new()
.route("/metrics", get(metrics))
.route("/reset", post(reset));
let router = if ready_worker_endpoint {
router.route(
@@ -112,6 +127,7 @@ pub async fn serve_metrics(
};
tokio::spawn(async move {
tracing::info!("Serving metrics at: {addr}");
if let Err(e) = axum::Server::bind(&addr)
.serve(router.into_make_service())
.with_graceful_shutdown(async {
@@ -132,6 +148,10 @@ async fn metrics() -> Result<String, Error> {
.map_err(anyhow::Error::from)?)
}
async fn reset() -> () {
todo!()
}
#[cfg(feature = "sqlx")]
pub async fn connect_db(server_mode: bool) -> anyhow::Result<sqlx::Pool<sqlx::Postgres>> {
use anyhow::Context;
backend/windmill-queue/src/jobs.rs
+5 -5
View File
@@ -195,7 +195,7 @@ pub async fn add_completed_job_error<R: rsmq_async::RsmqConnection + Clone + Sen
metrics: Option<Metrics>,
rsmq: Option<R>,
) -> Result<WrappedError, Error> {
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
metrics.map(|m| m.worker_execution_failed.inc());
}
let result = WrappedError { error: e };
@@ -1076,7 +1076,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
|| pulled_job.concurrent_limit.is_none()
|| pulled_job.canceled
{
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PULL_COUNT.inc();
}
return Ok(Option::Some(pulled_job));
@@ -1162,7 +1162,7 @@ pub async fn pull<R: rsmq_async::RsmqConnection + Send + Clone>(
concurrent_jobs_for_this_script
);
if concurrent_jobs_for_this_script <= job_custom_concurrent_limit {
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PULL_COUNT.inc();
}
tx.commit().await?;
@@ -1503,7 +1503,7 @@ pub async fn delete_job<'c, R: rsmq_async::RsmqConnection + Clone + Send>(
w_id: &str,
job_id: Uuid,
) -> windmill_common::error::Result<QueueTransaction<'c, R>> {
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_DELETE_COUNT.inc();
}
let job_removed = sqlx::query_scalar!(
@@ -2211,7 +2211,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
.map_err(|e| Error::InternalErr(format!("Could not insert into queue {job_id}: {e}")))?;
// TODO: technically the job isn't queued yet, as the transaction can be rolled back. Should be solved when moving these metrics to the queue abstraction.
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
QUEUE_PUSH_COUNT.inc();
}
backend/windmill-worker/src/worker.rs
+7 -7
View File
@@ -656,7 +656,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let mut jobs_executed = 0;
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
WORKER_STARTED.inc();
}
@@ -1035,7 +1035,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
#[cfg(feature = "benchmark")]
let mut timing = vec![];
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
worker_busy.set(0);
uptime_metric.inc_by(
((start_time.elapsed().as_millis() as f64) / 1000.0 - uptime_metric.get())
@@ -1169,7 +1169,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()))
},
(job, timer) = {
let timer = if *METRICS_ENABLED { Some(worker_pull_duration.start_timer()) } else { None };
let timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) { Some(worker_pull_duration.start_timer()) } else { None };
let suspend_first = if last_checked_suspended.elapsed().as_secs() > 3 {
last_checked_suspended = Instant::now();
true
@@ -1188,7 +1188,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
};
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
worker_busy.set(1);
}
@@ -1244,7 +1244,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
.expect("no timer found")
.start_timer();
if *METRICS_ENABLED {
if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
worker_execution_count
.get(&language)
.expect("no timer found")
@@ -1363,7 +1363,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
last_executed_job = Some(Instant::now());
}
}
let _timer = if *METRICS_ENABLED {
let _timer = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
Some(Instant::now())
} else {
None
@@ -1560,7 +1560,7 @@ fn build_language_metrics(
>,
language: &Option<ScriptLang>,
) -> Option<Metrics> {
let metrics = if *METRICS_ENABLED {
let metrics = if METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed) {
Some(Metrics {
worker_execution_failed: worker_execution_failed
.get(language)
docker-compose.yml
-7
View File
@@ -30,10 +30,7 @@ services:
- 8000
environment:
- DATABASE_URL=${DATABASE_URL}
- RUST_LOG=info
- MODE=server
## You can set the number of workers to 1 and not need any separate worker service but not recommended
- METRICS_ADDR=false # (ee only, if set to true, metrics will be exposed on port 8001)
depends_on:
db:
condition: service_healthy
@@ -50,10 +47,8 @@ services:
restart: unless-stopped
environment:
- DATABASE_URL=${DATABASE_URL}
- RUST_LOG=info
- MODE=worker
- KEEP_JOB_DIR=false
- METRICS_ADDR=false
- WORKER_GROUP=default
depends_on:
db:
@@ -78,10 +73,8 @@ services:
restart: unless-stopped
environment:
- DATABASE_URL=${DATABASE_URL}
- RUST_LOG=info
- MODE=worker
- WORKER_GROUP=native
- METRICS_ADDR=false # (ee only, if set to true, metrics will be exposed on port 8001)
depends_on:
db:
condition: service_healthy
frontend/src/lib/components/InstanceSettings.svelte
+8
View File
@@ -84,6 +84,14 @@
storage: 'setting',
ee_only:
'You can still set this setting by using NPM_CONFIG_REGISTRY as env variable to the worker containers'
},
{
label: 'Expose metrics',
description: 'Expose prometheus metrics for workers and servers on port 8001 at /metrics',
key: 'expose_metrics',
fieldType: 'boolean',
storage: 'setting',
ee_only: 'No workaround around this'
}
],
SMTP: [