From 2d6ba9528b9db0b030e5ee5e9e25aa86a17b273b Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 23 Feb 2024 21:06:42 +0100 Subject: [PATCH] fix: improve support for singlescriptflow --- ...6d40b2b7853cfae1207de381c30aa9307405a.json | 34 +++ ...f1eac35a6fdbc82a634b7eb9da4d7931f3582.json | 82 ------ backend/src/ee.rs | 17 +- backend/src/monitor.rs | 11 +- backend/windmill-api/src/ee.rs | 7 +- backend/windmill-api/src/job_helpers_ee.rs | 6 +- backend/windmill-api/src/jobs.rs | 2 +- backend/windmill-api/src/oauth2_ee.rs | 213 +------------- backend/windmill-api/src/oidc_ee.rs | 18 +- backend/windmill-api/src/saml_ee.rs | 26 +- backend/windmill-api/src/scim_ee.rs | 23 +- backend/windmill-api/src/stripe_ee.rs | 8 +- backend/windmill-audit/src/audit_ee.rs | 49 +--- backend/windmill-common/src/ee.rs | 21 +- backend/windmill-common/src/jobs.rs | 5 +- backend/windmill-git-sync/src/git_sync_ee.rs | 18 +- backend/windmill-queue/src/jobs.rs | 12 +- backend/windmill-worker/src/worker.rs | 270 +++++++++--------- .../lib/components/ScheduleEditorInner.svelte | 2 +- 19 files changed, 192 insertions(+), 632 deletions(-) create mode 100644 backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json delete mode 100644 backend/.sqlx/query-fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582.json mode change 100644 => 120000 backend/src/ee.rs mode change 100644 => 120000 backend/windmill-api/src/ee.rs mode change 100644 => 120000 backend/windmill-api/src/job_helpers_ee.rs mode change 100644 => 120000 backend/windmill-api/src/oauth2_ee.rs mode change 100644 => 120000 backend/windmill-api/src/oidc_ee.rs mode change 100644 => 120000 backend/windmill-api/src/saml_ee.rs mode change 100644 => 120000 backend/windmill-api/src/scim_ee.rs mode change 100644 => 120000 backend/windmill-api/src/stripe_ee.rs mode change 100644 => 120000 backend/windmill-audit/src/audit_ee.rs mode change 100644 => 120000 backend/windmill-common/src/ee.rs mode change 100644 => 120000 backend/windmill-git-sync/src/git_sync_ee.rs diff --git a/backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json b/backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json new file mode 100644 index 0000000000..d4289d89b2 --- /dev/null +++ b/backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval\n AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow' AND same_worker = false RETURNING id, workspace_id, last_ping", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "last_ping", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a" +} diff --git a/backend/.sqlx/query-fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582.json b/backend/.sqlx/query-fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582.json deleted file mode 100644 index a32600cede..0000000000 --- a/backend/.sqlx/query-fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582.json +++ /dev/null @@ -1,82 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND job_kind != $3 AND same_worker = false RETURNING id, workspace_id, last_ping", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - }, - { - "ordinal": 1, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 2, - "name": "last_ping", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [ - "Text", - { - "Custom": { - "name": "job_kind", - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies", - "http", - "graphql", - "postgresql", - "noop", - "appdependencies", - "deploymentcallback", - "singlescriptflow" - ] - } - } - }, - { - "Custom": { - "name": "job_kind", - "kind": { - "Enum": [ - "script", - "preview", - "flow", - "dependencies", - "flowpreview", - "script_hub", - "identity", - "flowdependencies", - "http", - "graphql", - "postgresql", - "noop", - "appdependencies", - "deploymentcallback", - "singlescriptflow" - ] - } - } - } - ] - }, - "nullable": [ - false, - false, - false - ] - }, - "hash": "fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582" -} diff --git a/backend/src/ee.rs b/backend/src/ee.rs deleted file mode 100644 index ef944b984b..0000000000 --- a/backend/src/ee.rs +++ /dev/null @@ -1,16 +0,0 @@ -use anyhow::anyhow; -#[cfg(feature = "enterprise")] -use windmill_common::error::{Error, Result}; - -pub async fn set_license_key(_license_key: String) -> anyhow::Result<()> { - // Implementation is not open source - Err(anyhow!("License cannot be set in Windmill CE")) -} - -#[cfg(feature = "enterprise")] -pub async fn verify_license_key() -> Result<()> { - // Implementation is not open source - Err(Error::InternalErr( - "License always invalid in Windmill CE".to_string(), - )) -} diff --git a/backend/src/ee.rs b/backend/src/ee.rs new file mode 120000 index 0000000000..ca288038c6 --- /dev/null +++ b/backend/src/ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/src/ee.rs \ No newline at end of file diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 45bfdcd67e..6fb92b9fd2 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -28,7 +28,7 @@ use windmill_common::{ REQUEST_SIZE_LIMIT_SETTING, REQUIRE_PREEXISTING_USER_FOR_OAUTH_SETTING, RETENTION_PERIOD_SECS_SETTING, SAML_METADATA_SETTING, SCIM_TOKEN_SETTING, }, - jobs::{JobKind, QueuedJob}, + jobs::QueuedJob, oauth2::REQUIRE_PREEXISTING_USER_FOR_OAUTH, server::load_server_config, users::truncate_token, @@ -766,10 +766,9 @@ async fn handle_zombie_jobs ) { if *RESTART_ZOMBIE_JOBS { let restarted = sqlx::query!( - "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND job_kind != $3 AND same_worker = false RETURNING id, workspace_id, last_ping", + "UPDATE queue SET running = false, started_at = null, logs = logs || '\nRestarted job after not receiving job''s ping for too long the ' || now() || '\n\n' WHERE last_ping < now() - ($1 || ' seconds')::interval + AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow' AND same_worker = false RETURNING id, workspace_id, last_ping", *ZOMBIE_JOB_TIMEOUT, - JobKind::Flow as JobKind, - JobKind::FlowPreview as JobKind, ) .fetch_all(db) .await @@ -789,14 +788,12 @@ async fn handle_zombie_jobs } } - let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != $2 AND job_kind != $3".to_string(); + let mut timeout_query = "SELECT * FROM queue WHERE last_ping < now() - ($1 || ' seconds')::interval AND running = true AND job_kind != 'flow' AND job_kind != 'flowpreview' AND job_kind != 'singlescriptflow'".to_string(); if *RESTART_ZOMBIE_JOBS { timeout_query.push_str(" AND same_worker = true"); }; let timeouts = sqlx::query_as::<_, QueuedJob>(&timeout_query) .bind(ZOMBIE_JOB_TIMEOUT.as_str()) - .bind(JobKind::Flow) - .bind(JobKind::FlowPreview) .fetch_all(db) .await .ok() diff --git a/backend/windmill-api/src/ee.rs b/backend/windmill-api/src/ee.rs deleted file mode 100644 index a9d170a3ac..0000000000 --- a/backend/windmill-api/src/ee.rs +++ /dev/null @@ -1,6 +0,0 @@ -use anyhow::anyhow; - -pub async fn validate_license_key(_license_key: String) -> anyhow::Result { - // Implementation is not open source - Err(anyhow!("License can't be validated in Windmill CE")) -} diff --git a/backend/windmill-api/src/ee.rs b/backend/windmill-api/src/ee.rs new file mode 120000 index 0000000000..0267c6e6fd --- /dev/null +++ b/backend/windmill-api/src/ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/job_helpers_ee.rs b/backend/windmill-api/src/job_helpers_ee.rs deleted file mode 100644 index 61a946bb84..0000000000 --- a/backend/windmill-api/src/job_helpers_ee.rs +++ /dev/null @@ -1,5 +0,0 @@ -use axum::Router; - -pub fn workspaced_service() -> Router { - Router::new() -} diff --git a/backend/windmill-api/src/job_helpers_ee.rs b/backend/windmill-api/src/job_helpers_ee.rs new file mode 120000 index 0000000000..8cc4fbe8bf --- /dev/null +++ b/backend/windmill-api/src/job_helpers_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/job_helpers_ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index d9556b3c7c..7d0c4c342a 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -465,7 +465,7 @@ async fn get_flow_job_debug_info( ) -> error::Result { let job = get_queued_job(id, w_id.as_str(), &db).await?; if let Some(job) = job { - let is_flow = &job.job_kind == &JobKind::FlowPreview || &job.job_kind == &JobKind::Flow; + let is_flow = job.is_flow(); if job.is_flow_step || !is_flow { return Err(error::Error::BadRequest( "This endpoint is only for root flow jobs".to_string(), diff --git a/backend/windmill-api/src/oauth2_ee.rs b/backend/windmill-api/src/oauth2_ee.rs deleted file mode 100644 index 032e0c9a58..0000000000 --- a/backend/windmill-api/src/oauth2_ee.rs +++ /dev/null @@ -1,212 +0,0 @@ -/* - * 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 std::{collections::HashMap, fmt::Debug}; - -use axum::body::StreamBody; -use axum::response::IntoResponse; -use axum::{routing::get, Json, Router}; -use hmac::Mac; -use hyper::{HeaderMap, StatusCode}; - -use oauth2::{Client as OClient, *}; -use serde::{Deserialize, Serialize}; -use sqlx::{Postgres, Transaction}; -use windmill_common::more_serde::maybe_number_opt; - -use crate::{HTTP_CLIENT, OAUTH_CLIENTS}; -use windmill_common::error::{self, to_anyhow}; -use windmill_common::oauth2::*; - -use crate::db::DB; -use std::str; - -pub fn global_service() -> Router { - Router::new() - .route("/list_supabase", get(list_supabase)) - .route("/list_logins", get(list_logins)) - .route("/list_connects", get(list_connects)) -} - -pub fn workspaced_service() -> Router { - Router::new() -} - -#[derive(Serialize)] -#[serde(tag = "type")] -pub enum InstanceEvent { - UserAdded { email: String }, - // UserDeleted { email: String }, - // UserDeletedWorkspace { workspace: String, email: String }, - UserAddedWorkspace { workspace: String, email: String }, - UserInvitedWorkspace { workspace: String, email: String }, - UserJoinedWorkspace { workspace: String, email: String, username: String }, -} - -#[derive(Debug, Clone)] -pub struct ClientWithScopes { - _client: OClient, - scopes: Vec, - extra_params: Option>, - _extra_params_callback: Option>, - _allowed_domains: Option>, - _userinfo_url: Option, -} - -pub type BasicClientsMap = HashMap; - -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct OAuthConfig { - auth_url: String, - token_url: String, - userinfo_url: Option, - scopes: Option>, - extra_params: Option>, - extra_params_callback: Option>, - req_body_auth: Option, -} - -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct OAuthClient { - id: String, - secret: String, - allowed_domains: Option>, - connect_config: Option, - login_config: Option, -} - -#[derive(Debug)] -pub struct AllClients { - pub logins: BasicClientsMap, - pub connects: BasicClientsMap, - pub slack: Option, -} - -pub fn build_oauth_clients( - _base_url: &str, - _oauths_from_config: Option>, -) -> anyhow::Result { - // Implementation is not open source - return Ok(AllClients { - logins: HashMap::default(), - connects: HashMap::default(), - slack: None, - }); -} - -#[derive(Clone, Debug, Deserialize, Serialize)] -pub struct TokenResponse { - access_token: AccessToken, - #[serde(deserialize_with = "maybe_number_opt")] - #[serde(default)] - expires_in: Option, - refresh_token: Option, - #[serde(deserialize_with = "helpers::deserialize_space_delimited_vec")] - #[serde(serialize_with = "helpers::serialize_space_delimited_vec")] - #[serde(default)] - scope: Option>, -} - -#[derive(Serialize)] -struct Logins { - oauth: Vec, - saml: Option, -} -async fn list_logins() -> error::JsonResult { - // Implementation is not open source - return Ok(Json(Logins { oauth: vec![], saml: None })); -} - -#[derive(Serialize)] -struct ScopesAndParams { - scopes: Vec, - extra_params: Option>, -} -async fn list_connects() -> error::JsonResult> { - Ok(Json( - (&OAUTH_CLIENTS.read().await.connects) - .into_iter() - .map(|(k, v)| { - ( - k.to_owned(), - ScopesAndParams { - scopes: v.scopes.clone(), - extra_params: v.extra_params.clone(), - }, - ) - }) - .collect::>(), - )) -} - -pub async fn _refresh_token<'c>( - _tx: Transaction<'c, Postgres>, - _path: &str, - _w_id: &str, - _id: i32, -) -> error::Result { - // Implementation is not open source - Err(error::Error::BadRequest( - "Not implemented in Windmill's Open Source repository".to_string(), - )) -} - -async fn list_supabase(headers: HeaderMap) -> impl IntoResponse { - let token = headers - .get("X-Supabase-Token") - .map(|x| x.to_str().unwrap_or("")) - .unwrap_or(""); - let resp = HTTP_CLIENT - .get("https://api.supabase.com/v1/projects") - .bearer_auth(token) - .send() - .await - .map_err(to_anyhow)?; - - let status_code = resp.status(); - let stream = resp.bytes_stream(); - - Ok((status_code, StreamBody::new(stream))) as error::Result<(StatusCode, StreamBody<_>)> -} - -pub async fn check_nb_of_user(db: &DB) -> error::Result<()> { - let nb_users_sso = - sqlx::query_scalar!("SELECT COUNT(*) FROM password WHERE login_type != 'password'",) - .fetch_one(db) - .await?; - if nb_users_sso.unwrap_or(0) >= 10 { - return Err(error::Error::BadRequest( - "You have reached the maximum number of oauth users accounts (10) without an enterprise license" - .to_string(), - )); - } - - let nb_users = sqlx::query_scalar!("SELECT COUNT(*) FROM password",) - .fetch_one(db) - .await?; - if nb_users.unwrap_or(0) >= 50 { - return Err(error::Error::BadRequest( - "You have reached the maximum number of accounts (50) without an enterprise license" - .to_string(), - )); - } - return Ok(()); -} - -#[derive(Clone, Debug)] -pub struct SlackVerifier { - _mac: HmacSha256, -} - -impl SlackVerifier { - pub fn new>(secret: S) -> anyhow::Result { - HmacSha256::new_from_slice(secret.as_ref()) - .map(|mac| SlackVerifier { _mac: mac }) - .map_err(|_| anyhow::anyhow!("invalid secret")) - } -} diff --git a/backend/windmill-api/src/oauth2_ee.rs b/backend/windmill-api/src/oauth2_ee.rs new file mode 120000 index 0000000000..1ab1665454 --- /dev/null +++ b/backend/windmill-api/src/oauth2_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/oauth2_ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/oidc_ee.rs b/backend/windmill-api/src/oidc_ee.rs deleted file mode 100644 index 248b990f54..0000000000 --- a/backend/windmill-api/src/oidc_ee.rs +++ /dev/null @@ -1,17 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2023 - * 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 axum::Router; - -pub fn global_service() -> Router { - Router::new() -} - -pub fn workspaced_service() -> Router { - Router::new() -} diff --git a/backend/windmill-api/src/oidc_ee.rs b/backend/windmill-api/src/oidc_ee.rs new file mode 120000 index 0000000000..1e902de563 --- /dev/null +++ b/backend/windmill-api/src/oidc_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/oidc_ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/saml_ee.rs b/backend/windmill-api/src/saml_ee.rs deleted file mode 100644 index b3f1d4653c..0000000000 --- a/backend/windmill-api/src/saml_ee.rs +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2023 - * 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. - */ -#![allow(non_snake_case)] - -use axum::{routing::post, Router}; - -pub struct ServiceProviderExt(); - -pub async fn build_sp_extension() -> anyhow::Result { - return Ok(ServiceProviderExt()); -} - -pub fn global_service() -> Router { - Router::new().route("/acs", post(acs)) -} - -pub async fn acs() -> String { - // Implementation is not open source as it is a Windmill Enterprise Edition feature - "SAML available only in enterprise version".to_string() -} diff --git a/backend/windmill-api/src/saml_ee.rs b/backend/windmill-api/src/saml_ee.rs new file mode 120000 index 0000000000..65286e3f12 --- /dev/null +++ b/backend/windmill-api/src/saml_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/saml_ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/scim_ee.rs b/backend/windmill-api/src/scim_ee.rs deleted file mode 100644 index 8552405296..0000000000 --- a/backend/windmill-api/src/scim_ee.rs +++ /dev/null @@ -1,22 +0,0 @@ -/* - * Author: Ruben Fiszel - * Copyright: Windmill Labs, Inc 2023 - * 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 axum::{middleware::Next, response::Response, routing::get, Router}; -use hyper::Request; - -pub fn global_service() -> Router { - Router::new().route("/ee", get(ee)) -} - -pub async fn ee() -> String { - return "Enterprise Edition".to_string(); -} - -pub async fn has_scim_token(request: Request, next: Next) -> Response { - return next.run(request).await; -} diff --git a/backend/windmill-api/src/scim_ee.rs b/backend/windmill-api/src/scim_ee.rs new file mode 120000 index 0000000000..cd27dc4526 --- /dev/null +++ b/backend/windmill-api/src/scim_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/scim_ee.rs \ No newline at end of file diff --git a/backend/windmill-api/src/stripe_ee.rs b/backend/windmill-api/src/stripe_ee.rs deleted file mode 100644 index 1aea2ecd2d..0000000000 --- a/backend/windmill-api/src/stripe_ee.rs +++ /dev/null @@ -1,7 +0,0 @@ -#[cfg(feature = "stripe")] -use axum::Router; - -#[cfg(feature = "stripe")] -pub fn add_stripe_routes(router: Router) -> Router { - return router; -} diff --git a/backend/windmill-api/src/stripe_ee.rs b/backend/windmill-api/src/stripe_ee.rs new file mode 120000 index 0000000000..a6c7d1230e --- /dev/null +++ b/backend/windmill-api/src/stripe_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-api/src/stripe_ee.rs \ No newline at end of file diff --git a/backend/windmill-audit/src/audit_ee.rs b/backend/windmill-audit/src/audit_ee.rs deleted file mode 100644 index 5bb52e7b3f..0000000000 --- a/backend/windmill-audit/src/audit_ee.rs +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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 std::collections::HashMap; - -use windmill_common::{ - error::{Error, Result}, - utils::Pagination, -}; - -use crate::{ActionKind, AuditLog, ListAuditLogQuery}; -use sqlx::{Postgres, Transaction}; - -#[tracing::instrument(level = "trace", skip_all)] -pub async fn audit_log<'c, E: sqlx::Executor<'c, Database = Postgres>>( - _db: E, - _username: &str, - mut _operation: &str, - _action_kind: ActionKind, - _w_id: &str, - mut _resource: Option<&str>, - _parameters: Option>, -) -> Result<()> { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature - Ok(()) -} - -pub async fn list_audit( - _tx: Transaction<'_, Postgres>, - _w_id: String, - _pagination: Pagination, - _lq: ListAuditLogQuery, -) -> Result> { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature - return Ok(vec![]); -} - -pub async fn get_audit(tx: Transaction<'_, Postgres>, _id: i32, _w_id: &str) -> Result { - // Implementation is not open source as Audit logs is a Windmill Enterprise Edition feature - tx.commit().await?; - Err(Error::NotFound( - "Audit log not not available in Windmill Community edition".to_string(), - )) -} diff --git a/backend/windmill-audit/src/audit_ee.rs b/backend/windmill-audit/src/audit_ee.rs new file mode 120000 index 0000000000..f8d4a81d03 --- /dev/null +++ b/backend/windmill-audit/src/audit_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-audit/src/audit_ee.rs \ No newline at end of file diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs deleted file mode 100644 index c942447bbc..0000000000 --- a/backend/windmill-common/src/ee.rs +++ /dev/null @@ -1,20 +0,0 @@ -use crate::ee::LicensePlan::Community; -use std::sync::Arc; -use tokio::sync::RwLock; - -lazy_static::lazy_static! { - pub static ref LICENSE_KEY_VALID: Arc> = Arc::new(RwLock::new(true)); - pub static ref LICENSE_KEY_ID: Arc> = Arc::new(RwLock::new("".to_string())); - pub static ref LICENSE_KEY: Arc> = Arc::new(RwLock::new("".to_string())); -} - -pub enum LicensePlan { - Community, - Pro, - Enterprise, -} - -pub async fn get_license_plan() -> LicensePlan { - // Implementation is not open source - return Community; -} diff --git a/backend/windmill-common/src/ee.rs b/backend/windmill-common/src/ee.rs new file mode 120000 index 0000000000..3220066a5d --- /dev/null +++ b/backend/windmill-common/src/ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-common/src/ee.rs \ No newline at end of file diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 451d8fdb71..5e3f770098 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -117,7 +117,10 @@ impl QueuedJob { .unwrap_or("tmp/main") } pub fn is_flow(&self) -> bool { - matches!(self.job_kind, JobKind::Flow | JobKind::FlowPreview) + matches!( + self.job_kind, + JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow + ) } pub fn full_path_with_workspace(&self) -> String { diff --git a/backend/windmill-git-sync/src/git_sync_ee.rs b/backend/windmill-git-sync/src/git_sync_ee.rs deleted file mode 100644 index 4be2858c27..0000000000 --- a/backend/windmill-git-sync/src/git_sync_ee.rs +++ /dev/null @@ -1,17 +0,0 @@ -use windmill_common::error::Result; - -use crate::{DeployedObject, DB}; - -pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send + Clone + 'c>( - _email: &str, - _created_by: &str, - _db: &DB, - _w_id: &str, - _obj: DeployedObject, - _deployment_message: Option, - _rsmq: Option, - _skip_db_insert: bool, -) -> Result<()> { - // Git sync is an enterprise feature and not part of the open-source version - return Ok(()); -} diff --git a/backend/windmill-git-sync/src/git_sync_ee.rs b/backend/windmill-git-sync/src/git_sync_ee.rs new file mode 120000 index 0000000000..0824f2c28c --- /dev/null +++ b/backend/windmill-git-sync/src/git_sync_ee.rs @@ -0,0 +1 @@ +/git/windmill/../windmill-ee-private/windmill-git-sync/src/git_sync_ee.rs \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 1ad6fd2f9a..72fb0dab5d 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -138,8 +138,7 @@ pub async fn cancel_job<'c: 'async_recursion>( } let job_running = job_running.unwrap(); - if ((job_running.running || job_running.root_job.is_some()) - || (job_running.job_kind == JobKind::Flow || job_running.job_kind == JobKind::FlowPreview)) + if ((job_running.running || job_running.root_job.is_some()) || (job_running.is_flow())) && !force_cancel { let id = sqlx::query_scalar!( @@ -444,8 +443,7 @@ pub async fn add_completed_job< )); } - let is_flow = - queued_job.job_kind == JobKind::Flow || queued_job.job_kind == JobKind::FlowPreview; + let is_flow = queued_job.is_flow(); let duration = if is_flow { let jobs = queued_job.parse_flow_status().map(|s| { let mut modules = s.modules; @@ -574,8 +572,7 @@ pub async fn add_completed_job< .await?; } if !queued_job.is_flow_step - && queued_job.job_kind != JobKind::Flow - && queued_job.job_kind != JobKind::FlowPreview + && !queued_job.is_flow() && queued_job.schedule_path.is_some() && queued_job.script_path.is_some() { @@ -2051,7 +2048,7 @@ async fn compute_leaf_jobs_for_completed_flow( } } } - JobKind::Flow | JobKind::FlowPreview => { + JobKind::Flow | JobKind::FlowPreview | JobKind::SingleScriptFlow => { // Extract the leaf job for this flow and add them to the result map for module in &flow_job_status_modules { // we add the module as an element of ListJob for this step ID and recursiively extract leaf job of the sub-flow @@ -2999,6 +2996,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection let default = || { let ntag = if job_kind == JobKind::Flow || job_kind == JobKind::FlowPreview + || job_kind == JobKind::SingleScriptFlow || job_kind == JobKind::Identity { "flow".to_string() diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 82b11b4349..b519ced1a8 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2260,10 +2260,7 @@ pub async fn handle_job_error( return Ok(()); } }; - match job.job_kind { - JobKind::FlowPreview | JobKind::Flow | JobKind::SingleScriptFlow => { - let timer = worker_flow_initial_transition_duration.map(|x| x.start_timer()); - handle_flow( + if job.is_flow() { + let timer = worker_flow_initial_transition_duration.map(|x| x.start_timer()); + handle_flow( + &job, + db, + &client.get_authed().await, + None, + same_worker_tx, + worker_dir, + rsmq, + job_completed_tx.0.clone(), + ) + .await?; + timer.map(|x| x.stop_and_record()); + } else { + let mut logs = "".to_string(); + let mut mem_peak: i32 = 0; + let mut canceled_by: Option = None; + // println!("handle queue {:?}", SystemTime::now()); + if let Some(log_str) = &job.logs { + logs.push_str(&log_str); + logs.push_str("\n"); + } + + logs.push_str(&format!( + "job {} on worker {} (tag: {})\n", + &job.id, &worker_name, &job.tag + )); + + #[cfg(not(feature = "enterprise"))] + if job.concurrent_limit.is_some() { + logs.push_str("---\n"); + logs.push_str("WARNING: This job has concurrency limits enabled. Concurrency limits are going to become an Enterprise Edition feature in the near future.\n"); + logs.push_str("---\n"); + } + + tracing::debug!( + worker = %worker_name, + job_id = %job.id, + workspace_id = %job.workspace_id, + "handling job {}", + job.id + ); + + let result = match job.job_kind { + JobKind::Dependencies => { + handle_dependency_job( + &job, + &mut logs, + &mut mem_peak, + &mut canceled_by, + job_dir, + db, + worker_name, + worker_dir, + base_internal_url, + &client.get_token().await, + rsmq.clone(), + ) + .await + } + JobKind::FlowDependencies => handle_flow_dependency_job( &job, - db, - &client.get_authed().await, - None, - same_worker_tx, - worker_dir, - rsmq, - job_completed_tx.0.clone(), - ) - .await?; - timer.map(|x| x.stop_and_record()); - } - _ => { - let mut logs = "".to_string(); - let mut mem_peak: i32 = 0; - let mut canceled_by: Option = None; - // println!("handle queue {:?}", SystemTime::now()); - if let Some(log_str) = &job.logs { - logs.push_str(&log_str); - logs.push_str("\n"); - } - - logs.push_str(&format!( - "job {} on worker {} (tag: {})\n", - &job.id, &worker_name, &job.tag - )); - - #[cfg(not(feature = "enterprise"))] - if job.concurrent_limit.is_some() { - logs.push_str("---\n"); - logs.push_str("WARNING: This job has concurrency limits enabled. Concurrency limits are going to become an Enterprise Edition feature in the near future.\n"); - logs.push_str("---\n"); - } - - tracing::debug!( - worker = %worker_name, - job_id = %job.id, - workspace_id = %job.workspace_id, - "handling job {}", - job.id - ); - - let result = match job.job_kind { - JobKind::Dependencies => { - handle_dependency_job( - &job, - &mut logs, - &mut mem_peak, - &mut canceled_by, - job_dir, - db, - worker_name, - worker_dir, - base_internal_url, - &client.get_token().await, - rsmq.clone(), - ) - .await - } - JobKind::FlowDependencies => handle_flow_dependency_job( - &job, - &mut logs, - &mut mem_peak, - &mut canceled_by, - job_dir, - db, - worker_name, - worker_dir, - base_internal_url, - &client.get_token().await, - rsmq.clone(), - ) - .await - .map(|()| serde_json::from_str("{}").unwrap()), - JobKind::AppDependencies => handle_app_dependency_job( - &job, - &mut logs, - &mut mem_peak, - &mut canceled_by, - job_dir, - db, - worker_name, - worker_dir, - base_internal_url, - &client.get_token().await, - rsmq.clone(), - ) - .await - .map(|()| serde_json::from_str("{}").unwrap()), - JobKind::Identity => Ok(job - .args - .as_ref() - .map(|x| x.get("previous_result")) - .flatten() - .map(|x| x.to_owned()) - .unwrap_or_else(|| serde_json::from_str("{}").unwrap())), - _ => { - let timer = worker_code_execution_duration.map(|x| x.start_timer()); - let r = handle_code_execution_job( - job.as_ref(), - db, - client, - job_dir, - worker_dir, - &mut logs, - &mut mem_peak, - &mut canceled_by, - base_internal_url, - worker_name, - ) - .await; - timer.map(|x| x.stop_and_record()); - r - } - }; - - //it's a test job, no need to update the db - if job.as_ref().workspace_id == "" { - return Ok(()); - } - process_result( - job, - result, + &mut logs, + &mut mem_peak, + &mut canceled_by, job_dir, - job_completed_tx, - logs, - mem_peak, - canceled_by, - cached_res_path, - client.get_token().await, + db, + worker_name, + worker_dir, + base_internal_url, + &client.get_token().await, + rsmq.clone(), ) - .await?; + .await + .map(|()| serde_json::from_str("{}").unwrap()), + JobKind::AppDependencies => handle_app_dependency_job( + &job, + &mut logs, + &mut mem_peak, + &mut canceled_by, + job_dir, + db, + worker_name, + worker_dir, + base_internal_url, + &client.get_token().await, + rsmq.clone(), + ) + .await + .map(|()| serde_json::from_str("{}").unwrap()), + JobKind::Identity => Ok(job + .args + .as_ref() + .map(|x| x.get("previous_result")) + .flatten() + .map(|x| x.to_owned()) + .unwrap_or_else(|| serde_json::from_str("{}").unwrap())), + _ => { + let timer = worker_code_execution_duration.map(|x| x.start_timer()); + let r = handle_code_execution_job( + job.as_ref(), + db, + client, + job_dir, + worker_dir, + &mut logs, + &mut mem_peak, + &mut canceled_by, + base_internal_url, + worker_name, + ) + .await; + timer.map(|x| x.stop_and_record()); + r + } + }; + + //it's a test job, no need to update the db + if job.as_ref().workspace_id == "" { + return Ok(()); } - } + process_result( + job, + result, + job_dir, + job_completed_tx, + logs, + mem_peak, + canceled_by, + cached_res_path, + client.get_token().await, + ) + .await?; + }; Ok(()) } diff --git a/frontend/src/lib/components/ScheduleEditorInner.svelte b/frontend/src/lib/components/ScheduleEditorInner.svelte index 53a897c305..cb0a98fe30 100644 --- a/frontend/src/lib/components/ScheduleEditorInner.svelte +++ b/frontend/src/lib/components/ScheduleEditorInner.svelte @@ -393,7 +393,7 @@ View Runs -
+