From 62c33eec83d6790948624a7f2c4bb58bf400d34c Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 15 Jul 2023 19:28:40 +0200 Subject: [PATCH] feat: add schedule to syncable resources --- backend/windmill-api/src/jobs.rs | 10 ++ backend/windmill-api/src/schedule.rs | 4 +- backend/windmill-api/src/workspaces.rs | 23 ++++ backend/windmill-common/src/schedule.rs | 1 + cli/folder.ts | 38 ++++--- cli/main.ts | 2 + cli/pull.ts | 7 +- cli/schedule.ts | 135 ++++++++++++++++++++++++ cli/sync.ts | 17 ++- cli/types.ts | 9 +- 10 files changed, 225 insertions(+), 21 deletions(-) create mode 100644 cli/schedule.ts diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 2ac87dd037..6c30388327 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -1986,6 +1986,11 @@ async fn run_preview_job( Json(preview): Json, ) -> error::Result<(StatusCode, String)> { check_scopes(&authed, || format!("runscript"))?; + if authed.is_operator { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, preview.args.unwrap_or_default()); @@ -2075,6 +2080,11 @@ async fn run_preview_flow_job( Json(raw_flow): Json, ) -> error::Result<(StatusCode, String)> { check_scopes(&authed, || format!("runflow"))?; + if authed.is_operator { + return Err(error::Error::NotAuthorized( + "Operators cannot run preview jobs for security reasons".to_string(), + )); + } let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); let scheduled_for = run_query.get_scheduled_for(tx.transaction_mut()).await?; let args = run_query.add_include_headers(headers, raw_flow.args.unwrap_or_default()); diff --git a/backend/windmill-api/src/schedule.rs b/backend/windmill-api/src/schedule.rs index 751463099a..38d18170cd 100644 --- a/backend/windmill-api/src/schedule.rs +++ b/backend/windmill-api/src/schedule.rs @@ -169,9 +169,11 @@ async fn edit_schedule( path, w_id ) - .fetch_one(&mut tx) + .fetch_optional(&mut tx) .await?; + let is_flow = not_found_if_none(is_flow, "Schedule", &path)?; + clear_schedule(tx.transaction_mut(), path, is_flow).await?; let schedule = sqlx::query_as!( Schedule, diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 961232f383..5e374e5a80 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -34,6 +34,7 @@ use magic_crypt::MagicCryptTrait; #[cfg(feature = "enterprise")] use stripe::CustomerId; use windmill_audit::{audit_log, ActionKind}; +use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::{ error::{to_anyhow, Error, JsonResult, Result}, @@ -1368,6 +1369,7 @@ struct ArchiveQueryParams { skip_secrets: Option, skip_variables: Option, skip_resources: Option, + include_schedules: Option, } #[inline] @@ -1395,6 +1397,7 @@ where "archived", "has_draft", "draft_only", + "error" ] { if obj.contains_key(key) { obj.remove(key); @@ -1422,6 +1425,7 @@ async fn tarball_workspace( skip_resources, skip_secrets, skip_variables, + include_schedules, }): Query, ) -> Result<([(headers::HeaderName, String); 2], impl IntoResponse)> { require_admin(authed.is_admin, &authed.username)?; @@ -1605,6 +1609,25 @@ async fn tarball_workspace( .await?; } } + + if include_schedules.unwrap_or(false) { + let schedules = sqlx::query_as!( + Schedule, + "SELECT * FROM schedule + WHERE workspace_id = $1", + &w_id + ) + .fetch_all(&db) + .await?; + + for schedule in schedules { + let app_str = &to_string_without_metadata(&schedule, false).unwrap(); + archive + .write_to_archive(&app_str, &format!("{}.schedule.json", schedule.path)) + .await?; + } + } + archive.finish().await?; let file = tokio::fs::File::open(file_path).await?; diff --git a/backend/windmill-common/src/schedule.rs b/backend/windmill-common/src/schedule.rs index bf2c5c267d..6b4b3b3de4 100644 --- a/backend/windmill-common/src/schedule.rs +++ b/backend/windmill-common/src/schedule.rs @@ -24,6 +24,7 @@ pub struct Schedule { pub args: Option, pub extra_perms: serde_json::Value, pub email: String, + #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, pub on_failure: Option, } diff --git a/cli/folder.ts b/cli/folder.ts index c2a898334d..39d7afa592 100644 --- a/cli/folder.ts +++ b/cli/folder.ts @@ -64,22 +64,32 @@ export async function pushFolder( return; } log.debug(`Folder ${name} is not up-to-date, updating...`); - await FolderService.updateFolder({ - workspace: workspace, - name: name, - requestBody: { - ...localFolder, - }, - }); + try { + await FolderService.updateFolder({ + workspace: workspace, + name: name, + requestBody: { + ...localFolder, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } } else { console.log(colors.bold.yellow("Creating new folder: " + name)); - await FolderService.createFolder({ - workspace: workspace, - requestBody: { - name: name, - ...localFolder, - }, - }); + try { + await FolderService.createFolder({ + workspace: workspace, + requestBody: { + name: name, + ...localFolder, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } } } diff --git a/cli/main.ts b/cli/main.ts index 6b8ab39711..b71a0581b4 100644 --- a/cli/main.ts +++ b/cli/main.ts @@ -16,6 +16,7 @@ import push from "./push.ts"; import pull from "./pull.ts"; import hub from "./hub.ts"; import folder from "./folder.ts"; +import schedule from "./schedule.ts"; import sync from "./sync.ts"; import dev from "./dev.ts"; import { tryResolveVersion } from "./context.ts"; @@ -61,6 +62,7 @@ let command: any = new Command() .command("variable", variable) .command("hub", hub) .command("folder", folder) + .command("schedule", schedule) .command("dev", dev) .command("sync", sync) .command("version", "Show version information") diff --git a/cli/pull.ts b/cli/pull.ts index 1230bdd069..ed40b487b4 100644 --- a/cli/pull.ts +++ b/cli/pull.ts @@ -8,7 +8,8 @@ export async function downloadZip( plainSecrets: boolean | undefined, skipVariables?: boolean, skipResources?: boolean, - skipSecrets?: boolean + skipSecrets?: boolean, + includeSchedules?: boolean ): Promise { const requestHeaders: HeadersInit = new Headers(); requestHeaders.set("Authorization", "Bearer " + workspace.token); @@ -22,7 +23,9 @@ export async function downloadZip( plainSecrets ?? false }&skip_variables=${skipVariables ?? false}&skip_resources=${ skipResources ?? false - }&skip_secrets=${skipSecrets ?? false}`, + }&skip_secrets=${skipSecrets ?? false}&include_schedules=${ + includeSchedules ?? false + }`, { headers: requestHeaders, method: "GET", diff --git a/cli/schedule.ts b/cli/schedule.ts new file mode 100644 index 0000000000..ba76ac70e9 --- /dev/null +++ b/cli/schedule.ts @@ -0,0 +1,135 @@ +// deno-lint-ignore-file no-explicit-any +import { + colors, + Command, + Schedule, + ScheduleService, + log, + Table, +} from "./deps.ts"; +import { requireLogin, resolveWorkspace, validatePath } from "./context.ts"; +import { + GlobalOptions, + isSuperset, + parseFromFile, + removeType, +} from "./types.ts"; + +export interface ScheduleFile { + schedule: string; + on_failure: string; + script_path: string; + args: any; + timezone: string; + is_flow: boolean; +} + +async function list(opts: GlobalOptions) { + const workspace = await resolveWorkspace(opts); + await requireLogin(opts); + + const schedules = await ScheduleService.listSchedules({ + workspace: workspace.workspaceId, + }); + + new Table() + .header(["Path", "Schedule"]) + .padding(2) + .border(true) + .body(schedules.map((x) => [x.path, x.schedule])) + .render(); +} + +export async function pushSchedule( + workspace: string, + path: string, + schedule: Schedule | ScheduleFile | undefined, + localSchedule: ScheduleFile, + raw: boolean +): Promise { + path = removeType(path, "schedule"); + + log.debug(`Processing local schedule ${path}`); + + if (raw) { + // deleting old app if it exists in raw mode + try { + schedule = await ScheduleService.getSchedule({ workspace, path }); + log.debug(`Schedule ${path} exists on remote`); + } catch { + log.debug(`Schedule ${path} does not exist on remote`); + //ignore + } + } + + if (schedule) { + if (isSuperset(localSchedule, schedule)) { + log.debug(`Schedule ${path} is up to date`); + return; + } + log.debug(`Schedule ${path} is not up-to-date, updating...`); + try { + await ScheduleService.updateSchedule({ + workspace: workspace, + path, + requestBody: { + ...localSchedule, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } + } else { + console.log(colors.bold.yellow("Creating new schedule: " + path)); + try { + await ScheduleService.createSchedule({ + workspace: workspace, + requestBody: { + path: path, + ...localSchedule, + }, + }); + } catch (e) { + console.error(e.body); + throw e; + } + } +} + +async function push(opts: GlobalOptions, filePath: string, remotePath: string) { + const workspace = await resolveWorkspace(opts); + await requireLogin(opts); + + if (!validatePath(remotePath)) { + return; + } + + const fstat = await Deno.stat(filePath); + if (!fstat.isFile) { + throw new Error("file path must refer to a file."); + } + + console.log(colors.bold.yellow("Pushing schedule...")); + + await pushSchedule( + workspace.workspaceId, + remotePath, + undefined, + parseFromFile(filePath), + false + ); + console.log(colors.bold.underline.green("Schedule pushed")); +} + +const command = new Command() + .description("schedule related commands") + .action(list as any) + .command( + "push", + "push a local schedule spec. This overrides any remote versions." + ) + .arguments(" ") + .action(push as any); + +export default command; diff --git a/cli/sync.ts b/cli/sync.ts index 850be983cc..cd7a186792 100644 --- a/cli/sync.ts +++ b/cli/sync.ts @@ -19,6 +19,7 @@ import { log, yamlStringify, yamlParse, + ScheduleService, } from "./deps.ts"; import { getTypeStrFromPath, @@ -398,6 +399,7 @@ async function pull( skipVariables?: boolean; skipResources?: boolean; skipSecrets?: boolean; + includeSchedules?: boolean; } ) { if (!opts.raw) { @@ -418,7 +420,8 @@ async function pull( opts.plainSecrets, opts.skipVariables, opts.skipResources, - opts.skipSecrets + opts.skipSecrets, + opts.includeSchedules ))!, !opts.json ); @@ -606,6 +609,7 @@ async function push( skipVariables?: boolean; skipResources?: boolean; skipSecrets?: boolean; + includeSchedules?: boolean; } ) { if (!opts.raw) { @@ -635,7 +639,8 @@ async function push( opts.plainSecrets, opts.skipVariables, opts.skipResources, - opts.skipSecrets + opts.skipSecrets, + opts.includeSchedules ))!, !opts.json ); @@ -788,6 +793,12 @@ async function push( path: removeSuffix(change.path, ".app.json"), }); break; + case "schedule": + await ScheduleService.deleteSchedule({ + workspace: workspaceId, + path: removeSuffix(change.path, ".schedule.json"), + }); + break; case "variable": await VariableService.deleteVariable({ workspace: workspaceId, @@ -828,6 +839,7 @@ const command = new Command() .option("--skip-variables", "Skip syncing variables (including secrets)") .option("--skip-secrets", "Skip syncing only secrets variables") .option("--skip-resources", "Skip syncing resources") + .option("--include-schedules", "Include syncing schedules") // deno-lint-ignore no-explicit-any .action(pull as any) .command("push") @@ -846,6 +858,7 @@ const command = new Command() .option("--skip-variables", "Skip syncing variables (including secrets)") .option("--skip-secrets", "Skip syncing only secrets variables") .option("--skip-resources", "Skip syncing resources") + .option("--include-schedules", "Include syncing schedules") // deno-lint-ignore no-explicit-any .action(push as any); diff --git a/cli/types.ts b/cli/types.ts index ae4acbf42e..344294507a 100644 --- a/cli/types.ts +++ b/cli/types.ts @@ -11,6 +11,7 @@ import * as Diff from "npm:diff"; import { yamlOptions } from "./sync.ts"; import { showDiffs } from "./main.ts"; import { deepEqual } from "./utils.ts"; +import { pushSchedule } from "./schedule.ts"; export interface DifferenceCreate { type: "CREATE"; @@ -115,6 +116,8 @@ export function pushObj( pushResource(workspace, p, befObj, newObj, checkForCreate); } else if (typeEnding === "resource-type") { pushResourceType(workspace, p, befObj, newObj, checkForCreate); + } else if (typeEnding === "schedule") { + pushSchedule(workspace, p, befObj, newObj, checkForCreate); } else { throw new Error("infer type unreachable"); } @@ -145,7 +148,8 @@ export function getTypeStrFromPath( | "resource" | "resource-type" | "folder" - | "app" { + | "app" + | "schedule" { if (p.includes(".flow/")) { return "flow"; } @@ -170,7 +174,8 @@ export function getTypeStrFromPath( typeEnding === "variable" || typeEnding === "resource" || typeEnding === "resource-type" || - typeEnding === "app" + typeEnding === "app" || + typeEnding === "schedule" ) { return typeEnding; } else {