diff --git a/@xen-orchestra/xapi/_watchUploadProgress.mjs b/@xen-orchestra/xapi/_watchUploadProgress.mjs new file mode 100644 index 0000000000..3c90878dc2 --- /dev/null +++ b/@xen-orchestra/xapi/_watchUploadProgress.mjs @@ -0,0 +1,34 @@ +// XCP-ng only updates the import task progress for some formats (e.g. it does +// for VHD but not for qcow2). To get a consistent progress regardless of the +// format, we count the uploaded bytes ourselves and update the task. +const IMPORT_PROGRESS_MAX_INTERVAL = 5e3 // maximum delay in ms between two progress updates +const IMPORT_PROGRESS_MIN_DELTA = 0.01 // minimal progress increase (1%) required to emit + +// Observe how much data flows through `stream` (which must expose its total +// `length`) and report the uploaded fraction (`0` → `1`) via `onProgress`, +// throttled to avoid spamming the XAPI task API. +export function watchUploadProgress(stream, length, onProgress) { + let sent = 0 + let lastDate = 0 + let lastValue = -1 + + const emit = (progress, force) => { + const now = Date.now() + if (!force && now - lastDate < IMPORT_PROGRESS_MAX_INTERVAL && progress - lastValue < IMPORT_PROGRESS_MIN_DELTA) { + return + } + lastDate = now + lastValue = progress + onProgress(progress) + } + + stream.on('data', chunk => { + sent += chunk.length + emit(length > 0 ? Math.min(sent / length, 1) : 1, false) + }) + stream.on('end', () => emit(1, true)) + + // adding a `data` listener switches the stream to flowing mode: pause it until + // the actual consumer starts reading so that no chunk is lost in the meantime + stream.pause() +} diff --git a/@xen-orchestra/xapi/_watchUploadProgress.test.mjs b/@xen-orchestra/xapi/_watchUploadProgress.test.mjs new file mode 100644 index 0000000000..a92acef678 --- /dev/null +++ b/@xen-orchestra/xapi/_watchUploadProgress.test.mjs @@ -0,0 +1,58 @@ +import { strict as assert } from 'node:assert' +import test from 'node:test' +import { Readable } from 'node:stream' + +import { watchUploadProgress } from './_watchUploadProgress.mjs' + +// drive `chunks` (an array of Buffers) through watchUploadProgress and resolve +// with the ordered list of reported progress values +async function run(chunks, length) { + const values = [] + const stream = Readable.from(chunks) + watchUploadProgress(stream, length, progress => values.push(progress)) + + // watchUploadProgress pauses the stream: resume it like the real consumer does + stream.resume() + await new Promise((resolve, reject) => { + stream.on('end', resolve) + stream.on('error', reject) + }) + return values +} + +test('reports a final progress of exactly 1 once the stream ends', async () => { + const values = await run([Buffer.alloc(50), Buffer.alloc(50)], 100) + assert.equal(values.at(-1), 1) +}) + +test('every reported value is a fraction in [0, 1] and never decreases', async () => { + const values = await run([Buffer.alloc(25), Buffer.alloc(25), Buffer.alloc(25), Buffer.alloc(25)], 100) + assert.ok(values.length > 0) + let previous = 0 + for (const value of values) { + assert.ok(value >= 0 && value <= 1, `value ${value} out of range`) + assert.ok(value >= previous, `value ${value} decreased from ${previous}`) + previous = value + } +}) + +test('a zero-length stream reports 1 without emitting NaN', async () => { + const values = await run([], 0) + assert.deepEqual(values, [1]) +}) + +test('throttles intermediate updates below the min delta but always ends at 1', async () => { + // 1-byte chunks over a 1000-byte total = 0.1% each: only the first is emitted + // (subsequent ones are below the 1% threshold and within the time window), + // then `end` forces the final 1 + const values = await run([Buffer.alloc(1), Buffer.alloc(1), Buffer.alloc(1)], 1000) + assert.deepEqual(values, [0.001, 1]) +}) + +test('caps the fraction at 1 even if more bytes than expected flow through', async () => { + const values = await run([Buffer.alloc(80), Buffer.alloc(80)], 100) + for (const value of values) { + assert.ok(value <= 1, `value ${value} exceeded 1`) + } + assert.equal(values.at(-1), 1) +}) diff --git a/@xen-orchestra/xapi/vdi.mjs b/@xen-orchestra/xapi/vdi.mjs index b9b34927be..73bb92b4d8 100644 --- a/@xen-orchestra/xapi/vdi.mjs +++ b/@xen-orchestra/xapi/vdi.mjs @@ -1,4 +1,5 @@ import CancelToken from 'promise-toolbox/CancelToken' +import ignoreErrors from 'promise-toolbox/ignoreErrors' import pCatch from 'promise-toolbox/catch' import pRetry from 'promise-toolbox/retry' import { createLogger } from '@xen-orchestra/log' @@ -9,6 +10,7 @@ import { strict as assert } from 'node:assert' import { SUPPORTED_VDI_FORMAT, VDI_FORMAT_RAW, VDI_FORMAT_QCOW2, VHD_MAX_SIZE } from './index.mjs' import { PREFERED_IMAGE_FORMAT_PROPERTY } from './pbd.mjs' +import { watchUploadProgress } from './_watchUploadProgress.mjs' const { warn, info } = createLogger('xo:xapi:vdi') @@ -288,11 +290,19 @@ class Vdi { const taskRef = await this.task_create(`Importing content into VDI ${vdi.name_label} on SR ${sr.name_label}`) const uuid = await this.getField('task', taskRef, 'uuid') await vdi.update_other_config({ 'xo:import:task': uuid, 'xo:import:length': stream.length.toString() }) + + // XCP-ng does not report progress on the import task for every format + watchUploadProgress(stream, stream.length, progress => { + ignoreErrors.call(this.call('task.set_progress', taskRef, progress)) + }) + await this.putResource(cancelToken, stream, '/import_raw_vdi/', { query: { format, vdi: ref, }, + // we keep the link to ensure the task is correctly marked as failed + // on any error task: taskRef, }) } catch (error) { diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md index bd4654b8c5..947330b462 100644 --- a/CHANGELOG.unreleased.md +++ b/CHANGELOG.unreleased.md @@ -70,7 +70,8 @@ - [Backups] write the complete disk metadata at once to improve compatibility with immutable backup repository (PR [#10104](https://github.com/vatesfr/xen-orchestra/pull/10104)) - [XO6/Host] Wrap the Network name in the PIF side panel (PR [#10155](https://github.com/vatesfr/xen-orchestra/pull/10155)) - [About/Hub] Fix "Failed to fetch latest master commit" and the microk8s version list: the default `Content-Security-Policy` now allows the browser to reach the GitHub API (PR [#10162](https://github.com/vatesfr/xen-orchestra/pull/10162)) -- +- [Qcow2 import] show a valid progress bar for qcow2 disk import from form , command line or V2V (PR [#10133](https://github.com/vatesfr/xen-orchestra/pull/10133)) + ### Packages to release > When modifying a package, add it here with its release type. @@ -95,6 +96,7 @@ - @xen-orchestra/rest-api minor - @xen-orchestra/web minor - @xen-orchestra/web-core minor +- @xen-orchestra/xapi patch - vhd-lib patch - xapi-explore-sr patch - xen-api minor diff --git a/packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs b/packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs index 9372abbb29..e378c2d845 100644 --- a/packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs +++ b/packages/xo-server/src/xo-mixins/vmware/importDisksfromDatastore.mjs @@ -5,6 +5,7 @@ import { toVhdStream } from 'vhd-lib/disk-consumer/index.mjs' import { NbdDisk } from '@vates/nbd-client/NbdDisk.mjs' import { createLogger } from '@xen-orchestra/log' import { toQcow2Stream } from '@xen-orchestra/qcow2' +import { TaskProgressHandler } from '@xen-orchestra/backups/_runners/_vmRunners/_TaskProgressHandler.mjs' const { warn } = createLogger('xo:importdiskfromdatastore') @@ -24,6 +25,8 @@ export async function importStream({ esxi, dataMap, disk, vmId, format }, consum signal?.throwIfAborted() vmdk = new ReadAhead(vmdk) + vmdk.addProgressHandler(new TaskProgressHandler()) + if (format === VDI_FORMAT_QCOW2) { stream = await toQcow2Stream(vmdk, { signal }) } else {