feat: add otlp support (#4869)

* all

* all

* add otel

* add otel

* update docker-image

* update

* update

* update

* update

* update

* update

* update

* update

* update

* update

* update
This commit is contained in:
Ruben Fiszel
2024-12-07 16:29:20 +01:00
committed by GitHub
parent 46194d5944
commit 8a5ba115c6
30 changed files with 2133 additions and 1532 deletions
+2 -2
View File
@@ -138,7 +138,7 @@ jobs:
platforms: linux/amd64,linux/arm64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka,otel
tags: |
${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}-ee:${{ env.DEV_SHA }}
${{ steps.meta-ee-public.outputs.tags }}
@@ -200,7 +200,7 @@ jobs:
platforms: linux/amd64
push: true
build-args: |
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka
features=enterprise,enterprise_saml,stripe,embedding,parquet,prometheus,openidconnect,cloud,jemalloc,tantivy,deno_core,kafka,otel
PYTHON_IMAGE=python:3.12.2-slim-bookworm
tags: |
${{ steps.meta-ee-public-py312.outputs.tags }}
+185 -3
View File
@@ -4173,6 +4173,19 @@ dependencies = [
"tower-service",
]
[[package]]
name = "hyper-timeout"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3203a961e5c83b6f5498933e78b6b263e208c197b63e9c6c53cc82ffd3f63793"
dependencies = [
"hyper 1.5.1",
"hyper-util",
"pin-project-lite",
"tokio",
"tower-service",
]
[[package]]
name = "hyper-tls"
version = "0.5.0"
@@ -4809,7 +4822,7 @@ version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "674883a98273598ac3aad4301724c56734bea90574c5033af067e8f9fb5eb399"
dependencies = [
"prost",
"prost 0.12.6",
"prost-types",
]
@@ -5637,6 +5650,91 @@ dependencies = [
"vcpkg",
]
[[package]]
name = "opentelemetry"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f3cebff57f7dbd1255b44d8bddc2cebeb0ea677dbaa2e25a3070a91b318f660"
dependencies = [
"futures-core",
"futures-sink",
"js-sys",
"once_cell",
"pin-project-lite",
"thiserror 1.0.69",
]
[[package]]
name = "opentelemetry-appender-tracing"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ab5feffc321035ad94088a7e5333abb4d84a8726e54a802e736ce9dd7237e85b"
dependencies = [
"opentelemetry",
"tracing",
"tracing-core",
"tracing-subscriber",
]
[[package]]
name = "opentelemetry-otlp"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "91cf61a1868dacc576bf2b2a1c3e9ab150af7272909e80085c3173384fe11f76"
dependencies = [
"async-trait",
"futures-core",
"http 1.2.0",
"opentelemetry",
"opentelemetry-proto",
"opentelemetry_sdk",
"prost 0.13.3",
"thiserror 1.0.69",
"tokio",
"tonic",
"tracing",
]
[[package]]
name = "opentelemetry-proto"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a6e05acbfada5ec79023c85368af14abd0b307c015e9064d249b2a950ef459a6"
dependencies = [
"opentelemetry",
"opentelemetry_sdk",
"prost 0.13.3",
"tonic",
]
[[package]]
name = "opentelemetry-semantic-conventions"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bc1b6902ff63b32ef6c489e8048c5e253e2e4a803ea3ea7e783914536eb15c52"
[[package]]
name = "opentelemetry_sdk"
version = "0.27.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27b742c1cae4693792cc564e58d75a2a0ba29421a34a85b50da92efa89ecb2bc"
dependencies = [
"async-trait",
"futures-channel",
"futures-executor",
"futures-util",
"glob",
"once_cell",
"opentelemetry",
"percent-encoding",
"rand 0.8.5",
"serde_json",
"thiserror 1.0.69",
"tokio",
"tokio-stream",
"tracing",
]
[[package]]
name = "option-ext"
version = "0.2.0"
@@ -6264,7 +6362,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "deb1435c188b76130da55f17a466d252ff7b1418b2ad3e037d127b94e3411f29"
dependencies = [
"bytes",
"prost-derive",
"prost-derive 0.12.6",
]
[[package]]
name = "prost"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b0487d90e047de87f984913713b85c601c05609aad5b0df4b4573fbf69aa13f"
dependencies = [
"bytes",
"prost-derive 0.13.3",
]
[[package]]
@@ -6280,13 +6388,26 @@ dependencies = [
"syn 2.0.90",
]
[[package]]
name = "prost-derive"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9552f850d5f0964a4e4d0bf306459ac29323ddfbae05e35a7c0d35cb0803cc5"
dependencies = [
"anyhow",
"itertools 0.13.0",
"proc-macro2",
"quote",
"syn 2.0.90",
]
[[package]]
name = "prost-types"
version = "0.12.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9091c90b0a32608e984ff2fa4091273cbdd755d54935c51d520887f4a1dbd5b0"
dependencies = [
"prost",
"prost 0.12.6",
]
[[package]]
@@ -9448,6 +9569,36 @@ dependencies = [
"winnow 0.6.20",
]
[[package]]
name = "tonic"
version = "0.12.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
dependencies = [
"async-stream",
"async-trait",
"axum",
"base64 0.22.1",
"bytes",
"h2 0.4.7",
"http 1.2.0",
"http-body 1.0.1",
"http-body-util",
"hyper 1.5.1",
"hyper-timeout",
"hyper-util",
"percent-encoding",
"pin-project",
"prost 0.13.3",
"socket2",
"tokio",
"tokio-stream",
"tower 0.4.13",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
name = "toolchain_find"
version = "0.4.0"
@@ -9469,11 +9620,16 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
dependencies = [
"futures-core",
"futures-util",
"indexmap 1.9.3",
"pin-project",
"pin-project-lite",
"rand 0.8.5",
"slab",
"tokio",
"tokio-util",
"tower-layer",
"tower-service",
"tracing",
]
[[package]]
@@ -9651,6 +9807,24 @@ dependencies = [
"url",
]
[[package]]
name = "tracing-opentelemetry"
version = "0.28.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "97a971f6058498b5c0f1affa23e7ea202057a7301dbff68e968b2d578bcbd053"
dependencies = [
"js-sys",
"once_cell",
"opentelemetry",
"opentelemetry_sdk",
"smallvec",
"tracing",
"tracing-core",
"tracing-log 0.2.0",
"tracing-subscriber",
"web-time",
]
[[package]]
name = "tracing-serde"
version = "0.1.3"
@@ -10644,6 +10818,11 @@ dependencies = [
"magic-crypt",
"mail-send",
"object_store",
"opentelemetry",
"opentelemetry-appender-tracing",
"opentelemetry-otlp",
"opentelemetry-semantic-conventions",
"opentelemetry_sdk",
"pin-project-lite",
"prometheus",
"quick_cache",
@@ -10662,6 +10841,7 @@ dependencies = [
"tracing-appender",
"tracing-flame",
"tracing-loki",
"tracing-opentelemetry",
"tracing-subscriber",
"uuid 1.11.0",
"windmill-macros",
@@ -10897,6 +11077,7 @@ dependencies = [
"hmac",
"itertools 0.13.0",
"lazy_static",
"opentelemetry",
"prometheus",
"regex",
"reqwest 0.12.9",
@@ -10961,6 +11142,7 @@ dependencies = [
"object_store",
"once_cell",
"openidconnect",
"opentelemetry",
"pem 3.0.4",
"postgres-native-tls",
"prometheus",
+8 -1
View File
@@ -63,6 +63,7 @@ tantivy = ["dep:windmill-indexer", "windmill-api/tantivy"]
sqlx = ["windmill-worker/sqlx"]
deno_core = ["windmill-worker/deno_core", "dep:deno_core"]
kafka = ["windmill-api/kafka"]
otel = ["windmill-common/otel", "windmill-worker/otel"]
[dependencies]
anyhow.workspace = true
@@ -95,7 +96,6 @@ deno_core = { workspace = true, optional = true }
object_store = { workspace = true, optional = true }
quote.workspace = true
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemallocator = { optional = true, workspace = true }
tikv-jemalloc-sys = { optional = true, workspace = true }
@@ -279,6 +279,13 @@ tar = "^0"
http = "^1"
async-stream = "^0"
opentelemetry = "0.27.0"
tracing-opentelemetry = "0.28.0"
opentelemetry_sdk = { version = "*", features = ["rt-tokio"] }
opentelemetry-otlp = "0.27.0"
opentelemetry-appender-tracing = "0.27.0"
opentelemetry-semantic-conventions = "*"
tikv-jemallocator = { version = "0.5" }
tikv-jemalloc-sys = { version = "^0.5" }
tikv-jemalloc-ctl = { version = "^0.5" }
+1 -1
View File
@@ -1 +1 @@
5066da602260334767186e69ae5b6821feca0c71
ad89b5a1566159490eceb59c3e5c9416d9aa855c
+79 -68
View File
@@ -8,7 +8,7 @@
use anyhow::Context;
use monitor::{
reload_delete_logs_periodically_setting, reload_indexer_config,
load_otel, reload_delete_logs_periodically_setting, reload_indexer_config,
reload_timeout_wait_result_setting, send_current_log_file_to_object_store,
send_logs_to_object_store,
};
@@ -37,7 +37,7 @@ use windmill_common::{
EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING,
JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING,
LICENSE_KEY_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING,
OAUTH_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
OAUTH_SETTING, OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
},
@@ -221,11 +221,66 @@ async fn windmill_main() -> anyhow::Result<()> {
let hostname = hostname();
#[cfg(not(feature = "flamegraph"))]
let _guard = windmill_common::tracing_init::initialize_tracing(&hostname);
let mut enable_standalone_indexer: bool = false;
let mode = std::env::var("MODE")
.map(|x| x.to_lowercase())
.map(|x| {
if &x == "server" {
println!("Binary is in 'server' mode");
Mode::Server
} else if &x == "worker" {
tracing::info!("Binary is in 'worker' mode");
#[cfg(windows)]
{
println!("It is highly recommended to use the agent mode instead on windows (MODE=agent) and to pass a BASE_INTERNAL_URL");
}
Mode::Worker
} else if &x == "agent" {
println!("Binary is in 'agent' mode");
if std::env::var("BASE_INTERNAL_URL").is_err() {
panic!("BASE_INTERNAL_URL is required in agent mode")
}
if std::env::var("JOB_TOKEN").is_err() {
println!("JOB_TOKEN is not passed, hence workers will still need to create permissions for each job and the DATABASE_URL needs to be of a role that can INSERT into the job_perms table")
}
#[cfg(not(feature = "enterprise"))]
{
panic!("Agent mode is only available in the EE, ignoring...");
}
#[cfg(feature = "enterprise")]
Mode::Agent
} else if &x == "indexer" {
tracing::info!("Binary is in 'indexer' mode");
#[cfg(not(feature = "tantivy"))]
{
eprintln!("Cannot start the indexer because tantivy is not included in this binary/image. Make sure you are using the EE image if you want to access the full text search features.");
panic!("Indexer mode requires compiling with the tantivy feature flag.");
}
#[cfg(feature = "tantivy")]
Mode::Indexer
} else if &x == "standalone+search"{
enable_standalone_indexer = true;
println!("Binary is in 'standalone' mode with search enabled");
Mode::Standalone
}
else {
if &x != "standalone" {
eprintln!("mode not recognized, defaulting to standalone: {x}");
} else {
println!("Binary is in 'standalone' mode");
}
Mode::Standalone
}
})
.unwrap_or_else(|_| {
tracing::info!("Mode not specified, defaulting to standalone");
Mode::Standalone
});
#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
tracing::info!("jemalloc enabled");
println!("jemalloc enabled");
#[cfg(feature = "flamegraph")]
let _guard = windmill_common::tracing_init::setup_flamegraph();
@@ -236,13 +291,13 @@ async fn windmill_main() -> anyhow::Result<()> {
"cache" => {
#[cfg(feature = "embedding")]
{
tracing::info!("Caching embedding model...");
println!("Caching embedding model...");
windmill_api::embeddings::ModelInstance::load_model_files().await?;
tracing::info!("Cached embedding model");
println!("Cached embedding model");
}
#[cfg(not(feature = "embedding"))]
{
tracing::warn!("Embeddings are not enabled, ignoring...");
println!("Embeddings are not enabled, ignoring...");
}
cache_hub_scripts(std::env::args().nth(2)).await?;
@@ -256,64 +311,6 @@ async fn windmill_main() -> anyhow::Result<()> {
_ => {}
}
let mut enable_standalone_indexer: bool = false;
let mode = std::env::var("MODE")
.map(|x| x.to_lowercase())
.map(|x| {
if &x == "server" {
tracing::info!("Binary is in 'server' mode");
Mode::Server
} else if &x == "worker" {
tracing::info!("Binary is in 'worker' mode");
#[cfg(windows)]
{
tracing::warn!("It is highly recommended to use the agent mode instead on windows (MODE=agent) and to pass a BASE_INTERNAL_URL");
}
Mode::Worker
} else if &x == "agent" {
tracing::info!("Binary is in 'agent' mode");
if std::env::var("BASE_INTERNAL_URL").is_err() {
panic!("BASE_INTERNAL_URL is required in agent mode")
}
if std::env::var("JOB_TOKEN").is_err() {
tracing::warn!("JOB_TOKEN is not passed, hence workers will still need to create permissions for each job and the DATABASE_URL needs to be of a role that can INSERT into the job_perms table")
}
#[cfg(not(feature = "enterprise"))]
{
panic!("Agent mode is only available in the EE, ignoring...");
}
#[cfg(feature = "enterprise")]
Mode::Agent
} else if &x == "indexer" {
tracing::info!("Binary is in 'indexer' mode");
#[cfg(not(feature = "tantivy"))]
{
tracing::error!("Cannot start the indexer because tantivy is not included in this binary/image. Make sure you are using the EE image if you want to access the full text search features.");
panic!("Indexer mode requires compiling with the tantivy feature flag.");
}
#[cfg(feature = "tantivy")]
Mode::Indexer
} else if &x == "standalone+search"{
enable_standalone_indexer = true;
tracing::info!("Binary is in 'standalone' mode with search enabled");
Mode::Standalone
}
else {
if &x != "standalone" {
tracing::error!("mode not recognized, defaulting to standalone: {x}");
} else {
tracing::info!("Binary is in 'standalone' mode");
}
Mode::Standalone
}
})
.unwrap_or_else(|_| {
tracing::info!("Mode not specified, defaulting to standalone");
Mode::Standalone
});
#[allow(unused_mut)]
let mut num_workers = if mode == Mode::Server || mode == Mode::Indexer {
0
@@ -325,7 +322,7 @@ async fn windmill_main() -> anyhow::Result<()> {
};
if num_workers > 1 {
tracing::warn!(
println!(
"We STRONGLY recommend using at most 1 worker per container, use at your own risks"
);
}
@@ -347,10 +344,15 @@ async fn windmill_main() -> anyhow::Result<()> {
IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))
};
tracing::info!("Connecting to database...");
println!("Connecting to database...");
let db = windmill_common::connect_db(server_mode, indexer_mode).await?;
load_otel(&db).await;
tracing::info!("Database connected");
#[cfg(not(feature = "flamegraph"))]
let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode);
let num_version = sqlx::query_scalar!("SELECT version()").fetch_one(&db).await;
tracing::info!(
@@ -776,6 +778,15 @@ Windmill Community Edition {GIT_VERSION}
tracing::error!(error = %e, "Could not reload debug metrics setting");
}
},
OTEL_SETTING => {
tracing::info!("OTEL setting changed, restarting");
// we wait a bit randomly to avoid having all servers 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");
}
},
REQUEST_SIZE_LIMIT_SETTING => {
if server_mode {
tracing::info!("Request limit size change detected, killing server expecting to be restarted");
+63 -3
View File
@@ -12,7 +12,7 @@ use std::{
use chrono::{NaiveDateTime, Utc};
use futures::{stream::FuturesUnordered, StreamExt};
use serde::de::DeserializeOwned;
use serde::{de::DeserializeOwned, Deserializer};
use sqlx::{Pool, Postgres};
use tokio::{
join,
@@ -40,7 +40,7 @@ use windmill_common::{
EXTRA_PIP_INDEX_URL_SETTING, HUB_BASE_URL_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING,
JWT_SECRET_SETTING, KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING,
MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NPM_CONFIG_REGISTRY_SETTING, OAUTH_SETTING,
PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
OTEL_SETTING, PIP_INDEX_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING,
REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING,
SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, TIMEOUT_WAIT_RESULT_SETTING,
},
@@ -58,7 +58,8 @@ use windmill_common::{
},
BASE_URL, CRITICAL_ALERT_MUTE_UI_ENABLED, CRITICAL_ERROR_CHANNELS, DB, DEFAULT_HUB_BASE_URL,
HUB_BASE_URL, JOB_RETENTION_SECS, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
MONITOR_LOGS_ON_OBJECT_STORE, SERVICE_LOG_RETENTION_SECS,
MONITOR_LOGS_ON_OBJECT_STORE, OTEL_LOGS_ENABLED, OTEL_METRICS_ENABLED, OTEL_TRACING_ENABLED,
SERVICE_LOG_RETENTION_SECS,
};
use windmill_queue::cancel_job;
use windmill_worker::{
@@ -199,6 +200,65 @@ pub async fn load_metrics_enabled(db: &DB) -> error::Result<()> {
Ok(())
}
fn empty_string_as_none<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
where
D: Deserializer<'de>,
{
let option = <Option<String> as serde::Deserialize>::deserialize(deserializer)?;
Ok(option.filter(|s| !s.is_empty()))
}
#[derive(serde::Deserialize)]
struct OtelSetting {
metrics_enabled: Option<bool>,
logs_enabled: Option<bool>,
tracing_enabled: Option<bool>,
#[serde(default, deserialize_with = "empty_string_as_none")]
otel_exporter_otlp_endpoint: Option<String>,
#[serde(default, deserialize_with = "empty_string_as_none")]
otel_exporter_otlp_headers: Option<String>,
#[serde(default, deserialize_with = "empty_string_as_none")]
otel_exporter_otlp_protocol: Option<String>,
#[serde(default, deserialize_with = "empty_string_as_none")]
otel_exporter_otlp_compression: Option<String>,
}
pub async fn load_otel(db: &DB) {
let otel = load_value_from_global_settings(db, OTEL_SETTING).await;
if let Ok(v) = otel {
if let Some(v) = v {
let deser = serde_json::from_value::<OtelSetting>(v);
if let Ok(o) = deser {
let metrics_enabled = o.metrics_enabled.unwrap_or(false);
let logs_enabled = o.logs_enabled.unwrap_or(false);
let tracing_enabled = o.tracing_enabled.unwrap_or(false);
OTEL_METRICS_ENABLED.store(metrics_enabled, Ordering::Relaxed);
OTEL_LOGS_ENABLED.store(logs_enabled, Ordering::Relaxed);
OTEL_TRACING_ENABLED.store(tracing_enabled, Ordering::Relaxed);
if let Some(endpoint) = o.otel_exporter_otlp_endpoint.as_ref() {
std::env::set_var("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
}
if let Some(headers) = o.otel_exporter_otlp_headers.as_ref() {
std::env::set_var("OTEL_EXPORTER_OTLP_HEADERS", headers);
}
if let Some(protocol) = o.otel_exporter_otlp_protocol {
std::env::set_var("OTEL_EXPORTER_OTLP_PROTOCOL", protocol);
}
if let Some(compression) = o.otel_exporter_otlp_compression {
std::env::set_var("OTEL_EXPORTER_OTLP_COMPRESSION", compression);
}
tracing::info!("OTEL settings loaded: tracing ({tracing_enabled}), logs ({logs_enabled}), metrics ({metrics_enabled}), endpoint ({:?}), headers defined: ({})",
o.otel_exporter_otlp_endpoint, o.otel_exporter_otlp_headers.is_some());
} else {
tracing::error!("Error deserializing otel settings");
}
}
} else {
tracing::error!("Error loading otel settings: {}", otel.unwrap_err());
}
}
pub async fn load_tag_per_workspace_enabled(db: &DB) -> error::Result<()> {
let metrics_enabled =
load_value_from_global_settings(db, DEFAULT_TAGS_PER_WORKSPACE_SETTING).await;
+4 -1
View File
@@ -77,7 +77,10 @@ async fn initialize_tracing() {
static ONCE: Once = Once::new();
ONCE.call_once(|| {
let _ = windmill_common::tracing_init::initialize_tracing("test");
let _ = windmill_common::tracing_init::initialize_tracing(
"test",
&windmill_common::utils::Mode::Standalone,
);
});
}
+13 -5
View File
@@ -29,11 +29,19 @@ impl<B> OnResponse<B> for MyOnResponse {
_span: &tracing::Span,
) {
if *LOG_REQUESTS {
tracing::info!(
latency = latency.as_millis(),
status = response.status().as_u16(),
"response"
)
if response.status().is_success() {
tracing::info!(
latency = latency.as_millis(),
status = response.status().as_u16(),
"response"
)
} else {
tracing::error!(
latency = latency.as_millis(),
status = response.status().as_u16(),
"response"
)
}
}
}
}
+8
View File
@@ -13,6 +13,7 @@ flamegraph = ["dep:tracing-flame"]
loki = ["dep:tracing-loki"]
benchmark = []
parquet = ["dep:object_store", "dep:aws-config", "dep:aws-sdk-sts"]
otel = ["dep:opentelemetry-semantic-conventions", "dep:opentelemetry-otlp", "dep:opentelemetry_sdk", "dep:opentelemetry", "dep:tracing-opentelemetry", "dep:opentelemetry-appender-tracing"]
[lib]
name = "windmill_common"
@@ -64,5 +65,12 @@ croner = "2.0.6"
quick_cache.workspace = true
pin-project-lite.workspace = true
opentelemetry-semantic-conventions = { workspace = true, optional = true }
opentelemetry-otlp = { workspace = true, optional = true }
opentelemetry_sdk = { workspace = true, optional = true }
opentelemetry = { workspace = true, optional = true }
tracing-opentelemetry = { workspace = true, optional = true }
opentelemetry-appender-tracing = { workspace = true, optional = true }
[target.'cfg(not(target_env = "msvc"))'.dependencies]
tikv-jemalloc-ctl = { optional = true, workspace = true }
@@ -35,8 +35,9 @@ pub const CRITICAL_ALERT_MUTE_UI_SETTING: &str = "critical_alert_mute_ui";
pub const DEV_INSTANCE_SETTING: &str = "dev_instance";
pub const JWT_SECRET_SETTING: &str = "jwt_secret";
pub const EMAIL_DOMAIN_SETTING: &str = "email_domain";
pub const OTEL_SETTING: &str = "otel";
pub const ENV_SETTINGS: [&str; 51] = [
pub const ENV_SETTINGS: [&str; 54] = [
"DISABLE_NSJAIL",
"MODE",
"NUM_WORKERS",
@@ -88,4 +89,7 @@ pub const ENV_SETTINGS: [&str; 51] = [
"WORKER_GROUP",
"SAML_METADATA",
"INSTANCE_IS_DEV",
"OTEL_METRICS",
"OTEL_TRACING",
"OTEL_LOGS",
];
+8 -2
View File
@@ -36,20 +36,20 @@ pub mod job_s3_helpers_ee;
pub mod jobs;
pub mod more_serde;
pub mod oauth2;
pub mod otel_ee;
pub mod queue;
pub mod s3_helpers;
pub mod schedule;
pub mod scripts;
pub mod server;
pub mod stats_ee;
pub mod tracing_init;
pub mod users;
pub mod utils;
pub mod variables;
pub mod worker;
pub mod workspaces;
pub mod tracing_init;
pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;
pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5;
@@ -87,6 +87,12 @@ lazy_static::lazy_static! {
.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 OTEL_METRICS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_METRICS").is_ok());
pub static ref OTEL_TRACING_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_TRACING").is_ok());
pub static ref OTEL_LOGS_ENABLED: AtomicBool = AtomicBool::new(std::env::var("OTEL_LOGS").is_ok());
pub static ref METRICS_DEBUG_ENABLED: AtomicBool = AtomicBool::new(false);
pub static ref CRITICAL_ALERT_MUTE_UI_ENABLED: AtomicBool = AtomicBool::new(false);
+54
View File
@@ -0,0 +1,54 @@
/*
* Author: Ruben Fiszel
* Copyright: Windmill Labs, Inc 2022
* This file and its contents are licensed under the AGPLv3 License.
* Please see the included NOTICE for copyright information and
* LICENSE-AGPL for a copy of the license.
*/
use crate::{jobs::QueuedJob, utils::Mode};
use uuid::Uuid;
pub fn set_span_parent(_span: &tracing::Span, _rj: &Uuid) {}
#[cfg(not(all(feature = "otel", feature = "enterprise")))]
pub(crate) type OtelProvider = Option<()>;
#[cfg(all(feature = "otel", feature = "enterprise"))]
pub(crate) type OtelProvider = Option<opentelemetry_sdk::metrics::SdkMeterProvider>;
#[cfg(not(feature = "otel"))]
pub fn otel_ctx() -> () {}
#[cfg(feature = "otel")]
#[inline(always)]
pub fn otel_ctx() -> opentelemetry::Context {
opentelemetry::Context::current()
}
#[cfg(not(feature = "otel"))]
impl<T: Sized> FutureExt for T {}
#[cfg(not(feature = "otel"))]
pub trait FutureExt: Sized {
fn with_context(self, _otel_cx: ()) -> Self {
self
}
}
use tracing_subscriber::EnvFilter;
pub(crate) fn init_logs_bridge(_mode: &Mode) -> Option<EnvFilter> {
None
}
#[cfg(all(feature = "otel", feature = "enterprise"))]
pub(crate) fn init_otlp_tracer(_mode: &Mode) -> Option<opentelemetry_sdk::trace::Tracer> {
None
}
pub(crate) fn init_meter_provider(_mode: &Mode) -> OtelProvider {
None
}
pub fn add_root_flow_job_to_otlp(_queued_job: &QueuedJob, _success: bool) {}
+57 -37
View File
@@ -7,13 +7,23 @@
*/
use const_format::concatcp;
use std::{
collections::HashMap,
sync::{Arc, RwLock},
};
use tracing::Event;
use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard};
use tracing_subscriber::layer::Context;
use tracing_subscriber::{
filter::Targets,
fmt::{format, Layer},
prelude::*,
EnvFilter,
};
use crate::utils::Mode;
fn json_layer<S>() -> Layer<S, format::JsonFields, format::Format<format::Json>> {
tracing_subscriber::fmt::layer()
.json()
@@ -34,7 +44,10 @@ pub const LOGS_SERVICE: &str = "logs/services/";
pub const TMP_WINDMILL_LOGS_SERVICE: &str = concatcp!("/tmp/windmill/", LOGS_SERVICE);
pub fn initialize_tracing(hostname: &str) -> WorkerGuard {
pub fn initialize_tracing(
hostname: &str,
mode: &Mode,
) -> (WorkerGuard, crate::otel_ee::OtelProvider) {
let style = std::env::var("RUST_LOG_STYLE").unwrap_or_else(|_| "auto".into());
if std::env::var("RUST_LOG").is_ok_and(|x| x == "debug" || x == "info") {
@@ -44,7 +57,17 @@ pub fn initialize_tracing(hostname: &str) -> WorkerGuard {
)
}
let env_filter = EnvFilter::from_default_env();
let meter_provider = crate::otel_ee::init_meter_provider(mode);
#[cfg(all(feature = "otel", feature = "enterprise"))]
let opentelemetry = crate::otel_ee::init_otlp_tracer(mode)
.map(|x| tracing_opentelemetry::layer().with_tracer(x));
#[cfg(not(all(feature = "otel", feature = "enterprise")))]
let opentelemetry: Option<EnvFilter> = None;
let logs_bridge = crate::otel_ee::init_logs_bridge(&mode);
use tracing_appender::rolling::{RollingFileAppender, Rotation};
let log_dir = format!("{}/{}/", TMP_WINDMILL_LOGS_SERVICE, hostname);
@@ -61,39 +84,59 @@ pub fn initialize_tracing(hostname: &str) -> WorkerGuard {
.finish(file_appender);
let stdout_and_log_file_writer = std::io::stdout.and(log_file_writer);
let ts_base = tracing_subscriber::registry().with(env_filter);
// let job_logs_filter = tracing_subscriber::filter::Targets::new()
// .with_target("windmill:job_log", tracing::Level::TRACE);
#[cfg(feature = "loki")]
let ts_base = {
let (layer, task) = tracing_loki::builder()
.build_url(reqwest::Url::parse("http://127.0.0.1:3100").unwrap())
.expect("build loki url");
tokio::spawn(task);
ts_base.with(layer)
};
let env_filter = EnvFilter::builder()
.with_default_directive(tracing::level_filters::LevelFilter::ERROR.into())
.from_env_lossy();
let ts_base = tracing_subscriber::registry().with(env_filter);
match *JSON_FMT {
true => ts_base
.with(logs_bridge)
.with(opentelemetry)
// .with(env_filter2.add_directive("windmill:job_log=off".parse().unwrap()))
.with(
json_layer()
.with_writer(stdout_and_log_file_writer)
.flatten_event(true),
.flatten_event(true)
.with_filter(
Targets::new()
.with_target(
"windmill:job_log",
tracing::level_filters::LevelFilter::OFF,
)
.with_default(tracing::level_filters::LevelFilter::INFO),
),
)
.with(CountingLayer::new())
.init(),
false => ts_base
.with(logs_bridge)
.with(opentelemetry)
// .with(env_filter2.add_directive("windmill:job_log=off".parse().unwrap()))
.with(
compact_layer()
.with_writer(stdout_and_log_file_writer)
.with_ansi(style.to_lowercase() != "never")
.with_file(true)
.with_line_number(true)
.with_target(false),
.with_target(false)
.with_filter(
Targets::new()
.with_target(
"windmill:job_log",
tracing::level_filters::LevelFilter::OFF,
)
.with_default(tracing::level_filters::LevelFilter::INFO),
),
)
.with(CountingLayer::new())
.init(),
}
_guard
(_guard, meter_provider)
}
#[cfg(feature = "flamegraph")]
@@ -112,13 +155,6 @@ pub fn setup_flamegraph() -> impl Drop {
_guard
}
use std::{
collections::HashMap,
sync::{Arc, RwLock},
};
use tracing::Event;
use tracing_subscriber::layer::Context;
lazy_static::lazy_static! {
pub static ref LOG_COUNTING_BY_MIN: Arc<RwLock<HashMap<String, LogCounter>>> = Arc::new(RwLock::new(HashMap::new()));
}
@@ -144,22 +180,6 @@ impl CountingLayer {
}
}
// impl CountingLayer {
// pub fn new() -> Self {
// CountingLayer { counter: Arc::new(Mutex::new(LogCounter::new())) }
// }
// pub fn get_counts(&self) -> (usize, usize) {
// let counter = self.counter.lock().unwrap();
// (counter.non_error_count, counter.error_count)
// }
// pub fn reset_counts(&self) {
// let mut counter = self.counter.lock().unwrap();
// counter.reset();
// }
// }
pub const LOG_TIMESTAMP_FMT: &str = "%Y-%m-%d-%H-%M";
impl<S> tracing_subscriber::Layer<S> for CountingLayer
+2 -1
View File
@@ -43,4 +43,5 @@ bigdecimal.workspace = true
axum.workspace = true
serde_urlencoded.workspace = true
regex.workspace = true
backon.workspace = true
backon.workspace = true
opentelemetry.workspace = true
-8
View File
@@ -32,7 +32,6 @@ use sqlx::{types::Json, FromRow, Pool, Postgres, Transaction};
#[cfg(feature = "benchmark")]
use std::time::Instant;
use tokio::{sync::RwLock, time::sleep};
use tracing::{instrument, Instrument};
use ulid::Ulid;
use uuid::Uuid;
use windmill_audit::audit_ee::{audit_log, AuditAuthor};
@@ -454,7 +453,6 @@ where
}
}
#[instrument(level = "trace", skip_all)]
pub async fn add_completed_job_error(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
@@ -510,7 +508,6 @@ lazy_static::lazy_static! {
pub static ref GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE: Option<String> = std::env::var("GLOBAL_ERROR_HANDLER_PATH_IN_ADMINS_WORKSPACE").ok();
}
#[instrument(level = "trace", skip_all, name = "add_completed_job")]
pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
db: &Pool<Postgres>,
queued_job: &QueuedJob,
@@ -643,9 +640,7 @@ pub async fn add_completed_job<T: Serialize + Send + Sync + ValidableJson>(
.fetch_one(&mut *tx)
.await
.map_err(|e| Error::InternalErr(format!("Could not add completed job {job_id}: {e:#}")))?;
// tracing::error!("2 {:?}", start.elapsed());
// add_time!(bench, "add_completed_job query END");
if !queued_job.is_flow_step {
if _duration > 500
@@ -1259,7 +1254,6 @@ pub async fn send_error_to_workspace_handler<'a, 'c, T: Serialize + Send + Sync>
Ok(())
}
#[instrument(level = "trace", skip_all)]
pub async fn handle_maybe_scheduled_job<'c>(
db: &Pool<Postgres>,
job: &QueuedJob,
@@ -2429,7 +2423,6 @@ async fn extract_result_from_job_result(
}
}
#[instrument(level = "trace", skip_all)]
pub async fn delete_job<'c>(
mut tx: Transaction<'c, Postgres>,
w_id: &str,
@@ -4039,7 +4032,6 @@ pub async fn push<'c, 'd>(
script_path.as_ref().map(|x| x.as_str()),
Some(hm),
)
.instrument(tracing::info_span!("job_run", email = &email))
.await?;
}
+4
View File
@@ -19,6 +19,7 @@ flow_testing = []
cloud = []
sqlx = []
deno_core = ["dep:deno_fetch", "dep:deno_webidl", "dep:deno_web", "dep:deno_net", "dep:deno_console", "dep:deno_url", "dep:deno_core", "dep:deno_ast", "dep:deno_tls"]
otel = ["windmill-common/otel", "dep:opentelemetry"]
[dependencies]
windmill-queue.workspace = true
@@ -92,6 +93,9 @@ yaml-rust.workspace = true
swc_ecma_parser.workspace = true
backon.workspace = true
opentelemetry = { workspace = true, optional = true }
[build-dependencies]
deno_fetch = { workspace = true, optional = true }
deno_webidl = { workspace = true, optional = true }
+5 -5
View File
@@ -866,11 +866,11 @@ fn tentatively_improve_error(err: Error, executable: &str) -> Error {
#[cfg(windows)]
let err_msg = "program not found";
if err
.to_string()
.contains(&err_msg)
{
return Error::InternalErr(format!("Executable {executable} not found on worker. PATH: {}", *PATH_ENV));
if err.to_string().contains(&err_msg) {
return Error::InternalErr(format!(
"Executable {executable} not found on worker. PATH: {}",
*PATH_ENV
));
}
return err;
}
+6 -9
View File
@@ -87,7 +87,7 @@ async fn kill_process_tree(pid: Option<u32>) -> Result<(), String> {
/// - update the `last_line` and `logs` strings with the program output
/// - update "queue"."last_ping" every five seconds
/// - kill process if we exceed timeout or "queue"."canceled" is set
#[tracing::instrument(level = "trace", skip_all)]
#[tracing::instrument(name="run_subprocess", level = "info", skip_all, fields(otel.name = %child_name))]
pub async fn handle_child(
job_id: &Uuid,
db: &Pool<Postgres>,
@@ -435,8 +435,6 @@ pub async fn handle_child(
}
}
async fn get_mem_peak(pid: Option<u32>, nsjail: bool) -> i32 {
if pid.is_none() {
return -1;
@@ -692,7 +690,10 @@ fn child_joined_output_stream(
let stdout = BufReader::new(stdout).lines();
let stderr = BufReader::new(stderr).lines();
stream::select(lines_to_stream(stderr, true), lines_to_stream(stdout, false))
stream::select(
lines_to_stream(stderr, true),
lines_to_stream(stdout, false),
)
}
pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
@@ -702,14 +703,10 @@ pub fn lines_to_stream<R: tokio::io::AsyncBufRead + Unpin>(
stream::poll_fn(move |cx| {
std::pin::Pin::new(&mut lines)
.poll_next_line(cx)
.map(|result| {
process_streaming_log_lines(result, stderr)
})
.map(|result| process_streaming_log_lines(result, stderr))
})
}
pub fn process_status(status: ExitStatus) -> error::Result<()> {
if status.success() {
Ok(())
+92 -27
View File
@@ -1,3 +1,6 @@
#[cfg(feature = "otel")]
use opentelemetry::trace::FutureExt;
use serde::Serialize;
use sqlx::{types::Json, Pool, Postgres};
use std::{
@@ -7,6 +10,9 @@ use std::{
Arc,
},
};
use tracing::{field, Instrument};
#[cfg(not(feature = "otel"))]
use windmill_common::otel_ee::FutureExt;
use uuid::Uuid;
@@ -85,7 +91,49 @@ pub fn start_background_processor(
JobKind::Dependencies | JobKind::FlowDependencies
);
handle_receive_completed_job(
let success = jc.success;
let span = tracing::span!(
tracing::Level::INFO,
"job_postprocessing",
job_id = %jc.job.id, root_job = field::Empty, workspace_id = %jc.job.workspace_id, worker = %worker_name,tag = %jc.job.tag,
// hostname = %hostname,
language = field::Empty,
script_path = field::Empty,
flow_step_id = field::Empty,
parent_job = field::Empty,
otel.name = field::Empty
);
let rj = if let Some(root_job) = jc.job.root_job {
root_job
} else {
jc.job.id
};
windmill_common::otel_ee::set_span_parent(&span, &rj);
if let Some(lg) = jc.job.language.as_ref() {
span.record("language", lg.as_str());
}
if let Some(step_id) = jc.job.flow_step_id.as_ref() {
span.record(
"otel.name",
format!("job_postprocessing {}", step_id).as_str(),
);
span.record("flow_step_id", step_id.as_str());
} else {
span.record("otel.name", "job postprocessing");
}
if let Some(parent_job) = jc.job.parent_job.as_ref() {
span.record("parent_job", parent_job.to_string().as_str());
}
if let Some(script_path) = jc.job.script_path.as_ref() {
span.record("script_path", script_path.as_str());
}
if let Some(root_job) = jc.job.root_job.as_ref() {
span.record("root_job", root_job.to_string().as_str());
}
let root_job = handle_receive_completed_job(
jc,
&base_internal_url,
&db,
@@ -96,8 +144,14 @@ pub fn start_background_processor(
#[cfg(feature = "benchmark")]
&mut bench,
)
.instrument(span)
.await;
if let Some(root_job) = root_job {
windmill_common::otel_ee::add_root_flow_job_to_otlp(&root_job, success);
tracing::error!(job_id = %root_job.id, parent_job = ?root_job.parent_job, "ADDDED root job completed");
}
if is_init_script_and_failure {
tracing::error!("init script errored, exiting");
killpill_tx.send(()).unwrap_or_default();
@@ -199,7 +253,11 @@ async fn send_job_completed(
token,
duration,
};
job_completed_tx.send(jc).await.expect("send job completed")
job_completed_tx
.send(jc)
.with_context(windmill_common::otel_ee::otel_ctx())
.await
.expect("send job completed")
}
pub async fn process_result(
@@ -271,6 +329,7 @@ pub async fn process_result(
token,
duration,
)
.with_context(windmill_common::otel_ee::otel_ctx())
.await;
Ok(true)
}
@@ -315,6 +374,7 @@ pub async fn process_result(
token,
duration,
)
.with_context(windmill_common::otel_ee::otel_ctx())
.await;
Ok(false)
}
@@ -330,7 +390,7 @@ pub async fn handle_receive_completed_job(
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) {
) -> Option<Arc<QueuedJob>> {
let token = jc.token.clone();
let workspace = jc.job.workspace_id.clone();
let client = AuthedClient {
@@ -342,7 +402,7 @@ pub async fn handle_receive_completed_job(
let job = jc.job.clone();
let mem_peak = jc.mem_peak.clone();
let canceled_by = jc.canceled_by.clone();
if let Err(err) = process_completed_job(
match process_completed_job(
jc,
&client,
db,
@@ -355,26 +415,29 @@ pub async fn handle_receive_completed_job(
)
.await
{
handle_job_error(
db,
&client,
job.as_ref(),
mem_peak,
canceled_by,
err,
false,
same_worker_tx.clone(),
&worker_dir,
worker_name,
job_completed_tx,
#[cfg(feature = "benchmark")]
bench,
)
.await;
Err(err) => {
handle_job_error(
db,
&client,
job.as_ref(),
mem_peak,
canceled_by,
err,
false,
same_worker_tx.clone(),
&worker_dir,
worker_name,
job_completed_tx,
#[cfg(feature = "benchmark")]
bench,
)
.await;
None
}
Ok(r) => r,
}
}
#[tracing::instrument(name = "completed_job", level = "info", skip_all, fields(job_id = %job.id))]
pub async fn process_completed_job(
JobCompleted { job, result, mem_peak, success, cached_res_path, canceled_by, duration, .. }: JobCompleted,
client: &AuthedClient,
@@ -384,7 +447,7 @@ pub async fn process_completed_job(
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> windmill_common::error::Result<()> {
) -> windmill_common::error::Result<Option<Arc<QueuedJob>>> {
if success {
// println!("bef completed job{:?}", SystemTime::now());
if let Some(cached_path) = cached_res_path {
@@ -414,8 +477,8 @@ pub async fn process_completed_job(
if is_flow_step {
if let Some(parent_job) = parent_job {
tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)");
update_flow_status_after_job_completion(
// tracing::info!(parent_flow = %parent_job, subflow = %job_id, "updating flow status (2)");
let r = update_flow_status_after_job_completion(
db,
client,
parent_job,
@@ -434,9 +497,10 @@ pub async fn process_completed_job(
)
.warn_after_seconds(10)
.await?;
add_time!(bench, "updated flow status END");
return Ok(r);
}
}
add_time!(bench, "updated flow status END");
} else {
let result = add_completed_job_error(
db,
@@ -454,7 +518,7 @@ pub async fn process_completed_job(
if job.is_flow_step {
if let Some(parent_job) = job.parent_job {
tracing::error!(parent_flow = %parent_job, subflow = %job.id, "process completed job error, updating flow status");
update_flow_status_after_job_completion(
let r = update_flow_status_after_job_completion(
db,
client,
parent_job,
@@ -473,10 +537,11 @@ pub async fn process_completed_job(
)
.warn_after_seconds(10)
.await?;
return Ok(r);
}
}
}
Ok(())
return Ok(None);
}
#[tracing::instrument(name = "job_error", level = "info", skip_all, fields(job_id = %job.id))]
+105 -48
View File
@@ -6,6 +6,10 @@
* LICENSE-AGPL for a copy of the license.
*/
// #[cfg(feature = "otel")]
// use opentelemetry::{global, KeyValue};
use windmill_common::{
apps::AppScriptId,
auth::{fetch_authed_from_permissioned_as, JWTAuthClaims, JobPerms, JWT_SECRET},
@@ -25,7 +29,7 @@ use const_format::concatcp;
#[cfg(feature = "prometheus")]
use prometheus::IntCounter;
use tracing::Instrument;
use tracing::{field, Instrument};
#[cfg(feature = "prometheus")]
use windmill_common::METRICS_DEBUG_ENABLED;
#[cfg(feature = "prometheus")]
@@ -88,30 +92,12 @@ use tokio::{
use rand::Rng;
use crate::{
ansible_executor::handle_ansible_job,
bash_executor::{handle_bash_job, handle_powershell_job},
bun_executor::handle_bun_job,
common::{
ansible_executor::handle_ansible_job, bash_executor::{handle_bash_job, handle_powershell_job}, bun_executor::handle_bun_job, common::{
build_args_map, get_cached_resource_value_if_valid, get_reserved_variables, hash_args,
update_worker_ping_for_failed_init_script, OccupancyMetrics,
},
deno_executor::handle_deno_job,
go_executor::handle_go_job,
graphql_executor::do_graphql,
handle_child::SLOW_LOGS,
handle_job_error,
job_logger::NO_LOGS_AT_ALL,
js_eval::{eval_fetch_timeout, transpile_ts},
mysql_executor::do_mysql,
pg_executor::do_postgresql,
php_executor::handle_php_job,
python_executor::handle_python_job,
result_processor::{process_result, start_background_processor},
rust_executor::handle_rust_job,
worker_flow::{handle_flow, update_flow_status_in_progress, Step},
worker_lockfiles::{
}, deno_executor::handle_deno_job, go_executor::handle_go_job, graphql_executor::do_graphql, handle_child::SLOW_LOGS, handle_job_error, job_logger::NO_LOGS_AT_ALL, js_eval::{eval_fetch_timeout, transpile_ts}, mysql_executor::do_mysql, pg_executor::do_postgresql, php_executor::handle_php_job, python_executor::handle_python_job, result_processor::{process_result, start_background_processor}, rust_executor::handle_rust_job, worker_flow::{handle_flow, update_flow_status_in_progress, Step}, worker_lockfiles::{
handle_app_dependency_job, handle_dependency_job, handle_flow_dependency_job,
},
}
};
use backon::ConstantBuilder;
@@ -720,7 +706,10 @@ fn add_outstanding_wait_time(
}.in_current_span());
}
#[tracing::instrument(name = "worker", level = "info", skip_all, fields(worker = %worker_name, hostname = %hostname))]
// struct WorkerMtrics {
// job_
// }
pub async fn run_worker(
db: &Pool<Postgres>,
hostname: &str,
@@ -736,6 +725,7 @@ pub async fn run_worker(
#[cfg(not(feature = "enterprise"))]
if !*DISABLE_NSJAIL {
tracing::warn!(
worker = %worker_name, hostname = %hostname,
"NSJAIL to sandbox process in untrusted environments is an enterprise feature but allowed to be used for testing purposes"
);
}
@@ -743,10 +733,10 @@ pub async fn run_worker(
let start_time = Instant::now();
let worker_dir = format!("{TMP_DIR}/{worker_name}");
tracing::debug!(worker_dir = %worker_dir, "Creating worker dir");
tracing::debug!(worker = %worker_name, hostname = %hostname, worker_dir = %worker_dir, "Creating worker dir");
if let Some(ref netrc) = *NETRC {
tracing::info!("Writing netrc at {}/.netrc", HOME_ENV.as_str());
tracing::info!(worker = %worker_name, hostname = %hostname, "Writing netrc at {}/.netrc", HOME_ENV.as_str());
write_file(&HOME_ENV, ".netrc", netrc).expect("could not write netrc");
}
@@ -961,6 +951,16 @@ pub async fn run_worker(
None
};
// let worker_resource = &[
// KeyValue::new("hostname", hostname.to_string()),
// KeyValue::new("worker", worker_name.to_string()),
// ];
// // Create a meter from the above MeterProvider.
// let meter = global::meter("windmill");
// let counter = meter.u64_counter("jobs.execution").build();
let mut occupancy_metrics = OccupancyMetrics::new(start_time);
let mut jobs_executed = 0;
@@ -1016,6 +1016,7 @@ pub async fn run_worker(
IS_READY.store(true, Ordering::Relaxed);
tracing::info!(
worker = %worker_name, hostname = %hostname,
"listening for jobs, WORKER_GROUP: {}, config: {:?}",
*WORKER_GROUP,
WORKER_CONFIG.read().await
@@ -1051,7 +1052,7 @@ pub async fn run_worker(
if i_worker == 1 {
if let Err(e) = queue_init_bash_maybe(db, same_worker_tx.clone(), &worker_name).await {
killpill_tx.send(()).unwrap_or_default();
tracing::error!("Error queuing init bash script for worker {worker_name}: {e:#}");
tracing::error!(worker = %worker_name, hostname = %hostname, "Error queuing init bash script for worker {worker_name}: {e:#}");
return;
}
}
@@ -1084,7 +1085,7 @@ pub async fn run_worker(
#[cfg(feature = "enterprise")]
{
if let Ok(_) = killpill_rx.try_recv() {
tracing::info!("killpill received on worker waiting for valid key");
tracing::info!(worker = %worker_name, hostname = %hostname, "killpill received on worker waiting for valid key");
job_completed_tx
.0
.send(SendResult::Kill)
@@ -1096,6 +1097,7 @@ pub async fn run_worker(
if !valid_key {
tracing::error!(
worker = %worker_name, hostname = %hostname,
"Invalid license key, workers require a valid license key, sleeping for 30s waiting for valid key to be set"
);
tokio::time::sleep(Duration::from_secs(10)).await;
@@ -1109,7 +1111,7 @@ pub async fn run_worker(
#[cfg(feature = "prometheus")]
if let Some(wk) = worker_busy.as_ref() {
wk.set(0);
tracing::debug!("set worker busy to 0");
tracing::debug!(worker = %worker_name, hostname = %hostname, "set worker busy to 0");
}
occupancy_metrics.running_job_started_at = None;
@@ -1121,7 +1123,7 @@ pub async fn run_worker(
.try_into()
.unwrap(),
);
tracing::debug!("set uptime metric");
tracing::debug!(worker = %worker_name, hostname = %hostname, "set uptime metric");
}
if last_ping.elapsed().as_secs() > NUM_SECS_PING {
@@ -1165,15 +1167,19 @@ pub async fn run_worker(
)
.notify(|err, dur| {
tracing::error!(
worker = %worker_name, hostname = %hostname,
"retrying updating worker ping in {dur:#?}, err: {err:#?}"
);
})
.sleep(tokio::time::sleep)
.await {
tracing::error!("failed to update worker ping, exiting: {}", e);
tracing::error!(
worker = %worker_name, hostname = %hostname,
"failed to update worker ping, exiting: {}", e);
killpill_tx.send(()).unwrap_or_default();
}
tracing::info!(
worker = %worker_name, hostname = %hostname,
"ping update, memory: container={}MB, windmill={}MB",
memory_usage.unwrap_or_default() / (1024 * 1024),
wm_memory_usage.unwrap_or_default() / (1024 * 1024)
@@ -1185,16 +1191,18 @@ pub async fn run_worker(
if (jobs_executed as u32 + vacuum_shift) % VACUUM_PERIOD == 0 {
let db2 = db.clone();
let current_span = tracing::Span::current();
let worker_name = worker_name.clone();
let hostname = hostname.to_string();
tokio::task::spawn(
(async move {
tracing::info!("vacuuming queue and completed_job");
tracing::info!(worker = %worker_name, hostname = %hostname, "vacuuming queue");
if let Err(e) = sqlx::query!("VACUUM (skip_locked) queue")
.execute(&db2)
.await
{
tracing::error!("failed to vacuum queue: {}", e);
tracing::error!(worker = %worker_name, hostname = %hostname, "failed to vacuum queue: {}", e);
}
tracing::info!("vacuumed queue and completed_job");
tracing::info!(worker = %worker_name, hostname = %hostname, "vacuumed queue");
})
.instrument(current_span),
);
@@ -1230,6 +1238,7 @@ pub async fn run_worker(
if let Ok(same_worker_job) = same_worker_rx.try_recv() {
same_worker_queue_size.fetch_sub(1, Ordering::SeqCst);
tracing::debug!(
worker = %worker_name, hostname = %hostname,
"received {} from same worker channel",
same_worker_job.job_id
);
@@ -1242,6 +1251,7 @@ pub async fn run_worker(
.map_err(|_| Error::InternalErr("Impossible to fetch same_worker job".to_string()));
if r.is_err() && !same_worker_job.recoverable {
tracing::error!(
worker = %worker_name, hostname = %hostname,
"failed to fetch same_worker job on a non recoverable job, exiting"
);
job_completed_tx
@@ -1255,7 +1265,7 @@ pub async fn run_worker(
}
} else if let Ok(_) = killpill_rx.try_recv() {
if !killed_but_draining_same_worker_jobs {
tracing::info!("received killpill for worker {}, jobs are not pulled anymore except same_worker jobs", i_worker);
tracing::info!(worker = %worker_name, hostname = %hostname, "received killpill for worker {}, jobs are not pulled anymore except same_worker jobs", i_worker);
killed_but_draining_same_worker_jobs = true;
job_completed_tx
.0
@@ -1266,10 +1276,10 @@ pub async fn run_worker(
continue;
} else if killed_but_draining_same_worker_jobs {
if job_completed_processor_is_done.load(Ordering::SeqCst) {
tracing::info!("all running jobs have completed and all completed jobs have been fully processed, exiting");
tracing::info!(worker = %worker_name, hostname = %hostname, "all running jobs have completed and all completed jobs have been fully processed, exiting");
break;
} else {
tracing::info!("there may be same_worker jobs to process later, waiting for job_completed_processor to finish progressing all remaining flows before exiting");
tracing::info!(worker = %worker_name, hostname = %hostname, "there may be same_worker jobs to process later, waiting for job_completed_processor to finish progressing all remaining flows before exiting");
tokio::time::sleep(Duration::from_millis(200)).await;
continue;
}
@@ -1294,7 +1304,7 @@ pub async fn run_worker(
if !agent_mode && duration_pull_s > 0.5 {
let empty = job.as_ref().is_ok_and(|x| x.0.is_none());
tracing::warn!("pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}");
tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.5s ({duration_pull_s}), this is a sign that the database is VERY undersized for this load. empty: {empty}, err: {err_pull}");
#[cfg(feature = "prometheus")]
if empty {
if let Some(wp) = worker_pull_over_500_counter_empty.as_ref() {
@@ -1305,7 +1315,7 @@ pub async fn run_worker(
}
} else if !agent_mode && duration_pull_s > 0.1 {
let empty = job.as_ref().is_ok_and(|x| x.0.is_none());
tracing::warn!("pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}");
tracing::warn!(worker = %worker_name, hostname = %hostname, "pull took more than 0.1s ({duration_pull_s}) this is a sign that the database is undersized for this load. empty: {empty}, err: {err_pull}");
#[cfg(feature = "prometheus")]
if empty {
if let Some(wp) = worker_pull_over_100_counter_empty.as_ref() {
@@ -1359,7 +1369,7 @@ pub async fn run_worker(
last_executed_job = None;
jobs_executed += 1;
tracing::debug!("started handling of job {}", job.id);
tracing::debug!(worker = %worker_name, hostname = %hostname, "started handling of job {}", job.id);
if matches!(job.job_kind, JobKind::Script | JobKind::Preview) {
if !dedicated_workers.is_empty() {
@@ -1424,6 +1434,11 @@ pub async fn run_worker(
)
.await;
// counter.add(
// 1,
// worker_resource
// );
#[cfg(feature = "prometheus")]
let _timer = register_metric(
&WORKER_EXECUTION_DURATION,
@@ -1511,6 +1526,40 @@ pub async fn run_worker(
let PulledJob { job, raw_code, raw_lock, raw_flow } = job;
let arc_job = Arc::new(job);
add_time!(bench, "handle_queued_job START");
let span = tracing::span!(tracing::Level::INFO, "job",
job_id = %arc_job.id, root_job = field::Empty, workspace_id = %arc_job.workspace_id, worker = %worker_name, hostname = %hostname, tag = %arc_job.tag,
language = field::Empty,
script_path = field::Empty, flow_step_id = field::Empty, parent_job = field::Empty,
otel.name = field::Empty);
let rj = if let Some(root_job) = arc_job.root_job {
root_job
} else {
arc_job.id
};
if let Some(lg) = arc_job.language.as_ref() {
span.record("language", lg.as_str());
}
if let Some(step_id) = arc_job.flow_step_id.as_ref() {
span.record("otel.name", format!("job {}", step_id).as_str());
span.record("flow_step_id", step_id.as_str());
} else {
span.record("otel.name", "job");
}
if let Some(parent_job) = arc_job.parent_job.as_ref() {
span.record("parent_job", parent_job.to_string().as_str());
}
if let Some(script_path) = arc_job.script_path.as_ref() {
span.record("script_path", script_path.as_str());
}
if let Some(root_job) = arc_job.root_job.as_ref() {
span.record("root_job", root_job.to_string().as_str());
}
windmill_common::otel_ee::set_span_parent(&span, &rj);
// span.context().span().add_event_with_timestamp("job created".to_string(), arc_job.created_at.into(), vec![]);
match handle_queued_job(
arc_job.clone(),
raw_code,
@@ -1529,6 +1578,7 @@ pub async fn run_worker(
#[cfg(feature = "benchmark")]
&mut bench,
)
.instrument(span)
.await
{
Err(err) => {
@@ -1568,6 +1618,8 @@ pub async fn run_worker(
_ => {}
}
#[cfg(feature = "prometheus")]
if let Some(duration) = _timer.map(|x| x.stop_and_record()) {
register_metric(
@@ -1607,7 +1659,7 @@ pub async fn run_worker(
if let Some(secs) = *EXIT_AFTER_NO_JOB_FOR_SECS {
if let Some(lj) = last_executed_job {
if lj.elapsed().as_secs() > secs {
tracing::info!("no job for {} seconds, exiting", secs);
tracing::info!(worker = %worker_name, hostname = %hostname, "no job for {} seconds, exiting", secs);
break;
}
} else {
@@ -1638,12 +1690,12 @@ pub async fn run_worker(
});
}
Err(err) => {
tracing::error!("Failed to pull jobs: {}", err);
tracing::error!(worker = %worker_name, hostname = %hostname, "Failed to pull jobs: {}", err);
}
};
}
tracing::info!("worker {} exiting", worker_name);
tracing::info!(worker = %worker_name, hostname = %hostname, "worker {} exiting", worker_name);
#[cfg(feature = "benchmark")]
{
@@ -1658,22 +1710,24 @@ pub async fn run_worker(
if has_dedicated_workers {
for handle in dedicated_handles {
if let Err(e) = handle.await {
tracing::error!("error in dedicated worker waiting for it to end: {:?}", e)
tracing::error!(worker = %worker_name, hostname = %hostname, "error in dedicated worker waiting for it to end: {:?}", e)
}
}
tracing::info!("all dedicated workers have exited");
tracing::info!(worker = %worker_name, hostname = %hostname, "all dedicated workers have exited");
}
drop(job_completed_tx);
tracing::info!("waiting for job_completed_processor to finish processing remaining jobs");
tracing::info!(worker = %worker_name, hostname = %hostname, "waiting for job_completed_processor to finish processing remaining jobs");
if let Err(e) = send_result.await {
tracing::error!("error in awaiting send_result process: {e:?}")
}
tracing::info!("worker {} exited", worker_name);
tracing::info!("number of jobs executed: {}", jobs_executed);
tracing::info!(worker = %worker_name, hostname = %hostname, "worker {} exited", worker_name);
tracing::info!(worker = %worker_name, hostname = %hostname, "number of jobs executed: {}", jobs_executed);
}
async fn queue_init_bash_maybe<'c>(
db: &Pool<Postgres>,
same_worker_tx: SameWorkerSender,
@@ -1798,7 +1852,6 @@ pub struct PreviousResult<'a> {
pub previous_result: Option<&'a RawValue>,
}
#[tracing::instrument(name = "job", level = "info", skip_all, fields(job_id = %job.id))]
async fn handle_queued_job(
job: Arc<QueuedJob>,
raw_code: Option<String>,
@@ -1816,6 +1869,9 @@ async fn handle_queued_job(
occupancy_metrics: &mut OccupancyMetrics,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> windmill_common::error::Result<bool> {
// Extract the active span from the context
if job.canceled {
return Err(Error::JsonErr(canceled_job_to_result(&job)));
}
@@ -2159,6 +2215,7 @@ async fn handle_queued_job(
}
}
pub fn build_envs(
envs: Option<Vec<String>>,
) -> windmill_common::error::Result<HashMap<String, String>> {
+50 -37
View File
@@ -77,45 +77,34 @@ pub async fn update_flow_status_after_job_completion(
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<()> {
) -> error::Result<Option<Arc<QueuedJob>>> {
// this is manual tailrecursion because async_recursion blows up the stack
// todo!();
potentially_crash_for_testing();
let mut rec = update_flow_status_after_job_completion_internal(
db,
client,
let mut rec = RecUpdateFlowStatusAfterJobCompletion {
flow,
job_id_for_status,
w_id,
job_id_for_status: job_id_for_status.clone(),
success,
result,
unrecoverable,
same_worker_tx.clone(),
worker_dir,
stop_early_override,
false,
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
bench,
)
.await?;
while let Some(nrec) = rec {
skip_error_handler: false,
};
let mut unrecoverable = unrecoverable;
loop {
potentially_crash_for_testing();
rec = match update_flow_status_after_job_completion_internal(
let nrec = match update_flow_status_after_job_completion_internal(
db,
client,
nrec.flow,
&nrec.job_id_for_status,
rec.flow,
&rec.job_id_for_status,
w_id,
nrec.success,
nrec.result,
false,
rec.success,
rec.result,
unrecoverable,
same_worker_tx.clone(),
worker_dir,
nrec.stop_early_override,
nrec.skip_error_handler,
rec.stop_early_override,
rec.skip_error_handler,
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
@@ -125,12 +114,12 @@ pub async fn update_flow_status_after_job_completion(
{
Ok(j) => j,
Err(e) => {
tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", nrec.flow,&nrec.job_id_for_status);
tracing::error!("Error while updating flow status of {} after completion of {}, updating flow status again with error: {e:#}", rec.flow, &rec.job_id_for_status);
update_flow_status_after_job_completion_internal(
db,
client,
nrec.flow,
&nrec.job_id_for_status,
rec.flow,
&rec.job_id_for_status,
w_id,
false,
Arc::new(to_raw_value(&Json(&WrappedError {
@@ -139,8 +128,8 @@ pub async fn update_flow_status_after_job_completion(
true,
same_worker_tx.clone(),
worker_dir,
nrec.stop_early_override,
nrec.skip_error_handler,
rec.stop_early_override,
rec.skip_error_handler,
worker_name,
job_completed_tx.clone(),
#[cfg(feature = "benchmark")]
@@ -148,9 +137,33 @@ pub async fn update_flow_status_after_job_completion(
)
.await?
}
};
unrecoverable = false;
match nrec {
UpdateFlowStatusAfterJobCompletion::Done(job) => {
add_time!(bench, "update flow status internal END");
return Ok(Some(job));
}
UpdateFlowStatusAfterJobCompletion::Rec(nrec) => {
rec = nrec;
},
UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch => {
add_time!(bench, "update flow status internal END");
return Ok(None);
},
UpdateFlowStatusAfterJobCompletion::NotDone => {
add_time!(bench, "update flow status internal END");
return Ok(None);
}
}
}
Ok(())
}
pub enum UpdateFlowStatusAfterJobCompletion {
Rec(RecUpdateFlowStatusAfterJobCompletion),
Done(Arc<QueuedJob>),
NotDone,
NonLastParallelBranch,
}
pub struct RecUpdateFlowStatusAfterJobCompletion {
flow: uuid::Uuid,
@@ -188,7 +201,7 @@ pub async fn update_flow_status_after_job_completion_internal(
worker_name: &str,
job_completed_tx: Sender<SendResult>,
#[cfg(feature = "benchmark")] bench: &mut BenchmarkIter,
) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
) -> error::Result<UpdateFlowStatusAfterJobCompletion> {
add_time!(bench, "update flow status internal START");
let (
should_continue_flow,
@@ -603,7 +616,7 @@ pub async fn update_flow_status_after_job_completion_internal(
);
}
add_time!(bench, "non final parallel flow finished");
return Ok(None);
return Ok(UpdateFlowStatusAfterJobCompletion::NonLastParallelBranch);
}
}
FlowStatusModule::InProgress {
@@ -1148,7 +1161,7 @@ pub async fn update_flow_status_after_job_completion_internal(
if let Some(parent_job) = flow_job.parent_job {
tracing::info!(subflow_id = %flow_job.id, parent_id = %parent_job, "subflow is finished, updating parent flow status");
return Ok(Some(RecUpdateFlowStatusAfterJobCompletion {
return Ok(UpdateFlowStatusAfterJobCompletion::Rec(RecUpdateFlowStatusAfterJobCompletion {
flow: parent_job,
job_id_for_status: flow,
success: success && !is_failure_step,
@@ -1162,9 +1175,9 @@ pub async fn update_flow_status_after_job_completion_internal(
}));
}
}
Ok(None)
Ok(UpdateFlowStatusAfterJobCompletion::Done(flow_job))
} else {
Ok(None)
Ok(UpdateFlowStatusAfterJobCompletion::NotDone)
}
}
@@ -0,0 +1,255 @@
<script lang="typescript">
import { enterpriseLicense } from '$lib/stores'
import { Alert, Button, Tab, Tabs } from './common'
import OAuthSetting from '$lib/components/OAuthSetting.svelte'
import OktaSetting from './OktaSetting.svelte'
import CloseButton from './common/CloseButton.svelte'
import KeycloakSetting from './KeycloakSetting.svelte'
import CustomSso from './CustomSso.svelte'
import AuthentikSetting from '$lib/components/AuthentikSetting.svelte'
import AutheliaSetting from '$lib/components/AutheliaSetting.svelte'
import KanidmSetting from '$lib/components/KanidmSetting.svelte'
import ZitadelSetting from '$lib/components/ZitadelSetting.svelte'
import CustomOauth from './CustomOauth.svelte'
import { capitalize } from '$lib/utils'
import Toggle from './Toggle.svelte'
import { ExternalLink, Plus } from 'lucide-svelte'
export let snowflakeAccountIdentifier = ''
export let oauths: Record<string, any> = {}
const windmillBuiltins = [
'github',
'gitlab',
'bitbucket',
'slack',
'gsheets',
'gdrive',
'gmail',
'gcal',
'gforms',
'gcloud',
'gworkspace',
'basecamp',
'linkedin',
'quickbooks',
'visma',
'spotify',
'snowflake_oauth'
]
let oauth_name = undefined
let clientName = ''
let resourceName = ''
let tab: 'sso' | 'oauth' | 'scim' = 'sso'
let requirePreexistingUserForOauth: boolean = false
</script>
<div>
<Tabs bind:selected={tab} class="mt-2 mb-4">
<Tab value="sso">SSO</Tab>
<Tab value="oauth">OAuth</Tab>
<Tab value="scim">SCIM/SAML</Tab>
</Tabs>
</div>
<div class="mb-6">
{#if tab === 'sso'}
{#if !$enterpriseLicense || $enterpriseLicense.endsWith('_pro')}
<Alert type="warning" title="Limited to 10 SSO users">
Without EE, the number of SSO users is limited to 10. SCIM/SAML is available on EE
</Alert>
{/if}
<div class="py-1" />
<div class="mb-2">
<span class="text-primary text-sm"
>When at least one of the below options is set, users will be able to login to Windmill via
their third-party account.
<br /> To test SSO, the recommended workflow is to to save the settings and try to login in
an incognito window.
<a target="_blank" href="https://www.windmill.dev/docs/misc/setup_oauth#sso">Learn more</a
></span
>
</div>
<div class="flex flex-col gap-3 py-4">
<OAuthSetting name="google" bind:value={oauths['google']} />
<OAuthSetting name="microsoft" bind:value={oauths['microsoft']} />
<OktaSetting bind:value={oauths['okta']} />
<OAuthSetting name="github" bind:value={oauths['github']} />
<OAuthSetting name="gitlab" bind:value={oauths['gitlab']} />
<OAuthSetting name="jumpcloud" bind:value={oauths['jumpcloud']} />
<KeycloakSetting bind:value={oauths['keycloak']} />
<AuthentikSetting bind:value={oauths['authentik']} />
<AutheliaSetting bind:value={oauths['authelia']} />
<KanidmSetting bind:value={oauths['kanidm']} />
<ZitadelSetting bind:value={oauths['zitadel']} />
{#each Object.keys(oauths) as k}
{#if !['authelia', 'authentik', 'google', 'microsoft', 'github', 'gitlab', 'jumpcloud', 'okta', 'keycloak', 'slack', 'kanidm', 'zitadel'].includes(k) && 'login_config' in oauths[k]}
{#if oauths[k]}
<div class="flex flex-col gap-2 pb-4">
<div class="flex flex-row items-center gap-2">
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="text-md font-medium text-primary">{k}</label>
<CloseButton
on:close={() => {
delete oauths[k]
oauths = { ...oauths }
}}
/>
</div>
<div class="p-2 border rounded">
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Custom Name</span>
<input
type="text"
placeholder="Custom Name"
bind:value={oauths[k]['display_name']}
/>
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Client Id</span>
<input type="text" placeholder="Client Id" bind:value={oauths[k]['id']} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Client Secret</span>
<input type="text" placeholder="Client Secret" bind:value={oauths[k]['secret']} />
</label>
{#if !windmillBuiltins.includes(k) && k != 'slack'}
<CustomSso bind:login_config={oauths[k]['login_config']} />
{/if}
</div>
</div>
{/if}
{/if}
{/each}
</div>
<div class="flex gap-2 py-4">
<input type="text" placeholder="client_id" bind:value={clientName} />
<Button
variant="border"
color="blue"
hover="yo"
size="sm"
endIcon={{ icon: Plus }}
disabled={clientName == ''}
on:click={() => {
oauths[clientName] = { id: '', secret: '', login_config: {} }
clientName = ''
}}
>
Add custom SSO client {!$enterpriseLicense ? '(requires ee)' : ''}
</Button>
</div>
<div class="flex gap-2 py-4">
<Toggle
options={{
right: 'Require users to have been added manually to Windmill to sign in through OAuth'
}}
bind:checked={requirePreexistingUserForOauth}
/>
</div>
{:else if tab === 'oauth'}
<div class="mb-2">
<span class="text-primary text-sm"
>When one of the below options is set, you will be able to create a specific resource
containing a token automatically generated by the third-party provider.
<br />
To test it after setting an oauth client, go to the Resources menu and create a new one of the
type of your oauth client (i.e. a 'github' resource if you set Github OAuth).
<br /><a target="_blank" href="https://www.windmill.dev/docs/misc/setup_oauth#oauth"
>Learn more</a
></span
>
</div>
<div class="py-1" />
<OAuthSetting login={false} name="slack" bind:value={oauths['slack']} />
<div class="py-1" />
{#each Object.keys(oauths) as k}
{#if oauths[k] && !('login_config' in oauths[k])}
{#if !['slack'].includes(k) && oauths[k]}
<div class="flex flex-col gap-2 pb-4">
<div class="flex flex-row items-center gap-2">
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="text-md font-medium text-primary">{k}</label>
<CloseButton
on:close={() => {
delete oauths[k]
oauths = { ...oauths }
}}
/>
</div>
<div class="p-2 border rounded">
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Client Id</span>
<input type="text" placeholder="Client Id" bind:value={oauths[k]['id']} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Client Secret</span>
<input type="text" placeholder="Client Secret" bind:value={oauths[k]['secret']} />
</label>
{#if !windmillBuiltins.includes(k) && k != 'slack'}
<CustomOauth bind:connect_config={oauths[k]['connect_config']} />
{/if}
{#if k == 'snowflake_oauth'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm flex gap-2 items-center"
><a
href="https://docs.snowflake.com/en/user-guide/admin-account-identifier#using-an-account-name-as-an-identifier"
target="_blank">Snowflake Account Identifier</a
><ExternalLink size={12} /></span
>
<input
type="text"
placeholder="<orgname>-<account_name>"
required={true}
bind:value={snowflakeAccountIdentifier}
/>
</label>
{/if}
</div>
</div>
{/if}
{/if}
{/each}
<div class="flex gap-2">
<select name="oauth_name" id="oauth_name" bind:value={oauth_name}>
<option value={undefined}>Select an OAuth client</option>
<option value="custom">Fully Custom (requires ee)</option>
{#each windmillBuiltins as name}
<option value={name}>{capitalize(name)}</option>
{/each}
</select>
{#if oauth_name == 'custom'}
<input type="text" placeholder="client_id" bind:value={resourceName} />
{:else}
<input type="text" value={oauth_name ?? ''} disabled />
{/if}
<Button
variant="border"
color="blue"
hover="yo"
size="sm"
endIcon={{ icon: Plus }}
disabled={!oauth_name ||
(oauth_name == 'custom' && resourceName == '') ||
(oauth_name == 'custom' && !$enterpriseLicense)}
on:click={() => {
let name = oauth_name == 'custom' ? resourceName : oauth_name
oauths[name ?? ''] = { id: '', secret: '' }
resourceName = ''
}}
>
Add OAuth client {oauth_name == 'custom' && !$enterpriseLicense ? '(requires ee)' : ''}
</Button>
</div>
{:else if tab == 'scim'}
<slot name="scim" />
{/if}
</div>
@@ -0,0 +1,783 @@
<script lang="typescript">
import { isCloudHosted } from '$lib/cloud'
import { enterpriseLicense, isCriticalAlertsUIOpen } from '$lib/stores'
import {
AlertCircle,
AlertTriangle,
BadgeCheck,
BadgeX,
Info,
Plus,
Slack,
X
} from 'lucide-svelte'
import type { Setting } from './instanceSettings'
import Tooltip from './Tooltip.svelte'
import ObjectStoreConfigSettings from './ObjectStoreConfigSettings.svelte'
import { sendUserToast } from '$lib/toast'
import ConfirmButton from './ConfirmButton.svelte'
import { IndexSearchService, SettingService } from '$lib/gen'
import { Button, SecondsInput, Skeleton } from './common'
import Password from './Password.svelte'
import { classNames } from '$lib/utils'
import Popover from './Popover.svelte'
import Toggle from './Toggle.svelte'
import type { Writable } from 'svelte/store'
import { createEventDispatcher } from 'svelte'
import { fade } from 'svelte/transition'
import { base } from '$lib/base'
export let setting: Setting
export let version: string
export let values: Writable<Record<string, any>>
export let loading = true
const dispatch = createEventDispatcher()
let latestKeyRenewalAttempt: {
result: string
attempted_at: string
} | null
function showSetting(setting: string, values: Record<string, any>) {
if (setting == 'dev_instance') {
if (values['license_key'] == undefined) {
return false
}
}
return true
}
let licenseKeyChanged = false
let renewing = false
let opening = false
async function reloadKeyrenewalAttemptInfo() {
latestKeyRenewalAttempt = await SettingService.getLatestKeyRenewalAttempt()
}
if (setting.key == 'license_key') {
reloadKeyrenewalAttemptInfo()
}
export async function renewLicenseKey() {
renewing = true
try {
await SettingService.renewLicenseKey({
licenseKey: $values['license_key'] || undefined
})
sendUserToast('Key renewal successful')
reloadKeyrenewalAttemptInfo()
} catch (err) {
latestKeyRenewalAttempt = await SettingService.getLatestKeyRenewalAttempt()
throw err
} finally {
renewing = false
}
}
export async function openCustomerPortal() {
opening = true
try {
const url = await SettingService.createCustomerPortalSession({
licenseKey: $values['license_key'] || undefined
})
window.open(url, '_blank')
} finally {
opening = false
}
}
function parseLicenseKey(key: string): {
valid: boolean
expiration?: string
} {
let splitted = key.split('.')
if (splitted.length >= 3) {
try {
let i = parseInt(splitted[1])
let date = new Date(i * 1000)
const stringDate = date.toLocaleDateString()
if (stringDate !== 'Invalid Date') {
return {
valid: date.getTime() > Date.now(),
expiration: date.toLocaleDateString()
}
}
} catch {}
}
return {
valid: false
}
}
</script>
{#if (!setting.cloudonly || isCloudHosted()) && showSetting(setting.key, $values) && !(setting.hiddenIfNull && $values[setting.key] == null)}
{#if setting.ee_only != undefined && !$enterpriseLicense}
<div class="flex text-xs items-center gap-1 text-yellow-500 whitespace-nowrap">
<AlertTriangle size={16} />
EE only {#if setting.ee_only != ''}<Tooltip>{setting.ee_only}</Tooltip>{/if}
</div>
{/if}
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">{setting.label}</span>
{#if setting.description}
<span class="text-secondary text-xs">
{@html setting.description}
</span>
{/if}
{#if setting.tooltip}
<Tooltip>{setting.tooltip}</Tooltip>
{/if}
{#if $values}
{@const hasError = setting.isValid && !setting.isValid($values[setting.key])}
{#if loading}
<Skeleton layout={[[2.5]]} />
{:else if setting.fieldType == 'text'}
<input
disabled={setting.ee_only != undefined && !$enterpriseLicense}
type="text"
placeholder={setting.placeholder}
class={hasError
? 'border !border-red-700 !border-opacity-30 !focus:border-red-700 !focus:border-opacity-30'
: ''}
bind:value={$values[setting.key]}
/>
{#if setting.advancedToggle}
<div class="mt-1">
<Toggle
size="xs"
options={{ right: setting.advancedToggle.label }}
checked={setting.advancedToggle.checked($values)}
on:change={() => {
if (setting.advancedToggle) {
$values = setting.advancedToggle.onChange($values)
}
}}
/>
</div>
{/if}
{:else if setting.fieldType == 'textarea'}
<textarea
disabled={!$enterpriseLicense}
rows="2"
placeholder={setting.placeholder}
bind:value={$values[setting.key]}
/>
{#if setting.key == 'saml_metadata'}
<div class="flex mt-2">
<Button
disabled={!$enterpriseLicense}
on:click={async (e) => {
const res = await SettingService.testMetadata({
requestBody: $values[setting.key]
})
sendUserToast(`Metadata valid, see console for full content`)
console.log(`Metadata content:`, res)
}}>Test content/url</Button
>
</div>
{/if}
{:else if setting.fieldType == 'license_key'}
{@const { valid, expiration } = parseLicenseKey($values[setting.key] ?? '')}
<div class="flex gap-2">
<Password
small
placeholder={setting.placeholder}
on:keydown={() => {
licenseKeyChanged = true
}}
bind:password={$values[setting.key]}
/>
<Button
variant={$values[setting.key] ? 'contained' : 'border'}
size="xs"
on:click={async () => {
await SettingService.testLicenseKey({
requestBody: { license_key: $values[setting.key] }
})
sendUserToast('Valid key')
}}
>
Test Key
</Button>
</div>
<div class="mt-1 flex flex-col gap-1 items-start">
{#if $values[setting.key]?.length > 0}
{#if valid}
<div class="flex flex-row gap-1 items-center">
<Info size={12} class="text-tertiary" />
<span class="text-tertiary text-xs">License key expires on {expiration ?? ''}</span>
</div>
{:else if expiration}
<div class="flex flex-row gap-1 items-center">
<AlertCircle size={12} class="text-red-600" />
<span class="text-red-600 dark:text-red-400 text-xs"
>License key expired on {expiration}</span
>
</div>
{:else}
<div class="flex flex-row gap-1 items-center">
<AlertCircle size={12} class="text-red-600" />
<span class="text-red-600 dark:text-red-400 text-xs"
>Invalid license key format</span
>
</div>
{/if}
{/if}
{#if latestKeyRenewalAttempt}
{@const attemptedAt = new Date(latestKeyRenewalAttempt.attempted_at).toLocaleString()}
{@const isTrial = latestKeyRenewalAttempt.result.startsWith('error: trial:')}
<div class="relative">
<Popover notClickable>
<div class="flex flex-row items-center gap-1">
{#if latestKeyRenewalAttempt.result === 'success'}
<BadgeCheck class="text-green-600" size={12} />
{:else}
<BadgeX class={isTrial ? 'text-yellow-600' : 'text-red-600'} size={12} />
{/if}
<span
class={classNames(
'text-xs',
latestKeyRenewalAttempt.result === 'success'
? 'text-green-600'
: isTrial
? 'text-yellow-600'
: 'text-red-600'
)}
>
{latestKeyRenewalAttempt.result === 'success'
? 'Latest key renewal succeeded'
: isTrial
? 'Latest key renewal ignored because in trial'
: 'Latest key renewal failed'}
on {attemptedAt}
</span>
</div>
<div slot="text">
{#if latestKeyRenewalAttempt.result === 'success'}
<span class="text-green-300">
Latest key renewal succeeded on {attemptedAt}
</span>
{:else if isTrial}
<span class="text-yellow-300">
License key cannot be renewed during trial ({attemptedAt})
</span>
{:else}
<span class="text-red-300">
Latest key renewal failed on {attemptedAt}: {latestKeyRenewalAttempt.result.replace(
'error: ',
''
)}
</span>
{/if}
<br />
As long as invoices are paid and usage corresponds to the subscription, the key is
renewed daily with a validity of 35 days (grace period).
</div>
</Popover>
</div>
{/if}
{#if licenseKeyChanged && !$enterpriseLicense}
{#if version.startsWith('CE')}
<div class="text-red-400"
>License key is set but image used is the Community Edition {version}. Switch image
to EE.</div
>
{/if}
{/if}
{#if valid || expiration}
<div class="flex flex-row gap-2 mt-1">
<Button on:click={renewLicenseKey} loading={renewing} size="xs" color="dark"
>Renew key
</Button>
<Button color="dark" size="xs" loading={opening} on:click={openCustomerPortal}>
Open customer portal
</Button>
</div>
{/if}
</div>
{:else if setting.fieldType == 'email'}
<input type="email" placeholder={setting.placeholder} bind:value={$values[setting.key]} />
{:else if setting.key == 'critical_alert_mute_ui'}
<div class="flex flex-col gap-y-2 my-2 py-2">
<Toggle
disabled={!$enterpriseLicense}
bind:checked={$values[setting.key]}
options={{ right: setting.description }}
/>
<div class="flex flex-row">
<Button
variant="border"
color="light"
disabled={!$enterpriseLicense}
size="xs"
on:click={() => {
isCriticalAlertsUIOpen.set(true)
dispatch('closeDrawer')
}}
>
Show Critical Alerts
</Button>
</div>
</div>
{:else if setting.fieldType == 'critical_error_channels'}
<div class="w-full flex gap-x-16 flex-wrap">
<div class="w-full max-w-lg">
{#if $enterpriseLicense && Array.isArray($values[setting.key])}
{#each $values[setting.key] ?? [] as v, i}
<div class="flex w-full max-w-lg mt-1 gap-2 items-center">
<select
class="w-20"
on:change={(e) => {
if (e.target?.['value']) {
$values[setting.key][i] = {
[e.target['value']]: ''
}
}
}}
value={v && 'slack_channel' in v ? 'slack_channel' : 'email'}
>
<option value="email">Email</option>
<option value="slack_channel">Slack</option>
</select>
{#if v && 'slack_channel' in v}
<input
type="text"
placeholder="Slack channel"
on:input={(e) => {
if (e.target?.['value']) {
$values[setting.key][i] = {
slack_channel: e.target['value']
}
}
}}
value={v?.slack_channel ?? ''}
/>
{:else}
<input
type="email"
placeholder="Email address"
on:input={(e) => {
if (e.target?.['value']) {
$values[setting.key][i] = {
email: e.target['value']
}
}
}}
value={v?.email ?? ''}
/>
{/if}
<button
transition:fade|local={{ duration: 100 }}
class="rounded-full p-1 bg-surface-secondary duration-200 hover:bg-surface-hover"
aria-label="Clear"
on:click={() => {
$values[setting.key] = $values[setting.key].filter((_, index) => index !== i)
}}
>
<X size={14} />
</button>
</div>
{/each}
{/if}
</div>
</div>
<div class="flex mt-2 gap-20 items-center">
<Button
variant="border"
color="light"
size="md"
btnClasses="mt-1"
on:click={() => {
if ($values[setting.key] == undefined || !Array.isArray($values[setting.key])) {
$values[setting.key] = []
}
$values[setting.key] = $values[setting.key].concat('')
}}
id="arg-input-add-item"
startIcon={{ icon: Plus }}
disabled={!$enterpriseLicense}
>
Add channel
</Button>
<div class="flex mt-1">
<Button
disabled={!$enterpriseLicense}
variant="border"
color="light"
size="md"
on:click={async () => {
try {
await SettingService.testCriticalChannels({
requestBody: $values[setting.key]
})
sendUserToast('Test message sent successfully to critical channels', false)
} catch (error) {
sendUserToast('Failed to send test message: ' + error.message, true)
}
}}
>
Test Channels
</Button>
</div>
</div>
{:else if setting.fieldType == 'slack_connect'}
<div class="flex flex-col items-start self-start">
{#if $values[setting.key] && 'team_name' in $values[setting.key]}
<div class="text-sm">
Connected to <code>{$values[setting.key]['team_name']}</code>
</div>
<Button
size="sm"
endIcon={{ icon: Slack }}
btnClasses="mt-2"
variant="border"
on:click={async () => {
$values[setting.key] = undefined
}}
>
Disconnect Slack
</Button>
{:else}
<Button
size="xs"
color="dark"
href="{base}/api/oauth/connect_slack?instance=true"
startIcon={{ icon: Slack }}
disabled={!$enterpriseLicense}
>
Connect to Slack
</Button>
{/if}
</div>
{:else if setting.fieldType == 'indexer_rates'}
<div class="flex flex-col gap-4 mt-4">
{#if $values[setting.key]}
<div>
<label for="writer_memory_budget" class="block text-sm font-medium">
Index writer memory budget (MB)
<Tooltip>
The allocated memory arena for the indexer. A bigger value means less writing to
disk and potentially higher indexing throughput
</Tooltip>
</label>
<input
disabled={!$enterpriseLicense}
type="number"
id="writer_memory_budget"
placeholder="300"
on:input={(e) => {
if (e.target instanceof HTMLInputElement) {
if (e.target.valueAsNumber) {
$values[setting.key].writer_memory_budget =
e.target.valueAsNumber * (1024 * 1024)
}
}
}}
value={$values[setting.key].writer_memory_budget / (1024 * 1024)}
/>
</div>
<h3>Completed Job Index</h3>
<div>
<label for="commit_job_max_batch_size" class="block text-sm font-medium">
Commit max batch size <Tooltip>
The max amount of documents (here jobs) per commit. To optimize indexing
throughput, it is best to keep this as high as possible. However, especially when
reindexing the whole instance, it can be useful to have a limit on how many jobs
can be written without being commited. A commit will make the jobs available for
search, constitute a "checkpoint" state in the indexing and will be logged.
</Tooltip>
</label>
<input
disabled={!$enterpriseLicense}
type="number"
id="commit_job_max_batch_size"
placeholder="100000"
bind:value={$values[setting.key].commit_job_max_batch_size}
/>
</div>
<div>
<label for="refresh_index_period" class="block text-sm font-medium">
Refresh index period (s) <Tooltip>
The index will query new jobs peridically and write them on the index. This
setting sets that period.
</Tooltip></label
>
<input
disabled={!$enterpriseLicense}
type="number"
id="refresh_index_period"
placeholder="300"
bind:value={$values[setting.key].refresh_index_period}
/>
</div>
<div>
<label for="max_indexed_job_log_size" class="block text-sm font-medium">
Max indexed job log size (KB) <Tooltip>
Job logs are included when indexing, but to avoid the index size growing
artificially, the logs will be truncated after a size has been reached.
</Tooltip>
</label>
<input
disabled={!$enterpriseLicense}
type="number"
id="max_indexed_job_log_size"
placeholder="1024"
on:input={(e) => {
if (e.target instanceof HTMLInputElement) {
if (e.target.valueAsNumber) {
$values[setting.key].max_indexed_job_log_size = e.target.valueAsNumber * 1024
}
}
}}
value={$values[setting.key].max_indexed_job_log_size / 1024}
/>
</div>
<h3>Service Logs Index</h3>
<div>
<label for="commit_log_max_batch_size" class="block text-sm font-medium"
>Commit max batch size Commit max batch size <Tooltip>
The max amount of documents per commit. In this case 1 document is one log file
representing all logs during 1 minute for a specific host. To optimize indexing
throughput, it is best to keep this as high as possible. However, especially when
reindexing the whole instance, it can be useful to have a limit on how many logs
can be written without being commited. A commit will make the logs available for
search, appear as a log line, and be a "checkpoint" of the indexing progress.
</Tooltip>
</label>
<input
disabled={!$enterpriseLicense}
type="number"
id="commit_log_max_batch_size"
placeholder="10000"
bind:value={$values[setting.key].commit_log_max_batch_size}
/>
</div>
<div>
<label for="refresh_log_index_period" class="block text-sm font-medium">
Refresh index period (s) <Tooltip>
The index will query new service logs peridically and write them on the index.
This setting sets that period.
</Tooltip></label
>
<input
disabled={!$enterpriseLicense}
type="number"
id="refresh_log_index_period"
placeholder="300"
bind:value={$values[setting.key].refresh_log_index_period}
/>
</div>
<h3>Reset Index</h3>
This buttons will clear the whole index, and the service will start reindexing from scratch.
Full text search might be down during this time.
<div>
<ConfirmButton
on:click={async () => {
let r = await IndexSearchService.clearIndex({
idxName: 'JobIndex'
})
console.log('asasd')
sendUserToast(r)
}}>Clear <b>Jobs</b> Index</ConfirmButton
>
<ConfirmButton
on:click={async () => {
let r = await IndexSearchService.clearIndex({
idxName: 'ServiceLogIndex'
})
console.log('asasd')
sendUserToast(r)
}}>Clear <b>Service Logs</b> Index</ConfirmButton
>
</div>
{/if}
</div>
{:else if setting.fieldType == 'smtp_connect'}
<div class="flex flex-col gap-4 border rounded p-4">
{#if $values[setting.key]}
<div>
<label for="smtp_host" class="block text-sm font-medium">Host</label>
<input
type="text"
id="smtp_host"
placeholder="smtp.gmail.com"
bind:value={$values[setting.key].smtp_host}
/>
</div>
<div>
<label for="smtp_port" class="block text-sm font-medium">Port</label>
<input
type="number"
id="smtp_port"
placeholder="587"
bind:value={$values[setting.key].smtp_port}
/>
</div>
<div>
<label for="smtp_username" class="block text-sm font-medium">Username</label>
<input
type="text"
id="smtp_username"
placeholder="ruben@windmill.dev"
bind:value={$values[setting.key].smtp_username}
/>
</div>
<div>
<label for="smtp_password" class="block text-sm font-medium">Password</label>
<Password bind:password={$values[setting.key].smtp_password} />
</div>
<div>
<label for="smtp_from" class="block text-sm font-medium">From Address</label>
<input
type="email"
id="smtp_from"
placeholder="noreply@windmill.dev"
bind:value={$values[setting.key].smtp_from}
/>
</div>
<div>
<Toggle
disabled={$values[setting.key].smtp_disable_tls == true || !$enterpriseLicense}
id="smtp_tls_implicit"
bind:checked={$values[setting.key].smtp_tls_implicit}
options={{ right: 'Implicit TLS' }}
label="Implicit TLS"
/>
</div>
<div>
<Toggle
id="smtp_disable_tls"
disabled={!$enterpriseLicense}
bind:checked={$values[setting.key].smtp_disable_tls}
on:change={() => {
if ($values[setting.key].smtp_disable_tls) {
$values[setting.key].smtp_tls_implicit = false
}
}}
options={{ right: 'Disable TLS' }}
label="Disable TLS"
/>
</div>
{/if}
</div>
{:else if setting.fieldType == 'otel'}
<div class="flex flex-col gap-4 border rounded p-4">
{#if $values[setting.key]}
<div class="flex gap-8">
<Toggle
disabled={!$enterpriseLicense}
id="tracing_enabled"
bind:checked={$values[setting.key].tracing_enabled}
options={{ right: 'Tracing' }}
label="Tracing"
/>
<Toggle
disabled={!$enterpriseLicense}
id="logs_enabled"
bind:checked={$values[setting.key].logs_enabled}
options={{ right: 'Logs' }}
label="logs"
/>
<Toggle
disabled
id="metrics_enabled"
bind:checked={$values[setting.key].logs_enabled}
options={{ right: 'Metrics (coming soon)' }}
label="metrics"
/>
</div>
<div>
<label for="OTEL_EXPORTER_OTLP_ENDPOINT" class="block text-sm font-medium"
>Endpoint</label
>
<input
disabled={!$enterpriseLicense}
type="text"
id="OTEL_EXPORTER_OTLP_ENDPOINT"
placeholder="http://otel-collector.example.com:4317"
bind:value={$values[setting.key].otel_exporter_otlp_endpoint}
/>
</div>
<div>
<label for="OTEL_EXPORTER_OTLP_HEADERS" class="block text-sm font-medium"
>Headers</label
>
<input
disabled={!$enterpriseLicense}
type="text"
id="OTEL_EXPORTER_OTLP_HEADERS"
placeholder="Authorization=Bearer my-secret-token,Env=production"
bind:value={$values[setting.key].otel_exporter_otlp_headers}
/>
</div>
<!-- <div>
<label for="OTEL_EXPORTER_OTLP_PROTOCOL" class="block text-sm font-medium"
>Protocol<span class="text-2xs text-tertiary ml-4"
>grpc, http/protobuf, http/json</span
></label
>
<input
type="text"
id="OTEL_EXPORTER_OTLP_PROTOCOL"
placeholder="grpc"
bind:value={$values[setting.key].otel_exporter_otlp_protocol}
/>
</div>
<div>
<label for="OTEL_EXPORTER_OTLP_COMPRESSION" class="block text-sm font-medium"
>Compression <span class="text-2xs text-tertiary ml-4">none, gzip</span></label
>
<input
type="text"
id="OTEL_EXPORTER_OTLP_COMPRESSION"
placeholder="none"
bind:value={$values[setting.key].otel_exporter_otlp_compression}
/>
</div> -->
{/if}
</div>
{:else if setting.fieldType == 'object_store_config'}
<ObjectStoreConfigSettings bind:bucket_config={$values[setting.key]} />
<div class="mb-6" />
{:else if setting.fieldType == 'number'}
<input type="number" placeholder={setting.placeholder} bind:value={$values[setting.key]} />
{:else if setting.fieldType == 'password'}
<input
autocomplete="new-password"
type="password"
placeholder={setting.placeholder}
bind:value={$values[setting.key]}
/>
{:else if setting.fieldType == 'boolean'}
<div class="mt-0.5">
<Toggle
disabled={setting.ee_only != undefined && !$enterpriseLicense}
bind:checked={$values[setting.key]}
/>
</div>
{:else if setting.fieldType == 'seconds'}
<div>
<SecondsInput
max={setting.ee_only != undefined && !$enterpriseLicense
? 60 * 60 * 24 * 30
: undefined}
bind:seconds={$values[setting.key]}
/>
</div>
{/if}
{#if hasError}
<span class="text-red-500 dark:text-red-400 text-sm">
{setting.error ?? ''}
</span>
{/if}
{:else}
<input disabled placeholder="Loading..." />
{/if}
</label>
{/if}
File diff suppressed because it is too large Load Diff
@@ -50,9 +50,10 @@
}
</script>
<div class="flex flex-col gap-1">
<div class="flex flex-col">
<!-- svelte-ignore a11y-label-has-associated-control -->
<label class="text-sm flex gap-4 items-center font-medium text-primary"
<label
class="text-sm flex gap-4 items-center font-medium text-primary {enabled ? 'rounded py-2' : ''}"
><div class="w-[120px]"><IconedResourceType {name} after={true} /></div><Toggle
checked={enabled}
on:change={(e) => {
@@ -65,7 +66,7 @@
/></label
>
{#if enabled}
<div class="p-2 rounded border">
<div class="p-2 rounded border mb-4">
{#if name != 'slack'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Custom Name</span>
@@ -32,7 +32,7 @@
<div class="flex items-center mt-1">
<Button
variant="border"
color="blue"
color="light"
hover="yo"
size="sm"
endIcon={{ icon: Plus }}
@@ -11,7 +11,7 @@
<input type="text" bind:value={v} />
<Button
variant="border"
color="red"
color="light"
size="xs"
btnClasses="mx-6"
on:click={() => {
@@ -27,7 +27,7 @@
<div class="flex items-center mt-1">
<Button
variant="border"
color="blue"
color="light"
hover="yo"
size="sm"
endIcon={{ icon: Plus }}
@@ -5,6 +5,7 @@
import { SettingService } from '$lib/gen'
import { sendUserToast } from '$lib/toast'
import TestConnection from './TestConnection.svelte'
import { enterpriseLicense } from '$lib/stores'
type S3Config = {
type: 'S3'
@@ -51,9 +52,10 @@
}
</script>
<div>
<div class="my-0.5">
<Toggle
options={{ right: 'Enable' }}
disabled={!$enterpriseLicense}
options={{ right: bucket_config ? '' : 'set object store' }}
checked={Boolean(bucket_config)}
on:change={(e) => {
if (e.detail) {
@@ -72,134 +74,145 @@
/>
</div>
{#if bucket_config}
<div class="flex gap-2 py-1">
<Button
spacingSize="sm"
size="xs"
btnClasses="h-8"
color="light"
variant="border"
on:click={testConnection}
<div class="p-2">
<div class="flex gap-2 py-1">
<Button
spacingSize="sm"
size="xs"
btnClasses="h-8"
color="light"
variant="border"
on:click={testConnection}
>
{#if loading}
<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
{:else}
<Database class="mr-2 !h-4 !w-4" />
{/if}
Test from a server
</Button>
<TestConnection
args={bucket_config}
resourceType="s3_bucket"
workspaceOverride="admins"
buttonTextOverride="Test from a worker"
/>
</div>
<Tabs
bind:selected={bucket_config.type}
on:selected={(e) => {
if (e.detail === 'S3') {
bucket_config = {
type: 'S3',
bucket: '',
region: '',
access_key: '',
secret_key: '',
endpoint: ''
}
} else if (e.detail === 'Azure') {
bucket_config = {
type: 'Azure',
accountName: '',
containerName: '',
useSSL: false,
tenantId: '',
clientId: '',
accessKey: ''
}
}
}}
>
{#if loading}
<Loader2 class="animate-spin mr-2 !h-4 !w-4" />
{:else}
<Database class="mr-2 !h-4 !w-4" />
{/if}
Test from a server
</Button>
<TestConnection
args={bucket_config}
resourceType="s3_bucket"
workspaceOverride="admins"
buttonTextOverride="Test from a worker"
/>
</div>
<Tabs
bind:selected={bucket_config.type}
on:selected={(e) => {
if (e.detail === 'S3') {
bucket_config = {
type: 'S3',
bucket: '',
region: '',
access_key: '',
secret_key: '',
endpoint: ''
}
} else if (e.detail === 'Azure') {
bucket_config = {
type: 'Azure',
accountName: '',
containerName: '',
useSSL: false,
tenantId: '',
clientId: '',
accessKey: ''
}
}
}}
>
<Tab size="sm" value="S3">S3</Tab>
<Tab size="sm" value="Azure">Azure Blob</Tab>
</Tabs>
<div class="flex flex-col gap-2 mt-2 p-2 border rounded-md">
{#if bucket_config.type === 'S3'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Bucket</span>
<input type="text" placeholder="bucket-name" bind:value={bucket_config.bucket} />
</label>
<Tab size="sm" value="S3">S3</Tab>
<Tab size="sm" value="Azure">Azure Blob</Tab>
</Tabs>
<div class="flex flex-col gap-2 mt-2 p-2 border rounded-md">
{#if bucket_config.type === 'S3'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Bucket</span>
<input type="text" placeholder="bucket-name" bind:value={bucket_config.bucket} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Region</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_REGION</span
>
<input type="text" bind:value={bucket_config.region} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Access key ID</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_ACCESS_KEY_ID, pod or ec2 profile</span
>
<input type="text" bind:value={bucket_config.access_key} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Secret key</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_SECRET_KEY, pod or ec2 profile</span
>
<input type="password" autocomplete="new-password" bind:value={bucket_config.secret_key} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Endpoint</span>
<span class="text-tertiary text-2xs"
>Only needed for non AWS S3 providers like R2 or MinIo</span
>
<input type="text" bind:value={bucket_config.endpoint} />
</label>
<div class="block pb-2">
<span class="text-tertiary text-2xs">Disable if using https only policy</span>
<div>
<Toggle bind:checked={bucket_config.allow_http} options={{ right: 'Allow http' }} />
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Region</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_REGION</span
>
<input type="text" bind:value={bucket_config.region} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Access key ID</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_ACCESS_KEY_ID, pod or ec2
profile</span
>
<input type="text" bind:value={bucket_config.access_key} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Secret key</span>
<span class="text-tertiary text-2xs"
>If left empty, will be derived automatically from $AWS_SECRET_KEY, pod or ec2 profile</span
>
<input
type="password"
autocomplete="new-password"
bind:value={bucket_config.secret_key}
/>
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Endpoint</span>
<span class="text-tertiary text-2xs"
>Only needed for non AWS S3 providers like R2 or MinIo</span
>
<input type="text" bind:value={bucket_config.endpoint} />
</label>
<div class="block pb-2">
<span class="text-tertiary text-2xs">Disable if using https only policy</span>
<div>
<Toggle bind:checked={bucket_config.allow_http} options={{ right: 'Allow http' }} />
</div>
</div>
</div>
{:else if bucket_config.type === 'Azure'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Account name</span>
<input type="text" placeholder="account-name" bind:value={bucket_config.accountName} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Container name</span>
<input type="text" placeholder="container-name" bind:value={bucket_config.containerName} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Access key</span>
<input type="password" autocomplete="new-password" bind:value={bucket_config.accessKey} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Tenant ID <span class="text-2xs text-tertiary">(optional)</span></span
>
<input type="text" bind:value={bucket_config.tenantId} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Client ID <span class="text-2xs text-tertiary">(optional)</span></span
>
<input type="text" bind:value={bucket_config.clientId} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Endpoint <span class="text-2xs text-tertiary">(optional)</span></span
>
<span class="text-tertiary text-2xs"
>Only needed for non Azure Blob providers like Azurite</span
>
<input type="text" bind:value={bucket_config.endpoint} />
</label>
{:else}
<div>Unknown bucket type {bucket_config['type']}</div>
{/if}
{:else if bucket_config.type === 'Azure'}
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Account name</span>
<input type="text" placeholder="account-name" bind:value={bucket_config.accountName} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Container name</span>
<input
type="text"
placeholder="container-name"
bind:value={bucket_config.containerName}
/>
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm">Access key</span>
<input type="password" autocomplete="new-password" bind:value={bucket_config.accessKey} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Tenant ID <span class="text-2xs text-tertiary">(optional)</span></span
>
<input type="text" bind:value={bucket_config.tenantId} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Client ID <span class="text-2xs text-tertiary">(optional)</span></span
>
<input type="text" bind:value={bucket_config.clientId} />
</label>
<label class="block pb-2">
<span class="text-primary font-semibold text-sm"
>Endpoint <span class="text-2xs text-tertiary">(optional)</span></span
>
<span class="text-tertiary text-2xs"
>Only needed for non Azure Blob providers like Azurite</span
>
<input type="text" bind:value={bucket_config.endpoint} />
</label>
{:else}
<div>Unknown bucket type {bucket_config['type']}</div>
{/if}
</div>
</div>
{/if}
@@ -23,7 +23,7 @@
import { isCloudHosted } from '$lib/cloud'
import InstanceNameEditor from './InstanceNameEditor.svelte'
import Toggle from './Toggle.svelte'
import { instanceSettingsSelectedTab } from '$lib/stores';
import { instanceSettingsSelectedTab } from '$lib/stores'
let drawer: Drawer
let filter = ''
@@ -57,7 +57,7 @@
let tab: 'users' | string = 'users'
$: $instanceSettingsSelectedTab, tab = $instanceSettingsSelectedTab
$: $instanceSettingsSelectedTab, (tab = $instanceSettingsSelectedTab)
$: tab, instanceSettingsSelectedTab.set(tab)
let nbDisplayed = 50
@@ -133,7 +133,7 @@
</div>
<div class="pt-4 h-full">
<Tabs bind:selected={tab}>
<Tab value="users">Global users</Tab>
<Tab value="users">Users</Tab>
{#each settingsKeys as category}
<Tab value={category}>{category}</Tab>
{/each}
@@ -252,7 +252,7 @@
{/if}
<td>
<ToggleButtonGroup
selected={super_admin ? "super_admin" : devops ? "devops" : "user"}
selected={super_admin ? 'super_admin' : devops ? 'devops' : 'user'}
on:selected={async (e) => {
if (email == $userStore?.email) {
sendUserToast('You cannot demote yourself', true)
@@ -260,9 +260,9 @@
return
}
let role = e.detail;
let role = e.detail
if (role === "super_admin") {
if (role === 'super_admin') {
await UserService.globalUserUpdate({
email,
requestBody: {
@@ -271,7 +271,7 @@
}
})
}
if (role === "devops") {
if (role === 'devops') {
await UserService.globalUserUpdate({
email,
requestBody: {
@@ -279,9 +279,8 @@
is_devops: true
}
})
}
if (role === "user") {
if (role === 'user') {
await UserService.globalUserUpdate({
email,
requestBody: {
@@ -294,15 +293,20 @@
listUsers(activeOnly)
}}
>
<ToggleButton value={"user"} size="xs" label="User" />
<ToggleButton value={"devops"} size="xs" label="Devops" tooltip="Devops is a role that grants visibilty similar to that of a super admin, but without giving all rights. For example devops users can see service logs and crtical alerts. You can think of it as a 'readonly' super admin" />
<ToggleButton value={"super_admin"} size="xs" label="Superadmin" />
<ToggleButton value={'user'} size="xs" label="User" />
<ToggleButton
value={'devops'}
size="xs"
label="Devops"
tooltip="Devops is a role that grants visibilty similar to that of a super admin, but without giving all rights. For example devops users can see service logs and crtical alerts. You can think of it as a 'readonly' super admin"
/>
<ToggleButton value={'super_admin'} size="xs" label="Superadmin" />
</ToggleButtonGroup>
</td>
<td>
<div class="flex flex-row gap-x-1 justify-end">
<InstanceNameEditor
login_type={login_type}
{login_type}
value={name}
{username}
{email}
@@ -351,7 +355,13 @@
</div>
</TabContent>
<TabContent value="" values={settingsKeys}>
<InstanceSettings bind:this={instanceSettings} hideTabs hideSave {tab} {closeDrawer}/>
<InstanceSettings
bind:this={instanceSettings}
hideTabs
hideSave
{tab}
{closeDrawer}
/>
</TabContent>
</svelte:fragment>
</Tabs>
+100 -94
View File
@@ -7,20 +7,21 @@ export interface Setting {
tooltip?: string
key: string
fieldType:
| 'text'
| 'number'
| 'boolean'
| 'password'
| 'select'
| 'textarea'
| 'seconds'
| 'email'
| 'license_key'
| 'object_store_config'
| 'critical_error_channels'
| 'slack_connect'
| 'smtp_connect'
| 'indexer_rates'
| 'text'
| 'number'
| 'boolean'
| 'password'
| 'select'
| 'textarea'
| 'seconds'
| 'email'
| 'license_key'
| 'object_store_config'
| 'critical_error_channels'
| 'slack_connect'
| 'smtp_connect'
| 'indexer_rates'
| 'otel'
storage: SettingStorage
advancedToggle?: {
label: string
@@ -36,6 +37,27 @@ export interface Setting {
export type SettingStorage = 'setting'
export const scimSamlSetting: Setting[] = [
{
label: 'SCIM token',
description: 'Token used to authenticate requests from the IdP',
key: 'scim_token',
fieldType: 'text',
placeholder: 'mytoken',
storage: 'setting',
ee_only: ''
},
{
label: 'SAML metadata',
description: 'XML metadata url OR content for the SAML IdP',
key: 'saml_metadata',
fieldType: 'textarea',
placeholder: 'https://dev-2578259.okta.com/app/exkaell8gidiiUWrg5d7/sso/saml/metadata ',
storage: 'setting',
ee_only: ''
}
]
export const settings: Record<string, Setting[]> = {
Core: [
{
@@ -50,9 +72,9 @@ export const settings: Record<string, Setting[]> = {
isValid: (value: string | undefined) =>
value
? value?.startsWith('http') &&
value.includes('://') &&
!value?.endsWith('/') &&
!value?.endsWith(' ')
value.includes('://') &&
!value?.endsWith('/') &&
!value?.endsWith(' ')
: false
},
{
@@ -82,6 +104,13 @@ export const settings: Record<string, Setting[]> = {
storage: 'setting',
cloudonly: false
},
{
label: 'Keep job directories for debug',
key: 'keep_job_dir',
fieldType: 'boolean',
description: 'Keep Job directories after execution at /tmp/windmill/WORKER/JOB_ID',
storage: 'setting'
},
{
label: 'Max timeout for sync endpoints',
description:
@@ -107,7 +136,8 @@ export const settings: Record<string, Setting[]> = {
'Whether we should consider the reported usage of this instance as non-prod. <a href="https://www.windmill.dev/docs/advanced/instance_settings#non-prod-instance">Learn more</a>',
key: 'dev_instance',
fieldType: 'boolean',
storage: 'setting'
storage: 'setting',
ee_only: ''
},
{
label: 'Retention period in secs',
@@ -122,21 +152,14 @@ export const settings: Record<string, Setting[]> = {
},
{
label: 'Delete logs from s3 periodically',
description: 'Job and service logs are periodically deleted from disk. When this setting is on, they will also be deleted from the object storage.',
description:
'Job and service logs are periodically deleted from disk. When this setting is on, they will also be deleted from the object storage.',
key: 'monitor_logs_on_s3',
fieldType: 'boolean',
storage: 'setting',
ee_only: ''
},
{
label: 'Expose metrics',
description:
'Expose Prometheus metrics for workers and servers on port 8001 at /metrics. <a href="https://www.windmill.dev/docs/advanced/instance_settings#expose-metrics">Learn more</a>',
key: 'expose_metrics',
fieldType: 'boolean',
storage: 'setting',
ee_only: ''
},
{
label: 'Instance object storage',
description:
@@ -146,24 +169,7 @@ export const settings: Record<string, Setting[]> = {
storage: 'setting',
ee_only: ''
},
{
label: 'Critical alert channels',
description:
'Channels to send critical alerts to. SMTP must be configured for the email channel. A Slack workspace must be connected to the instance for the Slack channel. <a href="https://www.windmill.dev/docs/core_concepts/critical_alerts">Learn more</a>',
key: 'critical_error_channels',
fieldType: 'critical_error_channels',
storage: 'setting',
ee_only: 'Channels other than tracing are only available in the EE version'
},
{
label: 'Mute critical alerts in UI',
description: 'Enable to mute critical alerts in the UI',
key: 'critical_alert_mute_ui',
fieldType: 'boolean',
storage: 'setting',
requiresReloadOnChange: true,
ee_only: 'Critical alerts in UI are only available in the EE version'
},
{
label: 'Azure OpenAI base path',
description:
@@ -209,7 +215,7 @@ export const settings: Record<string, Setting[]> = {
requiresReloadOnChange: true
}
],
'SSO/OAuth': [],
'Auth/OAuth': [],
Registries: [
{
label: 'Pip Index Url',
@@ -249,7 +255,32 @@ export const settings: Record<string, Setting[]> = {
ee_only: ''
}
],
SMTP: [
Alerts: [
{
label: 'Critical alert channels',
description:
'Channels to send critical alerts to. SMTP and Slack must be configured below. <a href="https://www.windmill.dev/docs/core_concepts/critical_alerts">Learn more</a>',
key: 'critical_error_channels',
fieldType: 'critical_error_channels',
storage: 'setting',
ee_only: 'Channels other than tracing are only available in the EE version'
},
{
label: 'Mute critical alerts in UI',
description: 'Enable to mute critical alerts in the UI',
key: 'critical_alert_mute_ui',
fieldType: 'boolean',
storage: 'setting',
requiresReloadOnChange: true,
ee_only: 'Critical alerts in UI are only available in the EE version'
},
{
label: 'Slack',
key: 'slack',
fieldType: 'slack_connect',
storage: 'setting',
ee_only: ''
},
{
label: 'SMTP',
key: 'smtp_settings',
@@ -258,7 +289,26 @@ export const settings: Record<string, Setting[]> = {
ee_only: ''
}
],
'Indexer/Search': [
'OTEL/Prom': [
{
label: 'OpenTelemetry',
key: 'otel',
fieldType: 'otel',
storage: 'setting',
ee_only: ''
},
{
label: 'Prometheus',
description:
'Expose Prometheus metrics for workers and servers on port 8001 at /metrics. <a href="https://www.windmill.dev/docs/advanced/instance_settings#expose-metrics">Learn more</a>',
key: 'expose_metrics',
fieldType: 'boolean',
storage: 'setting',
ee_only: ''
}
],
Indexer: [
{
label: '',
key: 'indexer_settings',
@@ -267,51 +317,7 @@ export const settings: Record<string, Setting[]> = {
ee_only: 'Full text search across jobs and service logs is an EE feature'
}
],
Slack: [
{
label: 'Slack',
key: 'slack',
fieldType: 'slack_connect',
storage: 'setting',
ee_only: ''
}
],
'SCIM/SAML': [
{
label: 'SCIM token',
description: 'Token used to authenticate requests from the IdP',
key: 'scim_token',
fieldType: 'text',
placeholder: 'mytoken',
storage: 'setting',
ee_only: ''
},
{
label: 'SAML metadata',
description: 'XML metadata url OR content for the SAML IdP',
key: 'saml_metadata',
fieldType: 'textarea',
placeholder: 'https://dev-2578259.okta.com/app/exkaell8gidiiUWrg5d7/sso/saml/metadata ',
storage: 'setting',
ee_only: ''
}
],
Debug: [
{
label: 'Keep job directories',
key: 'keep_job_dir',
fieldType: 'boolean',
description: 'Keep Job directories after execution at /tmp/windmill/WORKER/JOB_ID',
storage: 'setting'
},
{
label: 'Expose debug metrics',
key: 'expose_debug_metrics',
fieldType: 'boolean',
description: 'Expose additional metrics (require metrics to be enabled)',
storage: 'setting'
}
],
Telemetry: [
{
label: 'Disable telemetry',