starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-21 00:02:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve support for singlescriptflow

Browse Source
This commit is contained in:
Ruben Fiszel
2024-02-23 21:06:55 +01:00
parent 3ba8db87eb
commit 2d6ba9528b
19 changed files with 192 additions and 632 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-6f65ce2598dfe64ee8b76ca66006d40b2b7853cfae1207de381c30aa9307405a.json
Generated
+34
View File
@@ -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"
}
backend/.sqlx/query-fc2ec6fc4e22e46cc35b00503eff1eac35a6fdbc82a634b7eb9da4d7931f3582.json
Generated
-82
View File
@@ -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"
}
backend/src/ee.rs
-16
View File
@@ -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(),
))
}
backend/src/ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/src/ee.rs
backend/src/monitor.rs
+4 -7
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
) {
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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
}
}
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()
backend/windmill-api/src/ee.rs
-6
View File
@@ -1,6 +0,0 @@
use anyhow::anyhow;
pub async fn validate_license_key(_license_key: String) -> anyhow::Result<String> {
// Implementation is not open source
Err(anyhow!("License can't be validated in Windmill CE"))
}
backend/windmill-api/src/ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/ee.rs
backend/windmill-api/src/job_helpers_ee.rs
-5
View File
@@ -1,5 +0,0 @@
use axum::Router;
pub fn workspaced_service() -> Router {
Router::new()
}
backend/windmill-api/src/job_helpers_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/job_helpers_ee.rs
backend/windmill-api/src/jobs.rs
+1 -1
View File
@@ -465,7 +465,7 @@ async fn get_flow_job_debug_info(
) -> error::Result<Response> {
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(),
backend/windmill-api/src/oauth2_ee.rs
-212
View File
@@ -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<String>,
extra_params: Option<HashMap<String, String>>,
_extra_params_callback: Option<HashMap<String, String>>,
_allowed_domains: Option<Vec<String>>,
_userinfo_url: Option<String>,
}
pub type BasicClientsMap = HashMap<String, ClientWithScopes>;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OAuthConfig {
auth_url: String,
token_url: String,
userinfo_url: Option<String>,
scopes: Option<Vec<String>>,
extra_params: Option<HashMap<String, String>>,
extra_params_callback: Option<HashMap<String, String>>,
req_body_auth: Option<bool>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OAuthClient {
id: String,
secret: String,
allowed_domains: Option<Vec<String>>,
connect_config: Option<OAuthConfig>,
login_config: Option<OAuthConfig>,
}
#[derive(Debug)]
pub struct AllClients {
pub logins: BasicClientsMap,
pub connects: BasicClientsMap,
pub slack: Option<OClient>,
}
pub fn build_oauth_clients(
_base_url: &str,
_oauths_from_config: Option<HashMap<String, OAuthClient>>,
) -> anyhow::Result<AllClients> {
// 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<u64>,
refresh_token: Option<RefreshToken>,
#[serde(deserialize_with = "helpers::deserialize_space_delimited_vec")]
#[serde(serialize_with = "helpers::serialize_space_delimited_vec")]
#[serde(default)]
scope: Option<Vec<Scope>>,
}
#[derive(Serialize)]
struct Logins {
oauth: Vec<String>,
saml: Option<String>,
}
async fn list_logins() -> error::JsonResult<Logins> {
// Implementation is not open source
return Ok(Json(Logins { oauth: vec![], saml: None }));
}
#[derive(Serialize)]
struct ScopesAndParams {
scopes: Vec<String>,
extra_params: Option<HashMap<String, String>>,
}
async fn list_connects() -> error::JsonResult<HashMap<String, ScopesAndParams>> {
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::<HashMap<String, ScopesAndParams>>(),
))
}
pub async fn _refresh_token<'c>(
_tx: Transaction<'c, Postgres>,
_path: &str,
_w_id: &str,
_id: i32,
) -> error::Result<String> {
// 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<S: AsRef<[u8]>>(secret: S) -> anyhow::Result<SlackVerifier> {
HmacSha256::new_from_slice(secret.as_ref())
.map(|mac| SlackVerifier { _mac: mac })
.map_err(|_| anyhow::anyhow!("invalid secret"))
}
}
backend/windmill-api/src/oauth2_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/oauth2_ee.rs
backend/windmill-api/src/oidc_ee.rs
-17
View File
@@ -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()
}
backend/windmill-api/src/oidc_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/oidc_ee.rs
backend/windmill-api/src/saml_ee.rs
-25
View File
@@ -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<ServiceProviderExt> {
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()
}
backend/windmill-api/src/saml_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/saml_ee.rs
backend/windmill-api/src/scim_ee.rs
-22
View File
@@ -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<B>(request: Request<B>, next: Next<B>) -> Response {
return next.run(request).await;
}
backend/windmill-api/src/scim_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/scim_ee.rs
backend/windmill-api/src/stripe_ee.rs
-7
View File
@@ -1,7 +0,0 @@
#[cfg(feature = "stripe")]
use axum::Router;
#[cfg(feature = "stripe")]
pub fn add_stripe_routes(router: Router) -> Router {
return router;
}
backend/windmill-api/src/stripe_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-api/src/stripe_ee.rs
backend/windmill-audit/src/audit_ee.rs
-48
View File
@@ -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<HashMap<&str, &str>>,
) -> 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<Vec<AuditLog>> {
// 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<AuditLog> {
// 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(),
))
}
backend/windmill-audit/src/audit_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-audit/src/audit_ee.rs
backend/windmill-common/src/ee.rs
-20
View File
@@ -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<RwLock<bool>> = Arc::new(RwLock::new(true));
pub static ref LICENSE_KEY_ID: Arc<RwLock<String>> = Arc::new(RwLock::new("".to_string()));
pub static ref LICENSE_KEY: Arc<RwLock<String>> = 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;
}
backend/windmill-common/src/ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-common/src/ee.rs
backend/windmill-common/src/jobs.rs
+4 -1
View File
@@ -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 {
backend/windmill-git-sync/src/git_sync_ee.rs
-17
View File
@@ -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<String>,
_rsmq: Option<R>,
_skip_db_insert: bool,
) -> Result<()> {
// Git sync is an enterprise feature and not part of the open-source version
return Ok(());
}
backend/windmill-git-sync/src/git_sync_ee.rs
Symlink
+1
View File
@@ -0,0 +1 @@
/git/windmill/../windmill-ee-private/windmill-git-sync/src/git_sync_ee.rs
backend/windmill-queue/src/jobs.rs
+5 -7
View File
@@ -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()
backend/windmill-worker/src/worker.rs
+132 -138
View File
@@ -2260,10 +2260,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
)
};
let update_job_future = if job.is_flow_step
|| job.job_kind == JobKind::FlowPreview
|| job.job_kind == JobKind::Flow
{
let update_job_future = if job.is_flow_step || job.is_flow() {
let (flow, job_status_to_update, update_job_future) =
if let Some(parent_job_id) = job.parent_job {
let _ = update_job_future().await;
@@ -2522,144 +2519,141 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
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<CanceledBy> = 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<CanceledBy> = 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(())
}
frontend/src/lib/components/ScheduleEditorInner.svelte
+1 -1
View File
@@ -393,7 +393,7 @@
View Runs
</Button>
</div>
<div class="mr-8 center-center -mt-2">
<div class="mr-8 center-center -mt-1">
<Toggle
disabled={!can_write}
checked={enabled}