fix(xo-server): ensure import task progress is correctly updated (#10133)

@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()
+}

@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)
+})

@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) {

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

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 {