NuKVM/xen-orchestra
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(backups/importIncrementalVm): move base detection to callers (#7165)

Browse Source
This commit is contained in:
Florent BEAUCHAMP
2023-11-20 14:52:59 +01:00
committed by GitHub
parent ecbbf878d0
commit 4504141cbf
5 changed files with 102 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
@xen-orchestra/backups/ImportVmBackup.mjs
View File

@@ -5,6 +5,16 @@ import { importIncrementalVm } from './_incrementalVm.mjs'
import { Task } from './Task.mjs'
import { watchStreamSize } from './_watchStreamSize.mjs'
async function resolveUuid(xapi, cache, uuid, type) {
if (uuid == null) {
return uuid
}
const ref = cache.get(uuid)
if (ref === undefined) {
cache.set(uuid, xapi.call(`${type}.get_by_uuid`, uuid))
}
return cache.get(uuid)
}
export class ImportVmBackup {
constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses, mapVdisSrs = {} } = {} }) {
this._adapter = adapter
@@ -14,13 +24,29 @@ export class ImportVmBackup {
this._xapi = xapi
}
async #decorateIncrementalVmMetadata(backup) {
const { mapVdisSrs } = this._importIncrementalVmSettings
const xapi = this._xapi
const cache = new Map()
const mapVdisSrRefs = {}
for (const [vdiUuid, srUuid] of Object.entries(mapVdisSrs)) {
mapVdisSrRefs[vdiUuid] = await resolveUuid(xapi, cache, srUuid, 'SR')
}
const sr = await resolveUuid(xapi, cache, this._srUuid, 'SR')
Object.values(backup.vdis).forEach(vdi => {
vdi.SR = mapVdisSrRefs[vdi.uuid] ?? sr.$ref
})
return backup
}
async run() {
const adapter = this._adapter
const metadata = this._metadata
const isFull = metadata.mode === 'full'
const sizeContainer = { size: 0 }
const { mapVdisSrs, newMacAddresses } = this._importIncrementalVmSettings
let backup
if (isFull) {
backup = await adapter.readFullVmBackup(metadata)
@@ -29,11 +55,11 @@ export class ImportVmBackup {
assert.strictEqual(metadata.mode, 'delta')
const ignoredVdis = new Set(
Object.entries(this._importIncrementalVmSettings.mapVdisSrs)
Object.entries(mapVdisSrs)
.filter(([_, srUuid]) => srUuid === null)
.map(([vdiUuid]) => vdiUuid)
)
backup = await adapter.readIncrementalVmBackup(metadata, ignoredVdis)
backup = await this.#decorateIncrementalVmMetadata(await adapter.readIncrementalVmBackup(metadata, ignoredVdis))
Object.values(backup.streams).forEach(stream => watchStreamSize(stream, sizeContainer))
}
@@ -48,8 +74,7 @@ export class ImportVmBackup {
const vmRef = isFull
? await xapi.VM_import(backup, srRef)
: await importIncrementalVm(backup, await xapi.getRecord('SR', srRef), {
...this._importIncrementalVmSettings,
detectBase: false,
newMacAddresses,
})
await Promise.all([

92
@xen-orchestra/backups/_incrementalVm.mjs
View File

@@ -1,4 +1,3 @@
import find from 'lodash/find.js'
import groupBy from 'lodash/groupBy.js'
import ignoreErrors from 'promise-toolbox/ignoreErrors'
import omit from 'lodash/omit.js'
@@ -12,24 +11,18 @@ import { cancelableMap } from './_cancelableMap.mjs'
import { Task } from './Task.mjs'
import pick from 'lodash/pick.js'
// in `other_config` of an incrementally replicated VM, contains the UUID of the source VM
export const TAG_BASE_DELTA = 'xo:base_delta'
// in `other_config` of an incrementally replicated VM, contains the UUID of the target SR used for replication
//
// added after the complete replication
export const TAG_BACKUP_SR = 'xo:backup:sr'
// in other_config of VDIs of an incrementally replicated VM, contains the UUID of the source VDI
export const TAG_COPY_SRC = 'xo:copy_of'
const TAG_BACKUP_SR = 'xo:backup:sr'
const ensureArray = value => (value === undefined ? [] : Array.isArray(value) ? value : [value])
const resolveUuid = async (xapi, cache, uuid, type) => {
if (uuid == null) {
return uuid
}
let ref = cache.get(uuid)
if (ref === undefined) {
ref = await xapi.call(`${type}.get_by_uuid`, uuid)
cache.set(uuid, ref)
}
return ref
}
export async function exportIncrementalVm(
vm,
@@ -147,7 +140,7 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
$defer,
incrementalVm,
sr,
{ cancelToken = CancelToken.none, detectBase = true, mapVdisSrs = {}, newMacAddresses = false } = {}
{ cancelToken = CancelToken.none, newMacAddresses = false } = {}
) {
const { version } = incrementalVm
if (compareVersions(version, '1.0.0') < 0) {
@@ -157,35 +150,6 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
const vmRecord = incrementalVm.vm
const xapi = sr.$xapi
let baseVm
if (detectBase) {
const remoteBaseVmUuid = vmRecord.other_config[TAG_BASE_DELTA]
if (remoteBaseVmUuid) {
baseVm = find(
xapi.objects.all,
obj => (obj = obj.other_config) && obj[TAG_COPY_SRC] === remoteBaseVmUuid && obj[TAG_BACKUP_SR] === sr.$id
)
if (!baseVm) {
throw new Error(`could not find the base VM (copy of ${remoteBaseVmUuid})`)
}
}
}
const cache = new Map()
const mapVdisSrRefs = {}
for (const [vdiUuid, srUuid] of Object.entries(mapVdisSrs)) {
mapVdisSrRefs[vdiUuid] = await resolveUuid(xapi, cache, srUuid, 'SR')
}
const baseVdis = {}
baseVm &&
baseVm.$VBDs.forEach(vbd => {
const vdi = vbd.$VDI
if (vdi !== undefined) {
baseVdis[vbd.VDI] = vbd.$VDI
}
})
const vdiRecords = incrementalVm.vdis
// 0. Create suspend_VDI
@@ -197,18 +161,7 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
vm: pick(vmRecord, 'uuid', 'name_label', 'suspend_VDI'),
})
} else {
suspendVdi = await xapi.getRecord(
'VDI',
await xapi.VDI_create({
...vdi,
other_config: {
...vdi.other_config,
[TAG_BASE_DELTA]: undefined,
[TAG_COPY_SRC]: vdi.uuid,
},
SR: mapVdisSrRefs[vdi.uuid] ?? sr.$ref,
})
)
suspendVdi = await xapi.getRecord('VDI', await xapi.VDI_create(vdi))
$defer.onFailure(() => suspendVdi.$destroy())
}
}
@@ -226,10 +179,6 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
ha_always_run: false,
is_a_template: false,
name_label: '[Importing…] ' + vmRecord.name_label,
other_config: {
...vmRecord.other_config,
[TAG_COPY_SRC]: vmRecord.uuid,
},
},
{
bios_strings: vmRecord.bios_strings,
@@ -250,14 +199,8 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
const vdi = vdiRecords[vdiRef]
let newVdi
const remoteBaseVdiUuid = detectBase && vdi.other_config[TAG_BASE_DELTA]
if (remoteBaseVdiUuid) {
const baseVdi = find(baseVdis, vdi => vdi.other_config[TAG_COPY_SRC] === remoteBaseVdiUuid)
if (!baseVdi) {
throw new Error(`missing base VDI (copy of ${remoteBaseVdiUuid})`)
}
newVdi = await xapi.getRecord('VDI', await baseVdi.$clone())
if (vdi.baseVdi !== undefined) {
newVdi = await xapi.getRecord('VDI', await vdi.baseVdi.$clone())
$defer.onFailure(() => newVdi.$destroy())
await newVdi.update_other_config(TAG_COPY_SRC, vdi.uuid)
@@ -268,18 +211,7 @@ export const importIncrementalVm = defer(async function importIncrementalVm(
// suspendVDI has already created
newVdi = suspendVdi
} else {
newVdi = await xapi.getRecord(
'VDI',
await xapi.VDI_create({
...vdi,
other_config: {
...vdi.other_config,
[TAG_BASE_DELTA]: undefined,
[TAG_COPY_SRC]: vdi.uuid,
},
SR: mapVdisSrRefs[vdi.uuid] ?? sr.$ref,
})
)
newVdi = await xapi.getRecord('VDI', await xapi.VDI_create(vdi))
$defer.onFailure(() => newVdi.$destroy())
}

10
@xen-orchestra/backups/_runners/_vmRunners/_forkDeltaExport.mjs
View File

@@ -1,11 +1,11 @@
import cloneDeep from 'lodash/cloneDeep.js'
import mapValues from 'lodash/mapValues.js'
import { forkStreamUnpipe } from '../_forkStreamUnpipe.mjs'
export function forkDeltaExport(deltaExport) {
return Object.create(deltaExport, {
streams: {
value: mapValues(deltaExport.streams, forkStreamUnpipe),
},
})
const { streams, ...rest } = deltaExport
const newMetadata = cloneDeep(rest)
newMetadata.streams = mapValues(streams, forkStreamUnpipe)
return newMetadata
}

3
@xen-orchestra/backups/_runners/_writers/IncrementalRemoteWriter.mjs
View File

@@ -11,6 +11,7 @@ import { dirname } from 'node:path'
import { formatFilenameDate } from '../../_filenameDate.mjs'
import { getOldEntries } from '../../_getOldEntries.mjs'
import { TAG_BASE_DELTA } from '../../_incrementalVm.mjs'
import { Task } from '../../Task.mjs'
import { MixinRemoteWriter } from './_MixinRemoteWriter.mjs'
@@ -195,7 +196,7 @@ export class IncrementalRemoteWriter extends MixinRemoteWriter(AbstractIncrement
assert.notStrictEqual(
parentPath,
undefined,
`missing parent of ${id} in ${dirname(path)}, looking for ${vdi.other_config['xo:base_delta']}`
`missing parent of ${id} in ${dirname(path)}, looking for ${vdi.other_config[TAG_BASE_DELTA]}`
)
parentPath = parentPath.slice(1) // remove leading slash

66
@xen-orchestra/backups/_runners/_writers/IncrementalXapiWriter.mjs
View File

@@ -4,12 +4,13 @@ import { formatDateTime } from '@xen-orchestra/xapi'
import { formatFilenameDate } from '../../_filenameDate.mjs'
import { getOldEntries } from '../../_getOldEntries.mjs'
import { importIncrementalVm, TAG_COPY_SRC } from '../../_incrementalVm.mjs'
import { importIncrementalVm, TAG_BACKUP_SR, TAG_BASE_DELTA, TAG_COPY_SRC } from '../../_incrementalVm.mjs'
import { Task } from '../../Task.mjs'
import { AbstractIncrementalWriter } from './_AbstractIncrementalWriter.mjs'
import { MixinXapiWriter } from './_MixinXapiWriter.mjs'
import { listReplicatedVms } from './_listReplicatedVms.mjs'
import find from 'lodash/find.js'
export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWriter) {
async checkBaseVdis(baseUuidToSrcVdi, baseVm) {
@@ -81,6 +82,54 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
return asyncMapSettled(this._oldEntries, vm => vm.$destroy())
}
#decorateVmMetadata(backup) {
const { _warmMigration } = this._settings
const sr = this._sr
const xapi = sr.$xapi
const vm = backup.vm
vm.other_config[TAG_COPY_SRC] = vm.uuid
const remoteBaseVmUuid = vm.other_config[TAG_BASE_DELTA]
let baseVm
if (remoteBaseVmUuid) {
baseVm = find(
xapi.objects.all,
obj => (obj = obj.other_config) && obj[TAG_COPY_SRC] === remoteBaseVmUuid && obj[TAG_BACKUP_SR] === sr.$id
)
if (!baseVm) {
throw new Error(`could not find the base VM (copy of ${remoteBaseVmUuid})`)
}
}
const baseVdis = {}
baseVm?.$VBDs.forEach(vbd => {
const vdi = vbd.$VDI
if (vdi !== undefined) {
baseVdis[vbd.VDI] = vbd.$VDI
}
})
vm.other_config[TAG_COPY_SRC] = vm.uuid
if (!_warmMigration) {
vm.tags.push('Continuous Replication')
}
Object.values(backup.vdis).forEach(vdi => {
vdi.other_config[TAG_COPY_SRC] = vdi.uuid
vdi.SR = sr.$ref
// vdi.other_config[TAG_BASE_DELTA] is never defined on a suspend vdi
if (vdi.other_config[TAG_BASE_DELTA]) {
const remoteBaseVdiUuid = vdi.other_config[TAG_BASE_DELTA]
const baseVdi = find(baseVdis, vdi => vdi.other_config[TAG_COPY_SRC] === remoteBaseVdiUuid)
if (!baseVdi) {
throw new Error(`missing base VDI (copy of ${remoteBaseVdiUuid})`)
}
vdi.baseVdi = baseVdi
}
})
return backup
}
async _transfer({ timestamp, deltaExport, sizeContainers, vm }) {
const { _warmMigration } = this._settings
const sr = this._sr
@@ -91,16 +140,7 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
let targetVmRef
await Task.run({ name: 'transfer' }, async () => {
targetVmRef = await importIncrementalVm(
{
__proto__: deltaExport,
vm: {
...deltaExport.vm,
tags: _warmMigration ? deltaExport.vm.tags : [...deltaExport.vm.tags, 'Continuous Replication'],
},
},
sr
)
targetVmRef = await importIncrementalVm(this.#decorateVmMetadata(deltaExport), sr)
return {
size: Object.values(sizeContainers).reduce((sum, { size }) => sum + size, 0),
}
@@ -121,13 +161,13 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
)
),
targetVm.update_other_config({
'xo:backup:sr': srUuid,
[TAG_BACKUP_SR]: srUuid,
// these entries need to be added in case of offline backup
'xo:backup:datetime': formatDateTime(timestamp),
'xo:backup:job': job.id,
'xo:backup:schedule': scheduleId,
'xo:backup:vm': vm.uuid,
[TAG_BASE_DELTA]: vm.uuid,
}),
])
}