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

update

Browse Source
This commit is contained in:
dieriba
2025-10-23 16:19:51 +02:00
parent b6ae5d4f57
commit 4934d113d4
9 changed files with 194 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/openapi.yaml
+31 -33
View File
@@ -770,10 +770,10 @@ paths:
schema:
type: boolean
/workspaces/migrate:
/workspaces/migrate/tables:
post:
summary: migrate workspace data from source to target
operationId: migrateWorkspace
operationId: migrateWorkspaceTables
tags:
- workspace
requestBody:
@@ -791,7 +791,29 @@ paths:
schema:
type: string
/workspaces/migration/status:
/workspaces/migrate/jobs:
post:
summary: migrate workspace jobs from source to target
operationId: migrateWorkspaceJobs
tags:
- workspace
requestBody:
description: migration jobs
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/MigrateJobsRequest"
responses:
"200":
description: migration completed
content:
application/json:
schema:
$ref: "#/components/schemas/MigrateJobsStatus"
/workspaces/migrate/status:
get:
summary: get migration status
operationId: getMigrationStatus
@@ -810,7 +832,7 @@ paths:
content:
application/json:
schema:
$ref: "#/components/schemas/MigrateJobResponse"
$ref: "#/components/schemas/MigrationStatus"
/settings/get_ducklake_instance_catalog_db_status:
post:
@@ -17667,21 +17689,12 @@ components:
target_workspace_name:
type: string
description: target workspace name
migration_type:
type: string
enum: [all, metadata, jobs]
description: type of migration to perform
disable_workspace:
type: boolean
default: true
description: whether to disable source workspace after migration
required:
- source_workspace_id
- target_workspace_name
- target_workspace_id
- migration_type
MigrateJobsBatchRequest:
MigrateJobsRequest:
type: object
properties:
source_workspace_id:
@@ -17699,38 +17712,23 @@ components:
- source_workspace_id
- target_workspace_id
MigrateJobsBatchResponse:
MigrateJobsStatus:
type: object
properties:
migrated_count:
type: integer
format: int64
description: number of jobs migrated in this batch
remaining_jobs:
type: integer
format: int64
description: number of jobs still remaining
total_jobs:
type: integer
format: int64
description: total number of jobs
migration_progress:
type: number
format: float
description: migration progress percentage
required:
- migrated_count
- remaining_jobs
- total_jobs
- migration_progress
MigrateJobResponse:
MigrationStatus:
type: object
properties:
processed_jobs:
type: integer
format: int64
description: number of jobs processed (remaining in source workspace)
description: number of jobs processed in source workspace
required:
- processed_jobs
backend/windmill-api/src/workspaces.rs
+20 -5
View File
@@ -67,6 +67,8 @@ use crate::teams_oss::{
workspaces_list_available_teams_channels, workspaces_list_available_teams_ids,
};
use crate::workspaces_extra::{get_migration_status, migrate_jobs, migrate_workspace};
lazy_static::lazy_static! {
static ref WORKSPACE_KEY_REGEXP: Regex = Regex::new("^[a-zA-Z0-9]{64}$").unwrap();
}
@@ -165,6 +167,16 @@ pub fn workspaced_service() -> Router {
#[cfg(not(feature = "stripe"))]
router
}
pub fn migrate_service() -> Router {
Router::new().nest(
"/migrate",
Router::new()
.route("/tables", post(migrate_workspace))
.route("/jobs", post(migrate_jobs))
.route("/status", get(get_migration_status)),
)
}
pub fn global_service() -> Router {
Router::new()
.route("/list_as_superadmin", get(list_workspaces_as_super_admin))
@@ -184,11 +196,7 @@ pub fn global_service() -> Router {
"/create_workspace_require_superadmin",
get(create_workspace_require_superadmin),
)
.route("/migrate", post(crate::workspaces_extra::migrate_workspace))
.route(
"/migration/status",
get(crate::workspaces_extra::get_migration_status),
)
.merge(migrate_service())
}
#[derive(FromRow, Serialize)]
@@ -2186,6 +2194,13 @@ lazy_static::lazy_static! {
}
};
pub static ref MIGRATE_JOBS_WORKSPACE_REQUIRE_SUPERADMIN: bool = {
match std::env::var("MIGRATE_JOBS_WORKSPACE_REQUIRE_SUPERADMIN") {
Ok(val) => val == "true",
Err(_) => true,
}
};
}
async fn create_workspace_require_superadmin() -> String {
backend/windmill-api/src/workspaces_extra.rs
+90 -125
View File
@@ -1,6 +1,6 @@
use crate::db::ApiAuthed;
use crate::workspaces::{check_w_id_conflict, CREATE_WORKSPACE_REQUIRE_SUPERADMIN, WM_FORK_PREFIX};
use crate::workspaces::{check_w_id_conflict, CREATE_WORKSPACE_REQUIRE_SUPERADMIN, MIGRATE_JOBS_WORKSPACE_REQUIRE_SUPERADMIN, WM_FORK_PREFIX};
use crate::{db::DB, utils::require_super_admin};
use axum::extract::Query;
@@ -13,6 +13,7 @@ use sqlx::{Postgres, Transaction};
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::db::UserDB;
use windmill_common::error::JsonResult;
use windmill_common::worker::CLOUD_HOSTED;
@@ -25,17 +26,22 @@ use windmill_common::{
use serde::{Deserialize, Serialize};
#[derive(Deserialize)]
pub struct MigrateWorkspaceRequest {
pub struct MigrateJobRequest {
source_workspace_id: String,
target_workspace_name: String,
target_workspace_id: String,
migration_type: MigrationType,
#[serde(default = "default_disable_workspace")]
disable_workspace: bool,
#[serde(default = "default_batch_size")]
batch_size: i64,
}
fn default_disable_workspace() -> bool {
true
#[derive(Deserialize)]
pub struct MigrateWorkspaceRequest {
source_workspace_id: String,
target_workspace_id: String,
target_workspace_name: String,
}
fn default_batch_size() -> i64 {
DEFAULT_BATCH_SIZE
}
#[derive(Default, Deserialize, Serialize, Debug, Clone, Copy, PartialEq)]
@@ -234,61 +240,52 @@ async fn is_workspace_owner(
Ok(owner.map(|o| o == authed.email).unwrap_or(false))
}
pub async fn migrate_workspace(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Json(req): Json<MigrateWorkspaceRequest>,
) -> Result<String> {
if *CLOUD_HOSTED && !is_super_admin_email(&db, &authed.email).await? {
#[inline]
pub async fn is_allowed_to_migrate(db: &DB, authed: &ApiAuthed, predicate: bool) -> Result<()> {
if *CLOUD_HOSTED && !is_super_admin_email(db, &authed.email).await? {
return Err(Error::BadRequest(
"This feature is not available on the cloud".to_string(),
));
}
if *CREATE_WORKSPACE_REQUIRE_SUPERADMIN {
require_super_admin(&db, &authed.email).await?;
if predicate {
require_super_admin(db, &authed.email).await?;
} else {
require_admin(authed.is_admin, &authed.username)?;
}
require_super_admin(&db, &authed.email).await?;
Ok(())
}
pub async fn migrate_workspace(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Json(req): Json<MigrateWorkspaceRequest>,
) -> Result<String> {
is_allowed_to_migrate(&db, &authed, *CREATE_WORKSPACE_REQUIRE_SUPERADMIN).await?;
let mut tx = db.begin().await?;
if req.migration_type != MigrationType::Jobs {
check_w_id_conflict(&mut tx, &req.target_workspace_id).await?;
check_w_id_conflict(&mut tx, &req.target_workspace_id).await?;
sqlx::query!(
"INSERT INTO
workspace SELECT $1, $2, owner, deleted, premium FROM workspace WHERE id = $3",
&req.target_workspace_id,
&req.target_workspace_name,
&req.source_workspace_id
)
.execute(&mut *tx)
.await?;
}
sqlx::query!(
r#"
INSERT INTO
workspace
SELECT $1, $2, owner, deleted, premium
FROM
workspace
WHERE
id = $3
"#,
&req.target_workspace_id,
&req.target_workspace_name,
&req.source_workspace_id
)
.execute(&mut *tx)
.await?;
match req.migration_type {
MigrationType::Metadata | MigrationType::All => {
migrate_metadata_tables(&mut tx, &req.source_workspace_id, &req.target_workspace_id)
.await?;
}
_ => {}
}
match req.migration_type {
MigrationType::Jobs | MigrationType::All => {
let result = migrate_jobs_batch(
&mut tx,
&req.source_workspace_id,
&req.target_workspace_id,
DEFAULT_BATCH_SIZE,
)
.await?;
}
_ => {}
}
migrate_metadata_tables(&mut tx, &req.source_workspace_id, &req.target_workspace_id).await?;
audit_log(
&mut *tx,
@@ -301,14 +298,6 @@ pub async fn migrate_workspace(
[
("source", req.source_workspace_id.as_str()),
("target", req.target_workspace_id.as_str()),
(
"type",
match req.migration_type {
MigrationType::All => "all",
MigrationType::Metadata => "metadata",
MigrationType::Jobs => "jobs",
},
),
]
.into(),
),
@@ -318,14 +307,8 @@ pub async fn migrate_workspace(
tx.commit().await?;
Ok(format!(
"Migrated {} from {} to {}",
match req.migration_type {
MigrationType::All => "all data",
MigrationType::Metadata => "metadata",
MigrationType::Jobs => "jobs",
},
&req.source_workspace_id,
&req.target_workspace_id
"Migrated from {} to {}",
&req.source_workspace_id, &req.target_workspace_id
))
}
@@ -334,7 +317,7 @@ async fn migrate_metadata_tables(
source: &str,
target: &str,
) -> Result<()> {
let simple_tables = vec![
let simple_tables = [
"account",
"app",
"audit",
@@ -366,6 +349,10 @@ async fn migrate_metadata_tables(
"workspace_invite",
"workspace_key",
"workspace_settings",
"job_logs",
"job_stats",
"v2_job_queue",
"v2_job",
];
for table in simple_tables {
@@ -425,54 +412,6 @@ async fn migrate_metadata_tables(
Ok(())
}
async fn migrate_job_tables(
tx: &mut Transaction<'_, Postgres>,
source: &str,
target: &str,
) -> Result<()> {
sqlx::query!(
"UPDATE job_logs SET workspace_id = $1 WHERE workspace_id = $2",
target,
source
)
.execute(&mut **tx)
.await?;
sqlx::query!(
"UPDATE job_stats SET workspace_id = $1 WHERE workspace_id = $2",
target,
source
)
.execute(&mut **tx)
.await?;
sqlx::query!(
"UPDATE v2_job_queue SET workspace_id = $1 WHERE workspace_id = $2",
target,
source
)
.execute(&mut **tx)
.await?;
sqlx::query!(
"UPDATE v2_job SET workspace_id = $1 WHERE workspace_id = $2",
target,
source
)
.execute(&mut **tx)
.await?;
sqlx::query!(
"UPDATE v2_job_completed SET workspace_id = $1 WHERE workspace_id = $2",
target,
source
)
.execute(&mut **tx)
.await?;
Ok(())
}
const DEFAULT_BATCH_SIZE: i64 = 10000;
#[derive(Serialize)]
@@ -507,12 +446,36 @@ pub async fn get_migration_status(
Ok(Json(MigrationStatus { processed_jobs: source_jobs }))
}
async fn migrate_jobs_batch(
tx: &mut Transaction<'_, Postgres>,
source_workspace_id: &str,
target_workspace_id: &str,
batch_size: i64,
) -> Result<MigrateJobsBatchResponse> {
pub async fn migrate_jobs(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Json(req): Json<MigrateJobRequest>,
) -> JsonResult<MigrateJobsBatchResponse> {
is_allowed_to_migrate(&db, &authed, *MIGRATE_JOBS_WORKSPACE_REQUIRE_SUPERADMIN).await?;
let mut tx = user_db.begin(&authed).await?;
let _ = sqlx::query_scalar!(
r#"
SELECT
1
FROM
workspace
WHERE
id = $1
"#,
&req.target_workspace_id
)
.fetch_one(&mut *tx)
.await
.map_err(|_| {
Error::NotFound(format!(
"Workspace: {} does not exists",
req.target_workspace_id
))
})?;
let migrated_count = sqlx::query_scalar!(
"WITH batch AS (
SELECT id FROM v2_job_completed
@@ -523,13 +486,15 @@ async fn migrate_jobs_batch(
SET workspace_id = $3
WHERE id IN (SELECT id FROM batch)
RETURNING 1",
source_workspace_id,
batch_size,
target_workspace_id
req.source_workspace_id,
req.batch_size,
req.target_workspace_id
)
.fetch_all(&mut **tx)
.fetch_all(&mut *tx)
.await?
.len() as i64;
Ok(MigrateJobsBatchResponse { migrated_count })
tx.commit().await?;
Ok(Json(MigrateJobsBatchResponse { migrated_count }))
}
cli/src/commands/workspace/migrate.ts
+42 -132
View File
@@ -4,22 +4,10 @@ import { GlobalOptions } from "../../types.ts";
import { requireLogin } from "../../core/auth.ts";
import { setClient } from "../../../deps.ts";
import * as wmill from "../../../gen/services.gen.ts";
import {
getActiveWorkspace,
setActiveWorkspace,
removeWorkspace,
addWorkspace,
} from "./workspace.ts";
import { getWorkspaceConfigFilePath } from "../../../windmill-utils-internal/src/config/config.ts";
import { getActiveWorkspace } from "./workspace.ts";
async function migrate(
opts: GlobalOptions & {
all?: boolean;
metadataOnly?: boolean;
jobsOnly?: boolean;
noDisableSource?: boolean;
targetName?: string;
noSwitchWorkspace?: boolean;
token?: string;
remote?: string;
sourceWorkspaceId?: string;
@@ -29,7 +17,6 @@ async function migrate(
let token: string;
let remote: string;
let sourceWorkspaceId: string;
let isCliMode: boolean;
if (opts.token && opts.remote && opts.sourceWorkspaceId) {
token = opts.token;
@@ -37,7 +24,6 @@ async function migrate(
? opts.remote.substring(0, opts.remote.length - 1)
: opts.remote;
sourceWorkspaceId = opts.sourceWorkspaceId;
isCliMode = false;
log.info(
colors.blue("Running in worker job mode with provided credentials")
);
@@ -51,7 +37,6 @@ async function migrate(
);
}
isCliMode = true;
token = workspace.token;
remote = workspace.remote.endsWith("/")
? workspace.remote.substring(0, workspace.remote.length - 1)
@@ -63,122 +48,65 @@ async function migrate(
setClient(token, remote);
let migrationType: "all" | "metadata" | "jobs" = "all";
if (opts.metadataOnly) {
migrationType = "metadata";
} else if (opts.jobsOnly) {
migrationType = "jobs";
}
const disableSource = !opts.noDisableSource;
const targetName = opts.targetName || targetWorkspaceId;
const shouldSwitchWorkspace = !opts.noSwitchWorkspace;
log.info(colors.blue("Starting workspace migration:"));
log.info(` Source: ${colors.bold(sourceWorkspaceId)}`);
log.info(` Target: ${colors.bold(targetWorkspaceId)} (${targetName})`);
log.info(` Type: ${colors.bold(migrationType)}`);
log.info(` Disable source: ${colors.bold(String(disableSource))}`);
if (shouldSwitchWorkspace) {
log.info(` Switch to target workspace: ${colors.bold("yes")}`);
}
log.info("");
try {
if (isCliMode && migrationType === "all") {
log.info(colors.blue("=".repeat(60)));
log.info(colors.blue("STEP 1: Migrating Metadata"));
log.info(colors.blue("=".repeat(60)));
log.info("");
log.info(colors.blue("=".repeat(60)));
log.info(colors.blue("Migrating jobs"));
log.info(colors.blue("=".repeat(60)));
log.info("");
const metadataResult = await wmill.migrateWorkspace({
const initialStatus = await wmill.getMigrationStatus({
sourceWorkspace: sourceWorkspaceId,
});
const totalJobs = initialStatus.processed_jobs || 0;
log.info(`Total jobs to migrate: ${colors.bold(totalJobs.toString())}`);
if (totalJobs === 0) {
log.info(colors.yellow("No jobs to migrate"));
return;
}
let totalMigrated = 0;
const batchSize = 10000;
while (true) {
log.info(`Processing batch (size: ${batchSize})...`);
const batchResult = await wmill.migrateWorkspaceJobs({
requestBody: {
source_workspace_id: sourceWorkspaceId,
target_workspace_id: targetWorkspaceId,
target_workspace_name: targetName,
migration_type: "metadata",
disable_workspace: disableSource,
batch_size: batchSize,
},
});
log.info(colors.green(`✅ ${metadataResult}`));
log.info("");
const migratedInBatch = batchResult.migrated_count || 0;
totalMigrated += migratedInBatch;
log.info(colors.blue("=".repeat(60)));
log.info(colors.blue("STEP 2: Migrating Job History (v2_job_completed)"));
log.info(colors.blue("=".repeat(60)));
log.info("");
const progress = Math.round((totalMigrated / totalJobs) * 100);
log.info(`${colors.green(migratedInBatch.toString())} jobs migrated`);
log.info(
`Progress: ${colors.cyan(
`${totalMigrated}/${totalJobs}`
)} (${colors.yellow(`${progress}%`)})`
);
const jobsResult = await wmill.migrateWorkspace({
requestBody: {
source_workspace_id: sourceWorkspaceId,
target_workspace_id: targetWorkspaceId,
target_workspace_name: targetName,
migration_type: "jobs",
disable_workspace: false,
},
});
log.info(colors.green(`✅ ${jobsResult}`));
log.info("");
log.info(colors.green("=".repeat(60)));
log.info(colors.green("✅ Complete migration finished successfully!"));
log.info(colors.green("=".repeat(60)));
} else {
const result = await wmill.migrateWorkspace({
requestBody: {
source_workspace_id: sourceWorkspaceId,
target_workspace_id: targetWorkspaceId,
target_workspace_name: targetName,
migration_type: migrationType,
disable_workspace: disableSource,
},
});
log.info(colors.green(`✅ ${result}`));
if (migrationType === "metadata") {
log.info("");
log.info(
colors.yellow(
"⚠️ Metadata migration complete. Run with --jobs-only to migrate job history."
)
);
if (migratedInBatch < batchSize) {
break;
}
}
if (isCliMode && shouldSwitchWorkspace && migrationType !== "jobs") {
const workspace = await getActiveWorkspace(opts);
if (workspace) {
await removeWorkspace(workspace.name, true, opts);
const jobsResult = `Successfully migrated ${totalMigrated} jobs`;
log.info("");
log.info(colors.blue("Switching to target workspace..."));
workspace.name = targetName;
workspace.workspaceId = targetWorkspaceId;
const filePath = await getWorkspaceConfigFilePath(opts.configDir);
const file = await Deno.open(filePath, {
append: true,
write: true,
read: true,
create: true,
});
await file.write(
new TextEncoder().encode(JSON.stringify(workspace) + "\n")
);
await setActiveWorkspace(targetName, opts.configDir);
log.info(
colors.green(`✅ Switched active workspace to ${targetWorkspaceId}`)
);
log.info(
colors.green(
`✅ Removed old workspace configuration for ${sourceWorkspaceId}`
)
);
}
}
log.info(colors.green(`✅ ${jobsResult}`));
log.info("");
log.info(colors.green("=".repeat(60)));
log.info(colors.green("✅ Complete migration finished successfully!"));
log.info(colors.green("=".repeat(60)));
} catch (error) {
log.error(colors.red(`❌ Migration failed: ${error}`));
throw error;
@@ -189,24 +117,6 @@ const command = new Command()
.name("migrate")
.description("Migrate workspace data from source to target workspace")
.arguments("<target_workspace_id:string>")
.option("--all", "Migrate all tables (default)")
.option("--metadata-only", "Migrate all tables except v2_job_completed")
.option(
"--jobs-only",
"Migrate only v2_job_completed table (workspace must already exist)"
)
.option(
"--no-disable-source",
"Do not disable source workspace after migration"
)
.option(
"--target-name <name:string>",
"Name for the target workspace (defaults to target workspace ID)"
)
.option(
"--no-switch-workspace",
"Do not switch active workspace to target after migration (by default, switches and removes old workspace)"
)
.option("--token <token:string>", "API token for worker job mode")
.option("--remote <url:string>", "Remote URL for worker job mode")
.option(
frontend/src/lib/components/settings/ChangeWorkspaceId.svelte
+7 -12
View File
@@ -8,8 +8,7 @@
import { Pen } from 'lucide-svelte'
import { isCloudHosted } from '$lib/cloud'
import { onDestroy } from 'svelte'
const HUB_MIGRATE_SCRIPT_PATH = 'u/admin/workspace_migrate'
import { hubPaths } from '$lib/hub'
let { open = $bindable(false) } = $props()
@@ -50,14 +49,11 @@
loading = true
oldWorkspaceId = $workspaceStore!
const jobId = await JobService.runScriptByPath({
workspace: $workspaceStore!,
path: HUB_MIGRATE_SCRIPT_PATH,
await WorkspaceService.migrateWorkspaceTables({
requestBody: {
source_workspace_id: $workspaceStore!,
source_workspace_id: oldWorkspaceId,
target_workspace_id: newId,
target_workspace_name: newName,
migration_type: 'metadata'
target_workspace_name: newName
}
})
@@ -101,12 +97,11 @@
jobMigrationJobId = await JobService.runScriptByPath({
workspace: $workspaceStore!,
path: HUB_MIGRATE_SCRIPT_PATH,
path: hubPaths.workspaceMigrator,
requestBody: {
source_workspace_id: oldWorkspaceId,
target_workspace_id: newId,
target_workspace_name: newName,
migration_type: 'jobs'
target_workspace_name: newName
}
})
@@ -159,7 +154,7 @@
function startPolling() {
if (!pollInterval) {
pollInterval = setInterval(checkJobStatus, 2000) as any
pollInterval = setInterval(checkJobStatus, 1000) as any
}
}
frontend/src/lib/hub.ts
+1
View File
@@ -95,6 +95,7 @@ type HubPaths = {
teamsSuccessHandler: string
emailErrorHandler: string
cloneRepoToS3forGitRepoViewer: string
workspaceMigrator: string
}
export const hubPaths = JSON.parse(rawHubPaths) as HubPaths
frontend/src/lib/hubPaths.json
+2 -1
View File
@@ -38,5 +38,6 @@
"discordReport": "hub/9085/discord",
"smtpReport": "hub/9086/smtp",
"cloneRepoToS3forGitRepoViewer_0": "hub/19825/clone_repo_and_upload_to_instance_storage",
"cloneRepoToS3forGitRepoViewer": "hub/19827/clone_repo_and_upload_to_instance_storage"
"cloneRepoToS3forGitRepoViewer": "hub/19827/clone_repo_and_upload_to_instance_storage",
"workspaceMigrator": "hub/28053/workspace%20job%20migrator"
}
frontend/src/routes/(root)/(logged)/workspace_settings/migration/+page.svelte
-24
View File
@@ -1,24 +0,0 @@
<script lang="ts">
import { page } from '$app/stores'
import WorkspaceMigration from '$lib/components/settings/WorkspaceMigration.svelte'
import { workspaceStore } from '$lib/stores'
const sourceWorkspace = $derived($page.url.searchParams.get('source') || '')
$effect(() => {
// Redirect if no source workspace provided
if (!sourceWorkspace) {
window.location.href = `/workspace_settings?workspace=${$workspaceStore}`
}
})
</script>
<div class="max-w-4xl mx-auto p-6">
{#if sourceWorkspace}
<WorkspaceMigration {sourceWorkspace} />
{:else}
<div class="flex items-center justify-center p-8">
<p class="text-secondary">Loading...</p>
</div>
{/if}
</div>
intra-uuid-e4eed185-2f29-415f-947e-b2bdc697c8a9-4888488-dtoure
Submodule
+1
Submodule intra-uuid-e4eed185-2f29-415f-947e-b2bdc697c8a9-4888488-dtoure added at 6668698e0b