mirror of
https://github.com/vatesfr/xen-orchestra.git
synced 2026-11-09 17:06:13 -06:00
feat(backups): add progress to incremental mirror (#9901)
This commit is contained in:
committed by
GitHub
parent
d8094a71e2
commit
8939a2deeb
@@ -1,4 +1,4 @@
|
||||
import { asyncMap } from '@xen-orchestra/async-map'
|
||||
import { asyncEach } from '@vates/async-each'
|
||||
import { Task } from '@vates/task'
|
||||
import Disposable from 'promise-toolbox/Disposable'
|
||||
import ignoreErrors from 'promise-toolbox/ignoreErrors'
|
||||
@@ -82,10 +82,11 @@ export const Metadata = class MetadataBackupRunner extends Abstract {
|
||||
// remove pools that failed (already handled)
|
||||
pools = pools.filter(_ => _ !== undefined)
|
||||
|
||||
const promises = []
|
||||
if (pools.length !== 0 && settings.retentionPoolMetadata !== 0) {
|
||||
promises.push(
|
||||
asyncMap(pools, async pool =>
|
||||
const tasks = []
|
||||
|
||||
if (settings.retentionPoolMetadata !== 0) {
|
||||
for (const pool of pools) {
|
||||
tasks.push(async () =>
|
||||
Task.run(
|
||||
{
|
||||
properties: {
|
||||
@@ -107,11 +108,11 @@ export const Metadata = class MetadataBackupRunner extends Abstract {
|
||||
}).run()
|
||||
).catch(noop)
|
||||
)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
if (job.xoMetadata !== undefined && settings.retentionXoMetadata !== 0) {
|
||||
promises.push(
|
||||
tasks.push(() =>
|
||||
Task.run(
|
||||
{
|
||||
properties: {
|
||||
@@ -130,7 +131,16 @@ export const Metadata = class MetadataBackupRunner extends Abstract {
|
||||
).catch(noop)
|
||||
)
|
||||
}
|
||||
await Promise.all(promises)
|
||||
|
||||
const total = tasks.length
|
||||
let transferred = 0
|
||||
|
||||
await asyncEach(tasks, async task => {
|
||||
await task()
|
||||
transferred++
|
||||
Task.set('progress', Math.round(transferred / total))
|
||||
})
|
||||
Task.set('progress', 1)
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
@@ -63,6 +63,8 @@ export const VmsRemote = class RemoteVmsBackupRunner extends Abstract {
|
||||
const vmsUuids = await sourceRemoteAdapter.listAllVms()
|
||||
|
||||
Task.info('vms', { vms: vmsUuids })
|
||||
const nbVms = vmsUuids.length
|
||||
let nbVmsDone = 0
|
||||
|
||||
remoteAdapters = getAdaptersByRemote(remoteAdapters)
|
||||
const allSettings = this._job.settings
|
||||
@@ -123,11 +125,14 @@ export const VmsRemote = class RemoteVmsBackupRunner extends Abstract {
|
||||
)
|
||||
.then(result => {
|
||||
if (taskError === undefined) {
|
||||
nbVmsDone++
|
||||
return task.success(result)
|
||||
}
|
||||
if (isLastRun) {
|
||||
nbVmsDone++
|
||||
return task.failure(taskError)
|
||||
}
|
||||
Task.set('progress', Math.round((nbVmsDone * 100) / nbVms))
|
||||
// don't end the task
|
||||
task.warning(`Retry the VM mirror backup due to an error`, {
|
||||
attempt: nTriesByVmId[vmUuid],
|
||||
@@ -148,6 +153,7 @@ export const VmsRemote = class RemoteVmsBackupRunner extends Abstract {
|
||||
|
||||
await asyncMapSettled(vmIds, _handleVm)
|
||||
}
|
||||
Task.set('progress', 100)
|
||||
}
|
||||
)
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ import { FullRemoteWriter } from '../_writers/FullRemoteWriter.mjs'
|
||||
import { forkStreamUnpipe } from '../_forkStreamUnpipe.mjs'
|
||||
import { watchStreamSize } from '../../_watchStreamSize.mjs'
|
||||
import { AggregatedFullRemoteWriter } from '../_writers/AggregatedFullRemoteWriter.mjs'
|
||||
import { Task } from '@vates/task'
|
||||
|
||||
export const FullRemote = class FullRemoteVmBackupRunner extends AbstractRemote {
|
||||
_getRemoteWriters() {
|
||||
@@ -15,7 +16,8 @@ export const FullRemote = class FullRemoteVmBackupRunner extends AbstractRemote
|
||||
|
||||
async _run() {
|
||||
const transferList = await this._computeTransferList(({ mode }) => mode === 'full')
|
||||
|
||||
const nbTransferrableVms = transferList.length
|
||||
let nbTransferredVms = 0
|
||||
for (const metadata of transferList) {
|
||||
const stream = this._throttleStream(await this._sourceRemoteAdapter.readFullVmBackup(metadata))
|
||||
const sizeContainer = watchStreamSize(stream)
|
||||
@@ -27,7 +29,7 @@ export const FullRemote = class FullRemoteVmBackupRunner extends AbstractRemote
|
||||
stream: forkStreamUnpipe(stream),
|
||||
// stream will be forked and transformed, it's not safe to attach additional properties to it
|
||||
streamLength: stream.length,
|
||||
maxStreamLength: stream.maxStreamLength, // on encrypted source
|
||||
maxStreamLength: stream.maxStreamLength, // for encrypted destination/source without length
|
||||
timestamp: metadata.timestamp,
|
||||
vm: metadata.vm,
|
||||
vmSnapshot: metadata.vmSnapshot,
|
||||
@@ -37,7 +39,10 @@ export const FullRemote = class FullRemoteVmBackupRunner extends AbstractRemote
|
||||
)
|
||||
// for healthcheck
|
||||
this._tags = metadata.vm.tags
|
||||
nbTransferredVms++
|
||||
Task.set('progress', Math.round((nbTransferredVms * 100) / nbTransferrableVms))
|
||||
}
|
||||
Task.set('progress', 100)
|
||||
this._hasTransferredData = transferList.length > 0
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,8 @@ import { openVhd } from 'vhd-lib'
|
||||
import { getVmBackupDir } from '../../_getVmBackupDir.mjs'
|
||||
import { SynchronizedDisk, ThrottledDisk } from '@xen-orchestra/disk-transform'
|
||||
import { AggregatedIncrementalRemoteWriter } from '../_writers/AggregatedIncrementalRemoteWriter.mjs'
|
||||
import { Task } from '@vates/task'
|
||||
import { TaskProgressHandler } from './_TaskProgressHandler.mjs'
|
||||
|
||||
const { warn } = createLogger('xo:backups:Incrementalremote')
|
||||
class IncrementalRemoteVmBackupRunner extends AbstractRemote {
|
||||
@@ -60,7 +62,8 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
|
||||
}
|
||||
async _run() {
|
||||
const transferList = await this._computeTransferList(({ mode }) => mode === 'delta')
|
||||
|
||||
const nbTransferrableVms = transferList.length
|
||||
let nbTransferredVms = 0
|
||||
for (const metadata of transferList) {
|
||||
assert.strictEqual(metadata.mode, 'delta')
|
||||
const incrementalExport = await this._sourceRemoteAdapter.readIncrementalVmBackup(metadata, undefined, {
|
||||
@@ -74,6 +77,7 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
|
||||
for (const key in incrementalExport.disks) {
|
||||
let disk = incrementalExport.disks[key]
|
||||
isVhdDifferencing[key] = disk.isDifferencing()
|
||||
disk.addProgressHandler(new TaskProgressHandler())
|
||||
disk = new ThrottledDisk(disk, this._throttleGenerator)
|
||||
incrementalExport.disks[key] = new SynchronizedDisk(disk)
|
||||
}
|
||||
@@ -116,7 +120,10 @@ class IncrementalRemoteVmBackupRunner extends AbstractRemote {
|
||||
await this._callWriters(writer => writer.cleanup(), 'writer.cleanup()')
|
||||
// for healthcheck
|
||||
this._tags = metadata.vm.tags
|
||||
nbTransferredVms++
|
||||
Task.set('progress', Math.round((nbTransferredVms * 100) / nbTransferrableVms))
|
||||
}
|
||||
Task.set('progress', 100)
|
||||
this._hasTransferredData = transferList.length > 0
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ import {
|
||||
import { ThrottledDisk, SynchronizedDisk } from '@xen-orchestra/disk-transform'
|
||||
import { AggregatedIncrementalRemoteWriter } from '../_writers/AggregatedIncrementalRemoteWriter.mjs'
|
||||
import { AggregatedIncrementalXapiWriter } from '../_writers/AggregatedIncrementalXapiWriter.mjs'
|
||||
import { TaskProgressHandler } from './_TaskProgressHandler.mjs'
|
||||
|
||||
const { debug } = createLogger('xo:backups:IncrementalXapiVmBackup')
|
||||
|
||||
@@ -52,6 +53,7 @@ export const IncrementalXapi = class IncrementalXapiVmBackupRunner extends Abstr
|
||||
let useNbd = false
|
||||
for (const key in deltaExport.disks) {
|
||||
let disk = deltaExport.disks[key]
|
||||
disk.addProgressHandler(new TaskProgressHandler())
|
||||
isVhdDifferencing[key] = disk.isDifferencing()
|
||||
if (!isFull && !isVhdDifferencing[key] && key !== exportedVm.$suspend_VDI?.$ref) {
|
||||
Task.warning('Backup fell back to a full')
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
import { Task } from '@vates/task'
|
||||
|
||||
const MAX_DURATION_BETWEEN_PROGRESS_EMIT = 5e3
|
||||
const MIN_THRESHOLD_PERCENT_BETWEEN_PROGRESS_EMIT = 0.01
|
||||
|
||||
export class TaskProgressHandler {
|
||||
#lastProgressDate
|
||||
#lastProgressValue
|
||||
constructor() {
|
||||
Task.set('progress', 0)
|
||||
}
|
||||
async setProgress(progress) {
|
||||
if (progress < 0 || progress > 1) {
|
||||
return
|
||||
}
|
||||
if (
|
||||
this.#lastProgressDate !== undefined &&
|
||||
this.#lastProgressValue !== undefined &&
|
||||
Date.now() - this.#lastProgressDate < MAX_DURATION_BETWEEN_PROGRESS_EMIT &&
|
||||
progress - this.#lastProgressValue < MIN_THRESHOLD_PERCENT_BETWEEN_PROGRESS_EMIT
|
||||
) {
|
||||
return
|
||||
}
|
||||
this.#lastProgressDate = Date.now()
|
||||
this.#lastProgressValue = progress
|
||||
Task.set('progress', Math.round(progress * 100))
|
||||
}
|
||||
done() {}
|
||||
}
|
||||
@@ -1,6 +1,16 @@
|
||||
import { TaskProgressHandler } from './_runners/_vmRunners/_TaskProgressHandler.mjs'
|
||||
export function watchStreamSize(stream, container = { size: 0 }) {
|
||||
const progressHandler = container.progressHandler ?? new TaskProgressHandler()
|
||||
stream.on('data', data => {
|
||||
container.size += data.length
|
||||
if (stream.length) {
|
||||
// empty for xva exported by xapi
|
||||
progressHandler.setProgress(container.size / stream.length)
|
||||
}
|
||||
})
|
||||
|
||||
stream.on('finish', () => {
|
||||
progressHandler.setProgress(1)
|
||||
})
|
||||
stream.pause()
|
||||
return container
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { ProgressHandler } from './ProgressHandler.mjs'
|
||||
import { ComposedProgressHandler, ProgressHandler } from './ProgressHandler.mjs'
|
||||
|
||||
export type DiskBlockData = Buffer
|
||||
export type DiskBlock = {
|
||||
@@ -11,7 +11,11 @@ export type BytesLength = number
|
||||
export abstract class Disk {
|
||||
#generatedDiskBlocks = 0
|
||||
#parent?: Disk
|
||||
progressHandler?: ProgressHandler
|
||||
#progressHandler = new ComposedProgressHandler()
|
||||
|
||||
get progressHandler(): ProgressHandler {
|
||||
return this.#progressHandler
|
||||
}
|
||||
get parent(): Disk | undefined {
|
||||
return this.#parent
|
||||
}
|
||||
@@ -66,9 +70,9 @@ export abstract class Disk {
|
||||
return Promise.resolve(true)
|
||||
}
|
||||
|
||||
// @todo readRawData will be needed for file level restore
|
||||
// @todo rename will be needed for merge
|
||||
// @todo : should the alias part moved from remote/vhd ?
|
||||
addProgressHandler(progressHandler: ProgressHandler) {
|
||||
this.#progressHandler.add(progressHandler)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -42,12 +42,12 @@ test('Disk class', async t => {
|
||||
|
||||
await t.test('diskBlocks calls progressHandler.done when consumer exits early', async () => {
|
||||
let doneCalled = false
|
||||
disk.progressHandler = {
|
||||
disk.addProgressHandler({
|
||||
setProgress: async () => {},
|
||||
done: async () => {
|
||||
doneCalled = true
|
||||
},
|
||||
}
|
||||
})
|
||||
disk.blocks[4] = true
|
||||
disk.blocks[5] = true
|
||||
disk.blocks[6] = true
|
||||
|
||||
@@ -2,3 +2,17 @@ export interface ProgressHandler {
|
||||
setProgress(total: number): void | Promise<void>
|
||||
done(): void | Promise<void>
|
||||
}
|
||||
|
||||
export class ComposedProgressHandler implements ProgressHandler {
|
||||
#handlers: Array<ProgressHandler> = []
|
||||
async setProgress(total: number) {
|
||||
await Promise.all(this.#handlers.map(handler => handler.setProgress(total)))
|
||||
}
|
||||
async done() {
|
||||
await Promise.all(this.#handlers.map(handler => handler.done()))
|
||||
}
|
||||
|
||||
add(handler: ProgressHandler) {
|
||||
this.#handlers.push(handler)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,42 +67,46 @@ export default class Tasks extends EventEmitter {
|
||||
this.#handleOnProgressScheduled = false
|
||||
}
|
||||
|
||||
#bufferTaskUpdate = taskLog => {
|
||||
const buf = this.#onProgressBuffer
|
||||
|
||||
if (!this.#handleOnProgressScheduled) {
|
||||
this.#handleOnProgressScheduled = true
|
||||
|
||||
// task events are buffered, deduplicated and will be handled one after
|
||||
// the others on the next tick to avoid race conditions
|
||||
process.nextTick(this.#handleOnProgressBuffer)
|
||||
}
|
||||
|
||||
// Error objects are not JSON-ifiable by default
|
||||
const { result } = taskLog
|
||||
if (result instanceof Error && result.toJSON === undefined) {
|
||||
taskLog.result = serializeError(result)
|
||||
}
|
||||
|
||||
const { $root } = taskLog
|
||||
const { status, id } = $root
|
||||
if (status !== 'pending') {
|
||||
if (this.#logsToClearOnSuccess.has(id)) {
|
||||
this.#logsToClearOnSuccess.delete(id)
|
||||
|
||||
if (status === 'success') {
|
||||
return buf.set(id, null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$root.updatedAt = Date.now()
|
||||
buf.set(id, $root)
|
||||
}
|
||||
|
||||
#onProgress = makeOnProgress({
|
||||
onRootTaskEnd: taskLog => {
|
||||
const { id } = taskLog
|
||||
this.#tasks.delete(id)
|
||||
},
|
||||
onTaskUpdate: taskLog => {
|
||||
const buf = this.#onProgressBuffer
|
||||
|
||||
if (!this.#handleOnProgressScheduled) {
|
||||
this.#handleOnProgressScheduled = true
|
||||
|
||||
// task events are buffered, deduplicated and will be handled one after
|
||||
// the others on the next tick to avoid race conditions
|
||||
process.nextTick(this.#handleOnProgressBuffer)
|
||||
}
|
||||
|
||||
// Error objects are not JSON-ifiable by default
|
||||
const { result } = taskLog
|
||||
if (result instanceof Error && result.toJSON === undefined) {
|
||||
taskLog.result = serializeError(result)
|
||||
}
|
||||
|
||||
const { $root } = taskLog
|
||||
const { status, id } = $root
|
||||
if (status !== 'pending') {
|
||||
if (this.#logsToClearOnSuccess.has(id)) {
|
||||
this.#logsToClearOnSuccess.delete(id)
|
||||
|
||||
if (status === 'success') {
|
||||
return buf.set(id, null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
$root.updatedAt = Date.now()
|
||||
buf.set(id, $root)
|
||||
this.#bufferTaskUpdate(taskLog)
|
||||
},
|
||||
})
|
||||
|
||||
@@ -181,6 +185,28 @@ export default class Tasks extends EventEmitter {
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates an onProgress callback that feeds raw @vates/task events from an
|
||||
* external source (worker process, XO Proxy) into Tasks.mjs's store and
|
||||
* event emitter, making the task visible in the UI.
|
||||
*
|
||||
* Each call returns an independent makeOnProgress context (isolated taskLogs
|
||||
* Map), so concurrent backup jobs don't interfere with each other.
|
||||
*
|
||||
* Abort support for external tasks is not implemented — the external process
|
||||
* controls its own lifecycle.
|
||||
*/
|
||||
createExternalProgressHandler({ onRootTaskEnd, onRootTaskStart, onTaskUpdate } = {}) {
|
||||
return makeOnProgress({
|
||||
onRootTaskEnd,
|
||||
onRootTaskStart,
|
||||
onTaskUpdate: (taskLog, event) => {
|
||||
this.#bufferTaskUpdate(taskLog)
|
||||
onTaskUpdate?.(taskLog, event)
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
async abort(id, reason) {
|
||||
const task = this.#tasks.get(id)
|
||||
if (task === undefined) {
|
||||
|
||||
@@ -125,7 +125,8 @@ export class XapiDiskSource extends DiskPassthrough {
|
||||
const readAhead = new ReadAhead(source)
|
||||
source = new TimeoutDisk(source, this.#timeout)
|
||||
const label = await xapi.getField('VDI', vdiRef, 'name_label')
|
||||
readAhead.progressHandler = new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD`)
|
||||
// manually create an export task for NBD since xapi xan't do it automatically
|
||||
readAhead.addProgressHandler(new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD`))
|
||||
return readAhead
|
||||
}
|
||||
|
||||
@@ -228,7 +229,8 @@ export class XapiDiskSource extends DiskPassthrough {
|
||||
source = new DiskLargerBlock(source, this.#blockSize)
|
||||
}
|
||||
const label = await xapi.getField('VDI', vdiRef, 'name_label')
|
||||
readAhead.progressHandler = new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD+CBT`)
|
||||
// manually create an export task for NBD since xapi xan't do it automatically
|
||||
readAhead.addProgressHandler(new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD+CBT`))
|
||||
return readAhead
|
||||
} catch (error) {
|
||||
if (error.code !== 'CBT_DISABLED') {
|
||||
|
||||
@@ -24,6 +24,9 @@
|
||||
- `POST /vifs/:id/actions/connect` (PR [#9889](https://github.com/vatesfr/xen-orchestra/pull/9889))
|
||||
- `POST /vifs/:id/actions/disconnect` (PR [#9889](https://github.com/vatesfr/xen-orchestra/pull/9889))
|
||||
|
||||
- [XO6] live update XO tasks (PR [#9901](https://github.com/vatesfr/xen-orchestra/pull/9901))
|
||||
- [XO6/Backup] add progress for backups tasks(PR [#9901](https://github.com/vatesfr/xen-orchestra/pull/9901))
|
||||
|
||||
### Bug fixes
|
||||
|
||||
> Users must be able to say: "I had this issue, happy to know it's fixed"
|
||||
@@ -49,10 +52,13 @@
|
||||
> Keep this list alphabetically ordered to avoid merge conflicts
|
||||
|
||||
<!--packages-start-->
|
||||
|
||||
- @vates/types minor
|
||||
- @xen-orchestra/acl minor
|
||||
- @xen-orchestra/qcow2 patch
|
||||
- @xen-orchestra/rest-api minor
|
||||
- @xen-orchestra/xapi minor
|
||||
- xo-server minor
|
||||
- xo-server-sdn-controller patch
|
||||
- xo-web patch
|
||||
|
||||
|
||||
@@ -40,7 +40,7 @@ export function forwardResult(log) {
|
||||
}
|
||||
|
||||
// it records logs generated by backup task logs
|
||||
export const handleBackupLog = (taskLog, event, { app, jobName, store }) => {
|
||||
export const handleBackupLog = (taskLog, event, { app, jobName } = {}) => {
|
||||
if (event.type === 'end') {
|
||||
taskLog.status = computeStatusAndSortTasks(getStatus(taskLog.result, taskLog.status), taskLog.tasks)
|
||||
}
|
||||
@@ -63,8 +63,6 @@ export const handleBackupLog = (taskLog, event, { app, jobName, store }) => {
|
||||
}
|
||||
}
|
||||
|
||||
store.put(taskLog.$root.id, taskLog.$root)
|
||||
|
||||
// end of the root task: return/throw the result
|
||||
if (event.type === 'end' && taskLog.$root === taskLog) {
|
||||
return forwardResult(taskLog)
|
||||
|
||||
@@ -3,7 +3,6 @@ import Disposable from 'promise-toolbox/Disposable'
|
||||
import forOwn from 'lodash/forOwn.js'
|
||||
import groupBy from 'lodash/groupBy.js'
|
||||
import merge from 'lodash/merge.js'
|
||||
import { makeOnProgress } from '@vates/task/combineEvents'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
import { createPredicate } from 'value-matcher'
|
||||
import { decorateWith } from '@vates/decorate-with'
|
||||
@@ -66,12 +65,9 @@ export default class BackupNg {
|
||||
|
||||
constructor(app) {
|
||||
this._app = app
|
||||
this._store = undefined
|
||||
this._runningRestores = new Set()
|
||||
|
||||
app.hooks.on('start', async () => {
|
||||
this._store = await app.getStore('tasks')
|
||||
|
||||
const executor = async ({
|
||||
cancelToken,
|
||||
data,
|
||||
@@ -162,12 +158,12 @@ export default class BackupNg {
|
||||
const targetRemoteIds = unboxIdsFromPattern(job.remotes)
|
||||
try {
|
||||
if (!useXoProxy && backupsConfig.disableWorkers) {
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
jobUpdateFct(log.id).catch(logger.warn) // is async, but makeOnProgress doesn't await onRootTaskXXX functions
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name, store: this._store })
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name })
|
||||
},
|
||||
})
|
||||
return await Task.run(
|
||||
@@ -309,7 +305,7 @@ export default class BackupNg {
|
||||
)
|
||||
|
||||
let result
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
jobUpdateFct(log.id).catch(logger.warn) // is async, but makeOnProgress doesn't await onRootTaskXXX functions
|
||||
},
|
||||
@@ -317,7 +313,7 @@ export default class BackupNg {
|
||||
result = forwardResult(log)
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name, store: this._store })
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name })
|
||||
},
|
||||
})
|
||||
|
||||
@@ -334,7 +330,7 @@ export default class BackupNg {
|
||||
}
|
||||
} else {
|
||||
let result
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
jobUpdateFct(log.id).catch(logger.warn) // is async, but makeOnProgress doesn't await onRootTaskXXX functions
|
||||
},
|
||||
@@ -342,7 +338,7 @@ export default class BackupNg {
|
||||
result = forwardResult(log)
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name, store: this._store })
|
||||
handleBackupLog(log, event, { app: this._app, jobName: job.name })
|
||||
},
|
||||
})
|
||||
|
||||
@@ -365,7 +361,6 @@ export default class BackupNg {
|
||||
}
|
||||
app.registerJobExecutor('backup', executor)
|
||||
app.registerJobExecutor('mirrorBackup', executor)
|
||||
return () => this._store.close()
|
||||
})
|
||||
}
|
||||
|
||||
@@ -515,7 +510,7 @@ export default class BackupNg {
|
||||
assertType: 'iterator',
|
||||
})
|
||||
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
this._runningRestores.add(log.id)
|
||||
rootTaskId = log.id
|
||||
@@ -524,7 +519,7 @@ export default class BackupNg {
|
||||
result = forwardResult(log)
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { store: this._store })
|
||||
handleBackupLog(log, event)
|
||||
},
|
||||
})
|
||||
|
||||
@@ -542,13 +537,13 @@ export default class BackupNg {
|
||||
result = await Disposable.use(app.getBackupsRemoteAdapter(remote), async adapter => {
|
||||
const metadata = await adapter.readVmBackupMetadata(metadataFilename)
|
||||
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
this._runningRestores.add(log.id)
|
||||
rootTaskId = log.id
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { store: this._store })
|
||||
handleBackupLog(log, event)
|
||||
},
|
||||
})
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@ import cloneDeep from 'lodash/cloneDeep.js'
|
||||
import Disposable from 'promise-toolbox/Disposable'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
import { createRunner } from '@xen-orchestra/backups/Backup.mjs'
|
||||
import { makeOnProgress } from '@vates/task/combineEvents'
|
||||
import { parseMetadataBackupId } from '@xen-orchestra/backups/parseMetadataBackupId.mjs'
|
||||
import { RestoreMetadataBackup } from '@xen-orchestra/backups/RestoreMetadataBackup.mjs'
|
||||
import { Task } from '@vates/task'
|
||||
@@ -24,7 +23,6 @@ export default class metadataBackup {
|
||||
|
||||
constructor(app) {
|
||||
this._app = app
|
||||
this._store = undefined
|
||||
this._runningMetadataRestores = new Set()
|
||||
|
||||
const debounceDelay = app.config.getDuration('backups.listingDebounce')
|
||||
@@ -32,10 +30,7 @@ export default class metadataBackup {
|
||||
this._listPoolMetadataBackups = debounceWithKey(this._listPoolMetadataBackups, debounceDelay, remoteId => remoteId)
|
||||
|
||||
app.hooks.on('start', async () => {
|
||||
this._store = await app.getStore('tasks')
|
||||
|
||||
app.registerJobExecutor(METADATA_BACKUP_JOB_TYPE, this._executor.bind(this))
|
||||
return () => this._store.close()
|
||||
})
|
||||
}
|
||||
|
||||
@@ -111,7 +106,7 @@ export default class metadataBackup {
|
||||
})
|
||||
|
||||
let result
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
jobUpdateFct(log.id).catch(logger.warn) // is async, but makeOnProgress doesn't await onRootTaskXXX functions
|
||||
},
|
||||
@@ -119,7 +114,7 @@ export default class metadataBackup {
|
||||
result = forwardResult(log)
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { store: this._store })
|
||||
handleBackupLog(log, event)
|
||||
},
|
||||
})
|
||||
|
||||
@@ -130,12 +125,12 @@ export default class metadataBackup {
|
||||
} else {
|
||||
cancelToken.throwIfRequested()
|
||||
|
||||
const onLogFct = makeOnProgress({
|
||||
const onLogFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
jobUpdateFct(log.id).catch(logger.warn) // is async, but makeOnProgress doesn't await onRootTaskXXX functions
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { store: this._store })
|
||||
handleBackupLog(log, event)
|
||||
},
|
||||
})
|
||||
return Task.run(
|
||||
@@ -354,13 +349,13 @@ export default class metadataBackup {
|
||||
|
||||
let rootTaskId
|
||||
|
||||
const onProgressFct = makeOnProgress({
|
||||
const onProgressFct = app.tasks.createExternalProgressHandler({
|
||||
onRootTaskStart: log => {
|
||||
this._runningMetadataRestores.add(log.id)
|
||||
rootTaskId = log.id
|
||||
},
|
||||
onTaskUpdate: (log, event) => {
|
||||
handleBackupLog(log, event, { store: this._store })
|
||||
handleBackupLog(log, event)
|
||||
},
|
||||
})
|
||||
const onLogFct = async event => {
|
||||
|
||||
Reference in New Issue
Block a user