diff --git a/backend/src/main.rs b/backend/src/main.rs index 89e43e4930..855262c5d5 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -34,7 +34,7 @@ use windmill_common::ee_oss::{ }; use windmill_common::{ - agent_workers::build_agent_http_client, + agent_workers::AgentConfig, global_settings::{ APP_WORKSPACED_ROUTE_SETTING, BASE_URL_SETTING, BUNFIG_INSTALL_SCOPES_SETTING, CRITICAL_ALERTS_ON_DB_OVERSIZE_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, @@ -43,13 +43,14 @@ use windmill_common::{ EXPOSE_DEBUG_METRICS_SETTING, EXPOSE_METRICS_SETTING, EXTRA_PIP_INDEX_URL_SETTING, HUB_API_SECRET_SETTING, HUB_BASE_URL_SETTING, INDEXER_SETTING, INSTANCE_PYTHON_VERSION_SETTING, JOB_DEFAULT_TIMEOUT_SECS_SETTING, JWT_SECRET_SETTING, - KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING, OTEL_TRACING_PROXY_SETTING, + KEEP_JOB_DIR_SETTING, LICENSE_KEY_SETTING, MAVEN_REPOS_SETTING, MONITOR_LOGS_ON_OBJECT_STORE_SETTING, NO_DEFAULT_MAVEN_SETTING, NPM_CONFIG_REGISTRY_SETTING, NUGET_CONFIG_SETTING, OAUTH_SETTING, OTEL_SETTING, - PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING, POWERSHELL_REPO_URL_SETTING, - REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, - RETENTION_PERIOD_SECS_SETTING, RUBY_REPOS_SETTING, SAML_METADATA_SETTING, - SCIM_TOKEN_SETTING, SMTP_SETTING, TEAMS_SETTING, TIMEOUT_WAIT_RESULT_SETTING, + OTEL_TRACING_PROXY_SETTING, PIP_INDEX_URL_SETTING, POWERSHELL_REPO_PAT_SETTING, + POWERSHELL_REPO_URL_SETTING, REQUEST_SIZE_LIMIT_SETTING, + REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, + RUBY_REPOS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, SMTP_SETTING, TEAMS_SETTING, + TIMEOUT_WAIT_RESULT_SETTING, }, scripts::ScriptLang, stats_oss::schedule_stats, @@ -99,9 +100,10 @@ use crate::monitor::{ reload_bunfig_install_scopes_setting, reload_critical_alert_mute_ui_setting, reload_critical_error_channels_setting, reload_extra_pip_index_url_setting, reload_hub_api_secret_setting, reload_hub_base_url_setting, reload_job_default_timeout_setting, - reload_jwt_secret_setting, reload_license_key, reload_otel_tracing_proxy_setting, - reload_npm_config_registry_setting, reload_pip_index_url_setting, reload_retention_period_setting, - reload_scim_token_setting, reload_smtp_config, reload_worker_config, MonitorIteration, + reload_jwt_secret_setting, reload_license_key, reload_npm_config_registry_setting, + reload_otel_tracing_proxy_setting, reload_pip_index_url_setting, + reload_retention_period_setting, reload_scim_token_setting, reload_smtp_config, + reload_worker_config, MonitorIteration, }; #[cfg(feature = "parquet")] @@ -410,7 +412,10 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyh let cache_path = format!("{}/{}", HUB_RT_CACHE_DIR, HUB_RT_CACHE_FILE); if tokio::fs::metadata(&cache_path).await.is_err() { - tracing::info!("No cached resource types found at {}, skipping sync", cache_path); + tracing::info!( + "No cached resource types found at {}, skipping sync", + cache_path + ); return Ok(()); } @@ -420,8 +425,8 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyh .await .with_context(|| format!("Failed to read cache file from {}", cache_path))?; - let cached_types: Vec = serde_json::from_str(&content) - .with_context(|| "Failed to parse cached resource types")?; + let cached_types: Vec = + serde_json::from_str(&content).with_context(|| "Failed to parse cached resource types")?; tracing::info!("Found {} cached resource types", cached_types.len()); @@ -433,11 +438,13 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyh .await .with_context(|| "Failed to fetch existing resource types")?; - let existing_map: std::collections::HashMap, Option)> = - existing_types - .into_iter() - .map(|(name, schema, desc)| (name, (schema, desc))) - .collect(); + let existing_map: std::collections::HashMap< + String, + (Option, Option), + > = existing_types + .into_iter() + .map(|(name, schema, desc)| (name, (schema, desc))) + .collect(); let mut synced_count = 0; let mut skipped_count = 0; @@ -478,36 +485,45 @@ pub async fn sync_cached_resource_types(db: &sqlx::Pool) -> anyh } fn print_help() { - println!("Windmill - a fast, open-source workflow engine and job runner."); - println!(); - println!("Usage:"); - println!(" windmill [SUBCOMMAND]"); - println!(); - println!("Subcommands:"); - println!(" help | -h | --help Show this help information and exit"); - println!(" version Show Windmill version and exit"); - println!(" cache [hubPaths.json] Pre-cache hub scripts (default: ./hubPaths.json)"); - println!(" cache-rt Pre-cache hub resource types"); - println!(); - println!("Environment variables (name = default):"); - println!(" DATABASE_URL = The Postgres database url."); - println!(" MODE = standalone Mode: standalone | worker | server | agent"); - println!(" BASE_URL = http://localhost:8000 Public base URL of your instance (overridden by instance settings)"); - println!(" PORT = {} HTTP port (server/indexer/MCP modes)", DEFAULT_PORT); - println!(" SERVER_BIND_ADDR = IP to bind to (server: {}, worker: {})", DEFAULT_SERVER_BIND_ADDR, DEFAULT_WORKER_BIND_ADDR); - println!(" NUM_WORKERS = {} Number of workers (standalone/worker modes)", DEFAULT_NUM_WORKERS); - println!(" WORKER_GROUP = default Worker group this worker belongs to",); - println!(" JSON_FMT = false Output logs in JSON instead of logfmt"); - println!(" METRICS_ADDR = None (EE only) Prometheus metrics addr at /metrics; set \"true\" to use :8001"); - println!(" SUPERADMIN_SECRET = None Virtual superadmin token (server)"); - println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)"); - println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup"); - println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool"); - println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start"); - println!(); - println!("Notes:"); - println!("- Advanced and less commonly used settings are managed via the database and are omitted here."); - println!("- At startup, Windmill logs currently set configuration keys for visibility."); + println!("Windmill - a fast, open-source workflow engine and job runner."); + println!(); + println!("Usage:"); + println!(" windmill [SUBCOMMAND]"); + println!(); + println!("Subcommands:"); + println!(" help | -h | --help Show this help information and exit"); + println!(" version Show Windmill version and exit"); + println!(" cache [hubPaths.json] Pre-cache hub scripts (default: ./hubPaths.json)"); + println!(" cache-rt Pre-cache hub resource types"); + println!(); + println!("Environment variables (name = default):"); + println!(" DATABASE_URL = The Postgres database url."); + println!(" MODE = standalone Mode: standalone | worker | server | agent"); + println!(" BASE_URL = http://localhost:8000 Public base URL of your instance (overridden by instance settings)"); + println!( + " PORT = {} HTTP port (server/indexer/MCP modes)", + DEFAULT_PORT + ); + println!( + " SERVER_BIND_ADDR = IP to bind to (server: {}, worker: {})", + DEFAULT_SERVER_BIND_ADDR, DEFAULT_WORKER_BIND_ADDR + ); + println!( + " NUM_WORKERS = {} Number of workers (standalone/worker modes)", + DEFAULT_NUM_WORKERS + ); + println!(" WORKER_GROUP = default Worker group this worker belongs to",); + println!(" JSON_FMT = false Output logs in JSON instead of logfmt"); + println!(" METRICS_ADDR = None (EE only) Prometheus metrics addr at /metrics; set \"true\" to use :8001"); + println!(" SUPERADMIN_SECRET = None Virtual superadmin token (server)"); + println!(" LICENSE_KEY = None (EE only) Enterprise license key (workers require valid key)"); + println!(" RUN_UPDATE_CA_CERTIFICATE_AT_START = false Run system CA update at startup"); + println!(" RUN_UPDATE_CA_CERTIFICATE_PATH = /usr/sbin/update-ca-certificates Path to CA update tool"); + println!(" SYNC_CACHED_RT = false Sync cached resource types to admins workspace on server start"); + println!(); + println!("Notes:"); + println!("- Advanced and less commonly used settings are managed via the database and are omitted here."); + println!("- At startup, Windmill logs currently set configuration keys for visibility."); } async fn windmill_main() -> anyhow::Result<()> { @@ -641,15 +657,23 @@ async fn windmill_main() -> anyhow::Result<()> { .and_then(|x| x.parse().ok()) .unwrap_or(IpAddr::from(default_bind_addr)); - let (conn, first_suffix) = if mode == Mode::Agent { + let (conn, first_suffix, agent_config) = if mode == Mode::Agent { + let agent_config = match AgentConfig::from_env() { + Ok(config) => config, + Err(e) => { + tracing::error!("{e}"); + std::process::exit(1); + } + }; tracing::info!( "Creating http client for cluster using base internal url {}", - std::env::var("BASE_INTERNAL_URL").unwrap_or_default() + agent_config.base_internal_url ); let suffix = create_default_worker_suffix(&hostname); ( - Connection::Http(build_agent_http_client(&suffix, None, None)), + Connection::Http(agent_config.build_http_client(&suffix)), Some(suffix), + Some(agent_config), ) } else { println!("Connecting to database..."); @@ -673,7 +697,8 @@ async fn windmill_main() -> anyhow::Result<()> { reload_otel_tracing_proxy_setting(&Connection::Sql(db.clone())).await; #[cfg(feature = "deno_core")] - if windmill_worker::is_otel_tracing_proxy_enabled_for_lang(&ScriptLang::Nativets).await { + if windmill_worker::is_otel_tracing_proxy_enabled_for_lang(&ScriptLang::Nativets).await + { match windmill_worker::load_internal_otel_exporter().await { Ok(()) => { tracing::info!("Internal OTEL exporter initialized for nativets tracing"); @@ -688,7 +713,7 @@ async fn windmill_main() -> anyhow::Result<()> { load_otel(&db).await; println!("Database connected"); - (Connection::Sql(db), None) + (Connection::Sql(db), None, None) }; let environment = if let Ok(environment) = std::env::var("OTEL_ENVIRONMENT") { @@ -799,6 +824,12 @@ Windmill Community Edition {GIT_VERSION} // if key still invalid and num_workers > 0, set to 0 if let Err(err) = reload_license_key(&conn).await { tracing::error!("Failed to reload license key: {err:#}"); + if is_agent { + tracing::error!( + "Agent worker cannot connect to server. Please check AGENT_TOKEN and BASE_INTERNAL_URL" + ); + std::process::exit(1); + } } let valid_key = *LICENSE_KEY_VALID.read().await; if !valid_key && !server_mode { @@ -1057,7 +1088,12 @@ Windmill Community Edition {GIT_VERSION} conn: if i == 0 || mode != Mode::Agent { conn.clone() } else { - Connection::Http(build_agent_http_client(&suffix, None, None)) + Connection::Http( + agent_config + .as_ref() + .expect("agent_config must be set in agent mode") + .build_http_client(&suffix), + ) }, worker_name: worker_name_with_suffix( mode == Mode::Agent, diff --git a/backend/tests/common/mod.rs b/backend/tests/common/mod.rs index 86b7443396..2279b2c53c 100644 --- a/backend/tests/common/mod.rs +++ b/backend/tests/common/mod.rs @@ -764,23 +764,25 @@ pub async fn run_preview_relative_imports( #[cfg(all(feature = "private", feature = "agent_worker_server"))] pub async fn testing_http_connection(port: u16) -> Connection { let suffix = windmill_common::utils::create_default_worker_suffix("test-agent-worker"); + let agent_token = format!( + "{}{}", + windmill_common::agent_workers::AGENT_JWT_PREFIX, + windmill_common::jwt::encode_with_internal_secret( + windmill_api::agent_workers_ee::AgentAuth { + worker_group: "testing-agent".to_owned(), + suffix: Some(suffix.clone()), + tags: vec!["flow".into(), "python3".into(), "dependency".into()], + exp: Some(usize::MAX), + } + ) + .await + .expect("JWT token to be created") + ); + let base_internal_url = format!("http://localhost:{port}"); Connection::Http(windmill_common::agent_workers::build_agent_http_client( &suffix, - Some(format!( - "{}{}", - windmill_common::agent_workers::AGENT_JWT_PREFIX, - windmill_common::jwt::encode_with_internal_secret( - windmill_api::agent_workers_ee::AgentAuth { - worker_group: "testing-agent".to_owned(), - suffix: Some(suffix.clone()), - tags: vec!["flow".into(), "python3".into(), "dependency".into()], - exp: Some(usize::MAX), - } - ) - .await - .expect("JWT token to be created") - )), - Some(format!("http://localhost:{port}")), + &agent_token, + &base_internal_url, )) } diff --git a/backend/windmill-common/src/agent_workers.rs b/backend/windmill-common/src/agent_workers.rs index c21deec70d..b501e78fd9 100644 --- a/backend/windmill-common/src/agent_workers.rs +++ b/backend/windmill-common/src/agent_workers.rs @@ -13,14 +13,55 @@ use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware}; use crate::{jwt::decode_without_verify, utils::configure_client, worker::HttpClient}; +/// Configuration required for agent mode. Both fields are mandatory when running in agent mode. +#[derive(Clone)] +pub struct AgentConfig { + pub agent_token: String, + pub base_internal_url: String, +} + +impl AgentConfig { + pub fn from_env() -> Result { + let agent_token = std::env::var("AGENT_TOKEN") + .map_err(|_| AgentConfigError::MissingAgentToken)?; + let base_internal_url = std::env::var("BASE_INTERNAL_URL") + .map_err(|_| AgentConfigError::MissingBaseInternalUrl)?; + Ok(Self { agent_token, base_internal_url }) + } + + pub fn build_http_client(&self, worker_suffix: &str) -> HttpClient { + build_agent_http_client(worker_suffix, &self.agent_token, &self.base_internal_url) + } +} + +#[derive(Debug)] +pub enum AgentConfigError { + MissingAgentToken, + MissingBaseInternalUrl, +} + +impl std::fmt::Display for AgentConfigError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + AgentConfigError::MissingAgentToken => { + write!(f, "AGENT_TOKEN environment variable is not set but required for agent mode") + } + AgentConfigError::MissingBaseInternalUrl => { + write!(f, "BASE_INTERNAL_URL environment variable is not set but required for agent mode") + } + } + } +} + +impl std::error::Error for AgentConfigError {} + lazy_static! { - pub static ref AGENT_TOKEN: String = std::env::var("AGENT_TOKEN").unwrap_or_default(); pub static ref DECODED_AGENT_TOKEN: Option = { - if AGENT_TOKEN.is_empty() { - None + let agent_token = std::env::var("AGENT_TOKEN"); + if let Ok(token) = agent_token { + decode_without_verify::(token.trim_start_matches(AGENT_JWT_PREFIX)).ok() } else { - decode_without_verify::(AGENT_TOKEN.trim_start_matches(AGENT_JWT_PREFIX)) - .ok() + None } }; } @@ -28,7 +69,7 @@ lazy_static! { #[derive(Clone, Debug, Deserialize, Serialize)] pub struct AgentAuth { pub worker_group: String, -pub suffix: Option, + pub suffix: Option, pub tags: Vec, pub exp: Option, } @@ -37,8 +78,8 @@ pub const AGENT_JWT_PREFIX: &str = "jwt_agent_"; pub fn build_agent_http_client( worker_suffix: &str, - agent_token: Option, - base_internal_url: Option, + agent_token: &str, + base_internal_url: &str, ) -> HttpClient { let client = ClientBuilder::new( configure_client( @@ -58,9 +99,7 @@ pub fn build_agent_http_client( "{}{}_{}", AGENT_JWT_PREFIX, worker_suffix, - agent_token - .unwrap_or(AGENT_TOKEN.clone()) - .trim_start_matches(AGENT_JWT_PREFIX) + agent_token.trim_start_matches(AGENT_JWT_PREFIX) ); headers.insert( "Authorization", @@ -76,7 +115,7 @@ pub fn build_agent_http_client( )) .build(); - HttpClient { client, base_internal_url } + HttpClient { client, base_internal_url: base_internal_url.to_string() } } #[derive(Deserialize, Serialize)] diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 1ac3f6c1bb..4b2794c399 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -151,10 +151,6 @@ lazy_static::lazy_static! { pub static ref HUB_API_SECRET: Arc>> = Arc::new(RwLock::new(None)); } -lazy_static::lazy_static! { - pub static ref AGENT_TOKEN: String = std::env::var("AGENT_TOKEN").unwrap_or_default(); -} - #[derive(Clone)] pub struct ModeAndAddons { pub indexer: bool, @@ -1190,7 +1186,8 @@ mod tests { #[test] fn test_merge_nested_raw_values_to_array_complex_types() { - let val1 = serde_json::value::RawValue::from_string("{\"name\":\"Alice\"}".to_string()).unwrap(); + let val1 = + serde_json::value::RawValue::from_string("{\"name\":\"Alice\"}".to_string()).unwrap(); let val2 = serde_json::value::RawValue::from_string("[1,2,3]".to_string()).unwrap(); let val3 = serde_json::value::RawValue::from_string("\"text\"".to_string()).unwrap(); let val4 = serde_json::value::RawValue::from_string("null".to_string()).unwrap(); @@ -1236,7 +1233,13 @@ mod tests { let inner3 = vec![val3]; let inner4 = vec![val4]; let inner5 = vec![val5]; - let nested = vec![inner1.iter(), inner2.iter(), inner3.iter(), inner4.iter(), inner5.iter()]; + let nested = vec![ + inner1.iter(), + inner2.iter(), + inner3.iter(), + inner4.iter(), + inner5.iter(), + ]; let result = merge_nested_raw_values_to_array(nested.into_iter()); diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 7934890f25..0b4b25472c 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -274,13 +274,10 @@ lazy_static::lazy_static! { pub const ROOT_CACHE_NOMOUNT_DIR: &str = concatcp!(TMP_DIR, "/cache_nomount/"); pub static MIN_VERSION_IS_LATEST: AtomicBool = AtomicBool::new(false); - -const DEFAULT_BASE_INTERNAL_URL: &str = "http://localhost:8000"; - #[derive(Clone)] pub struct HttpClient { pub client: ClientWithMiddleware, - pub base_internal_url: Option, + pub base_internal_url: String, } impl Deref for HttpClient { @@ -298,10 +295,7 @@ impl HttpClient { headers: Option, body: &T, ) -> anyhow::Result { - let base_url = self - .base_internal_url - .clone() - .unwrap_or(DEFAULT_BASE_INTERNAL_URL.to_owned()); + let base_url = self.base_internal_url.clone(); let response_builder = self.client.post(format!("{}{}", base_url, url)).json(body); @@ -327,11 +321,7 @@ impl HttpClient { } pub async fn get(&self, url: &str) -> anyhow::Result { - let base_url = self - .base_internal_url - .clone() - .unwrap_or(DEFAULT_BASE_INTERNAL_URL.to_owned()); - + let base_url = self.base_internal_url.clone(); let response = self .client .get(format!("{}{}", base_url, url))