starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-24 08:01:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add schedule to syncable resources

Browse Source
This commit is contained in:
Ruben Fiszel
2023-07-15 19:28:40 +02:00
parent b71362fc7f
commit 1956c43705
11 changed files with 249 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/jobs.rs
+10
View File
@@ -1986,6 +1986,11 @@ async fn run_preview_job(
Json(preview): Json<Preview>,
) -> 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<PreviewFlow>,
) -> 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());
backend/windmill-api/src/schedule.rs
+3 -1
View File
@@ -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,
backend/windmill-api/src/users.rs
+24 -9
View File
@@ -136,17 +136,24 @@ impl AuthCache {
(Some(owner), email, super_admin, _) if w_id.is_some() => {
if let Some((prefix, name)) = owner.split_once('/') {
if prefix == "u" {
let is_admin = super_admin
|| sqlx::query_scalar!(
"SELECT is_admin FROM usr where username = $1 AND \
let (is_admin, is_operator) = if super_admin {
(true, false)
} else {
let r = sqlx::query!(
"SELECT is_admin, operator FROM usr where username = $1 AND \
workspace_id = $2 AND disabled = false",
name,
&w_id.as_ref().unwrap()
)
.fetch_one(&self.db)
.await
.ok()
.unwrap_or(false);
.ok();
if let Some(r) = r {
(r.is_admin, r.operator)
} else {
(false, true)
}
};
let w_id = &w_id.unwrap();
let groups = get_groups_for_user(w_id, &name, &self.db)
@@ -165,6 +172,7 @@ impl AuthCache {
.unwrap_or_else(|| "missing@email.xyz".to_string()),
username: name.to_string(),
is_admin,
is_operator,
groups,
folders,
scopes: None,
@@ -186,6 +194,7 @@ impl AuthCache {
username: format!("group-{name}"),
is_admin: false,
groups,
is_operator: false,
folders,
scopes: None,
})
@@ -198,6 +207,7 @@ impl AuthCache {
.unwrap_or_else(|| "missing@email.xyz".to_string()),
username: owner,
is_admin: super_admin,
is_operator: true,
groups,
folders,
scopes: None,
@@ -206,18 +216,18 @@ impl AuthCache {
}
(_, Some(email), super_admin, scopes) => {
if w_id.is_some() {
let row_o = sqlx::query_as::<_, (String, bool)>(
"SELECT username, is_admin FROM usr where email = $1 AND \
let row_o = sqlx::query_as::<_, (String, bool, bool)>(
"SELECT username, is_admin, operator FROM usr where email = $1 AND \
workspace_id = $2 AND disabled = false",
)
.bind(&email)
.bind(&w_id.as_ref().unwrap())
.fetch_optional(&self.db)
.await
.unwrap_or(Some(("error".to_string(), false)));
.unwrap_or(Some(("error".to_string(), false, false)));
match row_o {
Some((username, is_admin)) => {
Some((username, is_admin, is_operator)) => {
let groups = get_groups_for_user(
&w_id.as_ref().unwrap(),
&username,
@@ -240,6 +250,7 @@ impl AuthCache {
email,
username,
is_admin: is_admin || super_admin,
is_operator,
groups,
folders,
scopes,
@@ -249,6 +260,7 @@ impl AuthCache {
email: email.clone(),
username: email,
is_admin: super_admin,
is_operator: false,
groups: vec![],
folders: vec![],
scopes,
@@ -260,6 +272,7 @@ impl AuthCache {
email: email.to_string(),
username: email,
is_admin: super_admin,
is_operator: true,
groups: Vec::new(),
folders: Vec::new(),
scopes,
@@ -285,6 +298,7 @@ impl AuthCache {
email: SUPERADMIN_SECRET_EMAIL.to_string(),
username: "superadmin_secret".to_string(),
is_admin: true,
is_operator: false,
groups: Vec::new(),
folders: Vec::new(),
scopes: None,
@@ -366,6 +380,7 @@ pub struct Authed {
pub email: String,
pub username: String,
pub is_admin: bool,
pub is_operator: bool,
pub groups: Vec<String>,
// (folder name, can write, is owner)
pub folders: Vec<(String, bool, bool)>,
backend/windmill-api/src/workspaces.rs
+23
View File
@@ -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<bool>,
skip_variables: Option<bool>,
skip_resources: Option<bool>,
include_schedules: Option<bool>,
}
#[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<ArchiveQueryParams>,
) -> 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?;
backend/windmill-common/src/schedule.rs
+1
View File
@@ -24,6 +24,7 @@ pub struct Schedule {
pub args: Option<serde_json::Value>,
pub extra_perms: serde_json::Value,
pub email: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
pub on_failure: Option<String>,
}
cli/folder.ts
+24 -14
View File
@@ -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;
}
}
}
cli/main.ts
+2
View File
@@ -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")
cli/pull.ts
+5 -2
View File
@@ -8,7 +8,8 @@ export async function downloadZip(
plainSecrets: boolean | undefined,
skipVariables?: boolean,
skipResources?: boolean,
skipSecrets?: boolean
skipSecrets?: boolean,
includeSchedules?: boolean
): Promise<JSZip | undefined> {
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",
cli/schedule.ts
+135
View File
@@ -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<void> {
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("<file_path:string> <remote_path:string>")
.action(push as any);
export default command;
cli/sync.ts
+15 -2
View File
@@ -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);
cli/types.ts
+7 -2
View File
@@ -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 {