mirror of
https://github.com/vatesfr/xen-orchestra.git
synced 2026-10-12 02:24:05 +00:00
feat(backups): bi directional replication (#9806)
following the symmetrical backup PR
reuse any snapshot in common between the source and destination if
* using incremental replication
* only one target
* a common snapshot exists between source and destination , even from another job, based on the CONTENT_KEY
The same rule of reuse the existing VM target / create a new one of symmetrical backups applies
if the destination don't have any data between the snapshot and the active disk
update an existing VM
else create a new VM but chain disks
This commit is contained in:
Florent BEAUCHAMP
@@ -85,6 +85,7 @@ export async function getVmDeltaChainLength(xapi, vmRef) {
|
||||
export function resetVmOtherConfig(xapi, vmRef) {
|
||||
return applyToVmAndVdis(xapi, vmRef, (type, ref) => {
|
||||
return xapi.setFieldEntries(type, ref, 'other_config', {
|
||||
[CONTENT_KEY]: null,
|
||||
[COPY_OF]: null,
|
||||
[DATETIME]: null,
|
||||
[DELTA_CHAIN_LENGTH]: null,
|
||||
@@ -92,6 +93,7 @@ export function resetVmOtherConfig(xapi, vmRef) {
|
||||
[JOB_ID]: null,
|
||||
[SCHEDULE_ID]: null,
|
||||
[VM_UUID]: null,
|
||||
|
||||
// REPLICATED_TO_SR_UUID is not reset since we can replicate a replication
|
||||
})
|
||||
})
|
||||
|
||||
@@ -8,6 +8,7 @@ import { exportIncrementalVm } from '../../_incrementalVm.mjs'
|
||||
import { IncrementalRemoteWriter } from '../_writers/IncrementalRemoteWriter.mjs'
|
||||
import { IncrementalXapiWriter } from '../_writers/IncrementalXapiWriter.mjs'
|
||||
import {
|
||||
CONTENT_KEY,
|
||||
DATETIME,
|
||||
DELTA_CHAIN_LENGTH,
|
||||
EXPORTED_SUCCESSFULLY,
|
||||
@@ -101,63 +102,136 @@ export const IncrementalXapi = class IncrementalXapiVmBackupRunner extends Abstr
|
||||
await this._callWriters(writer => writer.cleanup(), 'writer.cleanup()')
|
||||
}
|
||||
|
||||
async _selectBaseVm() {
|
||||
const xapi = this._xapi
|
||||
|
||||
// filter _jobSnapshotVdis to have only the last successfully exported vdi
|
||||
// compute delta chain length
|
||||
// fill baseUuidToSrcVdi with this data
|
||||
|
||||
const exportedVdis = this._jobSnapshotVdis.filter(
|
||||
_ => EXPORTED_SUCCESSFULLY in _.other_config && DATETIME in _.other_config
|
||||
)
|
||||
let lastSuccessfullBackup
|
||||
let lastExportedVdis = []
|
||||
for (const exportedVdi of exportedVdis) {
|
||||
if (lastSuccessfullBackup === undefined || lastSuccessfullBackup < exportedVdi.other_config[DATETIME]) {
|
||||
lastExportedVdis = []
|
||||
lastSuccessfullBackup = exportedVdi.other_config[DATETIME]
|
||||
}
|
||||
if (lastSuccessfullBackup === exportedVdi.other_config[DATETIME]) {
|
||||
lastExportedVdis.push(exportedVdi)
|
||||
/**
|
||||
* Groups snapshot VDIs by their DATETIME other_config value and sorts them most-recent-first.
|
||||
*
|
||||
* @param {import('@vates/types').XenApiVdi[]} snapshotVdis
|
||||
* @returns {{ datetime: string, vdis: import('@vates/types').XenApiVdi[] }[]}
|
||||
*/
|
||||
_groupSnapshotVdisByDatetime(snapshotVdis) {
|
||||
const byDatetime = new Map()
|
||||
for (const vdi of snapshotVdis) {
|
||||
const datetime = vdi.other_config[DATETIME]
|
||||
if (!byDatetime.has(datetime)) {
|
||||
byDatetime.set(datetime, [])
|
||||
}
|
||||
byDatetime.get(datetime).push(vdi)
|
||||
}
|
||||
// sort them by decreasing datetime
|
||||
return [...byDatetime.entries()]
|
||||
.sort(([a], [b]) => b.localeCompare(a))
|
||||
.map(([datetime, vdis]) => ({ datetime, vdis }))
|
||||
}
|
||||
|
||||
this._baseVdis = {}
|
||||
if (lastExportedVdis.length === 0) {
|
||||
debug('no base VDIS found', {
|
||||
jobLength: this._jobSnapshotVdis.length,
|
||||
lastSuccessfullBackup,
|
||||
exportedLength: exportedVdis.length,
|
||||
})
|
||||
return
|
||||
}
|
||||
const deltaChainLength = Math.max(
|
||||
...lastExportedVdis.map(({ other_config }) => Number(other_config[DELTA_CHAIN_LENGTH] ?? 0))
|
||||
)
|
||||
|
||||
const srcVdis = keyBy(await xapi.getRecords('VDI', await this._vm.$getDisks()), '$ref')
|
||||
|
||||
/**
|
||||
* For each snapshot VDI in `snapshotVdis`, maps it to its live source VDI,
|
||||
* attaches its CONTENT_KEY when present, calls checkBaseVdis on all writers,
|
||||
* and returns both the full mapping and the subset confirmed present by writers.
|
||||
*
|
||||
* @param {import('@vates/types').XenApiVdi[]} snapshotVdis - Snapshot VDI records to use as base candidates
|
||||
* @param {Record<string, object>} srcVdisByRef - Live VM VDI records keyed by $ref
|
||||
* @param {{ useContentKey?: boolean }} [options]
|
||||
* @returns {Promise<{ presentBaseVdis: Map<string, string>, baseUuidToSrcVdiUuid: Map<string, string>}>}
|
||||
*/
|
||||
async _checkBaseVdis(snapshotVdis, srcVdisByRef, { useContentKey = false } = {}) {
|
||||
const baseUuidToSrcVdiUuid = new Map()
|
||||
for (const lastExportedVdi of lastExportedVdis) {
|
||||
const baseUuid = lastExportedVdi.uuid
|
||||
const snapshotOf = lastExportedVdi.snapshot_of
|
||||
const srcVdi = srcVdis[snapshotOf]
|
||||
const baseUuidToContentKey = new Map()
|
||||
for (const snapshotVdi of snapshotVdis) {
|
||||
const baseUuid = snapshotVdi.uuid
|
||||
const srcVdi = srcVdisByRef[snapshotVdi.snapshot_of]
|
||||
if (srcVdi !== undefined) {
|
||||
baseUuidToSrcVdiUuid.set(baseUuid, srcVdi.uuid)
|
||||
if (useContentKey) {
|
||||
const contentKey = snapshotVdi.other_config[CONTENT_KEY]
|
||||
if (contentKey !== undefined) {
|
||||
baseUuidToContentKey.set(baseUuid, contentKey)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
debug('ignore snapshot VDI because no longer present on VM', {
|
||||
vdi: baseUuid,
|
||||
})
|
||||
debug('ignore snapshot VDI because no longer present on VM', { vdi: baseUuid })
|
||||
}
|
||||
}
|
||||
|
||||
const presentBaseVdis = new Map(baseUuidToSrcVdiUuid)
|
||||
await this._callWriters(
|
||||
writer => presentBaseVdis.size !== 0 && writer.checkBaseVdis(presentBaseVdis),
|
||||
writer => presentBaseVdis.size !== 0 && writer.checkBaseVdis(presentBaseVdis, baseUuidToContentKey),
|
||||
'writer.checkBaseVdis()',
|
||||
false
|
||||
)
|
||||
return { presentBaseVdis, baseUuidToSrcVdiUuid }
|
||||
}
|
||||
|
||||
async _selectBaseVm() {
|
||||
const xapi = this._xapi
|
||||
|
||||
const exportedVdis = this._jobSnapshotVdis.filter(
|
||||
_ => EXPORTED_SUCCESSFULLY in _.other_config && DATETIME in _.other_config
|
||||
)
|
||||
const grouped = this._groupSnapshotVdisByDatetime(exportedVdis)
|
||||
|
||||
this._baseVdis = {}
|
||||
if (grouped.length === 0) {
|
||||
debug('no base VDIs found', {
|
||||
jobLength: this._jobSnapshotVdis.length,
|
||||
exportedLength: exportedVdis.length,
|
||||
})
|
||||
}
|
||||
const srcVdis = keyBy(await xapi.getRecords('VDI', await this._vm.$getDisks()), '$ref')
|
||||
let lastExportedVdis = []
|
||||
let presentBaseVdis = new Map()
|
||||
let baseUuidToSrcVdiUuid = new Map()
|
||||
let deltaChainLength
|
||||
|
||||
if (grouped.length > 0) {
|
||||
lastExportedVdis = grouped[0].vdis
|
||||
// only compute it on the _jobSnapshotVdis , not on the reused snapshot from CONTENT_KEYS
|
||||
deltaChainLength = Math.max(
|
||||
...lastExportedVdis.map(({ other_config }) => Number(other_config[DELTA_CHAIN_LENGTH] ?? 0))
|
||||
)
|
||||
;({ presentBaseVdis, baseUuidToSrcVdiUuid } = await this._checkBaseVdis(lastExportedVdis, srcVdis))
|
||||
}
|
||||
|
||||
// if we don't find any candidates
|
||||
// if there is only one writer of type IncrementalXapiWriter
|
||||
// allow the reuse of any matching snapshot of this VM even from another job
|
||||
if (
|
||||
presentBaseVdis.size === 0 &&
|
||||
this._writers.size === 1 &&
|
||||
[...this._writers][0] instanceof IncrementalXapiWriter
|
||||
) {
|
||||
debug('fallback to content key')
|
||||
const candidates = {}
|
||||
for (const srcVdi of Object.values(srcVdis)) {
|
||||
const snapshots = srcVdi.$snapshots
|
||||
debug(`${srcVdi.uuid} got ${snapshots.length} snapshots`)
|
||||
for (const snapshot of snapshots) {
|
||||
if (DATETIME in snapshot.other_config) {
|
||||
candidates[snapshot.uuid] = snapshot
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
debug('got candidates ', Object.keys(candidates).length)
|
||||
this._filterValidSnapshotVdis(candidates)
|
||||
debug('got filtered ', Object.keys(candidates).length)
|
||||
for (const group of this._groupSnapshotVdisByDatetime(Object.values(candidates))) {
|
||||
debug('check by datetime ', [...group.vdis].length)
|
||||
debug(group.vdis)
|
||||
const result = await this._checkBaseVdis(group.vdis, srcVdis, { useContentKey: true })
|
||||
debug('result ', result)
|
||||
if (result.presentBaseVdis.size > 0) {
|
||||
lastExportedVdis = group.vdis
|
||||
|
||||
presentBaseVdis = result.presentBaseVdis
|
||||
baseUuidToSrcVdiUuid = result.baseUuidToSrcVdiUuid
|
||||
|
||||
break
|
||||
}
|
||||
debug('try next group if any')
|
||||
}
|
||||
} else {
|
||||
debug(' no fallback ')
|
||||
}
|
||||
|
||||
if (presentBaseVdis.size === 0) {
|
||||
debug('no base VM found')
|
||||
|
||||
@@ -236,28 +236,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
// ensure they are attached to only one vm snapshot
|
||||
// ensure any VM-snapshot harvested by this has all its disk harvested (no mix of vdi snapshot from this job and not)
|
||||
|
||||
async _fetchJobSnapshots() {
|
||||
const jobId = this._jobId
|
||||
const xapi = this._xapi
|
||||
|
||||
const vdiCandidates = {}
|
||||
const vdiUuids = this._vm.$VBDs.map(({ VDI }) => VDI)
|
||||
Object.values(xapi.objects.indexes.type.VDI)
|
||||
.filter(_ => !!_) // filter nullish
|
||||
.filter(({ other_config, snapshot_of }) => {
|
||||
return (
|
||||
vdiUuids.includes(snapshot_of) &&
|
||||
other_config[JOB_ID] === jobId &&
|
||||
other_config[VM_UUID] === this._vm.uuid &&
|
||||
other_config[COPY_OF] === undefined
|
||||
)
|
||||
})
|
||||
.forEach(vdi => {
|
||||
vdiCandidates[vdi.uuid] = vdi
|
||||
})
|
||||
|
||||
// check that user snapshots are clean
|
||||
|
||||
_filterValidSnapshotVdis(vdiCandidates) {
|
||||
for (const vdi of Object.values(vdiCandidates)) {
|
||||
// cbt metadata are always considered linked to a backup job
|
||||
// if they have the right other_config
|
||||
@@ -279,7 +258,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
const userVms = vbds.map(({ $VM }) => $VM)
|
||||
if (vbds.length > 1) {
|
||||
warn(
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is linked to multipe vms : ${userVms.map(({ name_label, uuid }) => `${name_label} ${uuid}`).join(', ')}.
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is linked to multipe vms : ${userVms.map(({ name_label, uuid }) => `${name_label} ${uuid}`).join(', ')}.
|
||||
This disk snapshot will be excluded from the backup cleaning`,
|
||||
{ vdi, userVms }
|
||||
)
|
||||
@@ -294,7 +273,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
// => we exclude these from the backup processing
|
||||
if (vm.$snapshot_of === undefined) {
|
||||
warn(
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is a snapshot linked to a non snapshot vm ${vm.name_label} ${vm.uuid}.
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is a snapshot linked to a non snapshot vm ${vm.name_label} ${vm.uuid}.
|
||||
This disk snapshot will be excluded from the backup cleaning`,
|
||||
{ vdi, vm }
|
||||
)
|
||||
@@ -311,7 +290,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
vm.other_config[VM_UUID] !== vdi.other_config[VM_UUID]
|
||||
) {
|
||||
warn(
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is a snapshot linked to a snapshot vm ${vm.name_label} ${vm.uuid} out of this backup job scope.
|
||||
`vdi ${vdi.name_label} (${vdi.uuid}) is a snapshot linked to a snapshot vm ${vm.name_label} ${vm.uuid} out of this backup job scope.
|
||||
This disk snapshot will be excluded from the backup cleaning`,
|
||||
{ vdi, vm }
|
||||
)
|
||||
@@ -328,7 +307,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
.forEach(({ $VDI: outOfSnapshotsVdi, ...other }) => {
|
||||
warn(
|
||||
`vdi ${vdi.name_label} ${vdi.uuid} is recognized as a snapshot of the backup job,
|
||||
linked to vm ${vm.name_label} ${vm.uuid} but vdi ${outOfSnapshotsVdi.name_label} ${outOfSnapshotsVdi.uuid}
|
||||
linked to vm ${vm.name_label} ${vm.uuid} but vdi ${outOfSnapshotsVdi.name_label} ${outOfSnapshotsVdi.uuid}
|
||||
is not linked to the job. This disk snapshot will be excluded from the backup cleaning`,
|
||||
{ vdi, vm, vbds, outOfSnapshotsVdi }
|
||||
)
|
||||
@@ -336,19 +315,44 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
delete vdiCandidates[vdi.uuid]
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async _fetchJobSnapshots() {
|
||||
const jobId = this._jobId
|
||||
const xapi = this._xapi
|
||||
|
||||
const vdiCandidates = {}
|
||||
const vdiUuids = this._vm.$VBDs.map(({ VDI }) => VDI)
|
||||
Object.values(xapi.objects.indexes.type.VDI)
|
||||
.filter(_ => !!_) // filter nullish
|
||||
.filter(({ other_config, snapshot_of }) => {
|
||||
return (
|
||||
vdiUuids.includes(snapshot_of) &&
|
||||
other_config[JOB_ID] === jobId &&
|
||||
other_config[VM_UUID] === this._vm.uuid &&
|
||||
other_config[COPY_OF] === undefined
|
||||
)
|
||||
})
|
||||
.forEach(vdi => {
|
||||
vdiCandidates[vdi.uuid] = vdi
|
||||
})
|
||||
|
||||
this._filterValidSnapshotVdis(vdiCandidates)
|
||||
|
||||
this._jobSnapshotVdis = Object.values(vdiCandidates)
|
||||
|
||||
// For VMs with no disks, retention must be tracked directly on VM snapshots
|
||||
// since there are no VDIs to anchor the discovery.
|
||||
if (vdiUuids.length === 0) {
|
||||
this._disklessJobSnapshotVms = this._vm.$snapshots.filter(Boolean).filter(
|
||||
({ other_config, $snapshot_of }) =>
|
||||
$snapshot_of !== undefined &&
|
||||
other_config[JOB_ID] === jobId &&
|
||||
other_config[VM_UUID] === this._vm.uuid &&
|
||||
other_config[COPY_OF] === undefined
|
||||
)
|
||||
this._disklessJobSnapshotVms = this._vm.$snapshots
|
||||
.filter(Boolean)
|
||||
.filter(
|
||||
({ other_config, $snapshot_of }) =>
|
||||
$snapshot_of !== undefined &&
|
||||
other_config[JOB_ID] === jobId &&
|
||||
other_config[VM_UUID] === this._vm.uuid &&
|
||||
other_config[COPY_OF] === undefined
|
||||
)
|
||||
} else {
|
||||
this._disklessJobSnapshotVms = []
|
||||
}
|
||||
@@ -447,7 +451,7 @@ export const AbstractXapi = class AbstractXapiVmBackupRunner extends Abstract {
|
||||
if (disklessVmSnapshots.length > 0) {
|
||||
const snapshotsPerSchedule = groupBy(disklessVmSnapshots, _ => _.other_config[SCHEDULE_ID])
|
||||
await asyncEach(Object.entries(snapshotsPerSchedule), async ([scheduleId, snapshots]) => {
|
||||
// we only have one snapshot per date time since it's at the VM level
|
||||
// we only have one snapshot per date time since it's at the VM level
|
||||
const snapshotPerDatetime = Object.fromEntries(snapshots.map(s => [s.other_config[DATETIME], s.$ref]))
|
||||
const datetimes = Object.keys(snapshotPerDatetime).sort()
|
||||
const settings = {
|
||||
|
||||
@@ -19,11 +19,14 @@ import {
|
||||
REPLICATED_TO_SR_UUID,
|
||||
DATETIME,
|
||||
VM_UUID,
|
||||
CONTENT_KEY,
|
||||
resetVmOtherConfig,
|
||||
} from '../../_otherConfig.mjs'
|
||||
import { formatFilenameDate } from '../../_filenameDate.mjs'
|
||||
import { XapiDiskSource } from '@xen-orchestra/xapi'
|
||||
import { asyncEach } from '@vates/async-each'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
import { VM_POWER_STATE } from '@vates/types'
|
||||
|
||||
const { debug } = createLogger('xo:backups:IncrementalXapiWriter')
|
||||
|
||||
@@ -32,7 +35,20 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
// Built by checkBaseVdis, consumed by #decorateVmMetadata to set baseVdi.
|
||||
#baseVdisBySourceUuid = new Map()
|
||||
|
||||
async checkBaseVdis(baseUuidToSrcVdi) {
|
||||
/**
|
||||
* Finds VDIs on the target SR that can serve as base for the next delta transfer.
|
||||
*
|
||||
* For each entry in `baseUuidToSrcVdi`, searches the target SR for a matching snapshot
|
||||
* using CONTENT_KEY when available, then falls back to COPY_OF matching for older snapshots.
|
||||
* Entries with no matching base found on the target SR are removed from `baseUuidToSrcVdi`.
|
||||
*
|
||||
* Side-effects: populates `#baseVdisBySourceUuid`; may set `_targetVmRef`.
|
||||
*
|
||||
* @param {Map<string, string>} baseUuidToSrcVdi - Source snapshot UUID → source active VDI UUID. Mutated in place.
|
||||
* @param {Map<string, string>} contentKeys - Source snapshot UUID → CONTENT_KEY value (empty for pre-CONTENT_KEY snapshots).
|
||||
* @returns {Promise<void>}
|
||||
*/
|
||||
async checkBaseVdis(baseUuidToSrcVdi, contentKeys) {
|
||||
const sr = this._sr
|
||||
this.#baseVdisBySourceUuid = new Map()
|
||||
|
||||
@@ -45,18 +61,45 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
// look for the same snapshot
|
||||
// ensure there are no data between the snapshot and the active disk
|
||||
|
||||
const snapshotCandidates = sr.$VDIs.filter(vdi => {
|
||||
return (
|
||||
vdi?.managed &&
|
||||
vdi?.is_a_snapshot &&
|
||||
vdi.other_config[JOB_ID] === this._job.id &&
|
||||
vdi.other_config[VM_UUID] === this._vmUuid &&
|
||||
baseUuidToSrcVdi.has(vdi?.other_config[COPY_OF])
|
||||
)
|
||||
})
|
||||
debug('checkBaseVdis, got snapshot candidates,', snapshotCandidates.length)
|
||||
const snapshotCandidates = new Map()
|
||||
|
||||
if (snapshotCandidates.length > 0) {
|
||||
for (const [baseUuid, srcVdiuid] of baseUuidToSrcVdi) {
|
||||
let target
|
||||
|
||||
const contentKey = contentKeys.get(baseUuid)
|
||||
|
||||
if (contentKey !== undefined) {
|
||||
debug('got one content key, look for a candidate')
|
||||
target = sr.$VDIs.find(
|
||||
vdi =>
|
||||
vdi?.managed &&
|
||||
vdi?.is_a_snapshot &&
|
||||
vdi.other_config[CONTENT_KEY] === contentKey && // &&
|
||||
// ensure we don't replicate on ourself or any of the vdi in the source chain
|
||||
vdi.$snapshot_of.uuid !== srcVdiuid
|
||||
)
|
||||
}
|
||||
|
||||
// fall back for older snapshots
|
||||
if (target === undefined) {
|
||||
debug('content key not here or not found , look by jobid ')
|
||||
target = sr.$VDIs.find(
|
||||
vdi =>
|
||||
vdi?.managed &&
|
||||
vdi?.is_a_snapshot &&
|
||||
vdi.other_config[JOB_ID] === this._job.id &&
|
||||
vdi.other_config[VM_UUID] === this._vmUuid &&
|
||||
vdi?.other_config[COPY_OF] === baseUuid
|
||||
)
|
||||
}
|
||||
if (target !== undefined) {
|
||||
snapshotCandidates.set(baseUuid, target)
|
||||
}
|
||||
}
|
||||
|
||||
debug('checkBaseVdis, got snapshot candidates,', snapshotCandidates.size)
|
||||
|
||||
if (snapshotCandidates.size > 0) {
|
||||
// reset before searching for candidates
|
||||
this.#baseVdisBySourceUuid = new Map()
|
||||
this._targetVmRef = undefined
|
||||
@@ -64,10 +107,12 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
for (const [sourceUuid, vdi] of baseVdisBySourceUuid) {
|
||||
this.#baseVdisBySourceUuid.set(sourceUuid, vdi)
|
||||
}
|
||||
debug(' this.#baseVdisBySourceUuid ', this.#baseVdisBySourceUuid.size)
|
||||
if (targetVmRef !== undefined) {
|
||||
this._targetVmRef = targetVmRef
|
||||
}
|
||||
} else {
|
||||
debug('legacy fallback ( no content key ) ')
|
||||
// Legacy fallback (upgrade from pre-6.3): no target snapshots exist yet,
|
||||
// look for active (non-snapshot) VDIs with matching COPY_OF, like the old code did.
|
||||
debug('checkBaseVdis, no snapshot candidates, falling back to legacy active VDI lookup')
|
||||
@@ -92,6 +137,9 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
* 6.3+ snapshot-based validation: for each snapshot candidate, check whether
|
||||
* the active VDI has diverged from the snapshot. Returns a baseVdisBySourceUuid
|
||||
* map and, when all disks are clean, the targetVmRef to reuse.
|
||||
*
|
||||
* @param {Map<XenApiVdi['id'], import('@vates/types').XenApiVdi>} snapshotCandidates - Snapshot VDIs on the target SR to validate.
|
||||
* @returns {Promise<{ baseVdisBySourceUuid: Map<string, import('@vates/types').XenApiVdi>, targetVmRef: import('@vates/types').XenApiVm['$ref'] | undefined }>}
|
||||
*/
|
||||
async #validateSnapshotCandidates(snapshotCandidates) {
|
||||
const sr = this._sr
|
||||
@@ -100,24 +148,36 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
let canChainToTargetVm = true
|
||||
|
||||
await asyncEach(
|
||||
snapshotCandidates,
|
||||
async snapshot => {
|
||||
snapshotCandidates.entries(),
|
||||
async ([sourceUuid, snapshot]) => {
|
||||
let diffDisk
|
||||
let activeVdi
|
||||
try {
|
||||
activeVdi = sr.$xapi.getObject(snapshot.$snapshot_of)
|
||||
const userVbds = activeVdi.$VBDs?.filter(vbd => vbd.$VM && !vbd.$VM.is_control_domain) ?? []
|
||||
if (userVbds.length !== 1) {
|
||||
canChainToTargetVm = false
|
||||
debug('checkBaseVdis, shared vbd ', { ref: snapshot.$ref, userVbds })
|
||||
// shared vdi ignore
|
||||
// shared vdi ignore / don't chain
|
||||
return
|
||||
}
|
||||
const vm = userVbds[0].$VM
|
||||
if (!('start' in vm.blocked_operations)) {
|
||||
debug('checkBaseVdis, vm not blocked', { vmRef: vm.$ref })
|
||||
// vm start unlocked
|
||||
// not really an issue since we have check the delta
|
||||
// but it indicates the users played with the blocked operations
|
||||
|
||||
// a running VM will fail to compute disk exports
|
||||
// also a running VM can be assumed to have changed data
|
||||
if (vm.power_state !== VM_POWER_STATE.HALTED && vm.power_state !== VM_POWER_STATE.SUSPENDED) {
|
||||
canChainToTargetVm = false
|
||||
debug('checkBaseVdis, target vm is not halted or suspended', {
|
||||
ref: snapshot.$ref,
|
||||
userVbds,
|
||||
powerState: vm.power_state,
|
||||
})
|
||||
}
|
||||
// from this disk of from another
|
||||
// skip the costly part, only do the disk chaining
|
||||
if (!canChainToTargetVm) {
|
||||
debug("Can't chain VM anyway , fast return and chain with snapshot")
|
||||
baseVdisBySourceUuid.set(sourceUuid, snapshot)
|
||||
return
|
||||
}
|
||||
diffDisk = new XapiDiskSource({
|
||||
@@ -127,18 +187,16 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
onlyListChangedBlocks: true,
|
||||
})
|
||||
await diffDisk.init()
|
||||
const sourceUuid = snapshot.other_config?.[COPY_OF]
|
||||
if (diffDisk.getBlockIndexes().length === 0) {
|
||||
debug(' NO CHANGE , source detected ? ', !!sourceUuid)
|
||||
// no block modification since the common snapshot, we can chain VM and disk
|
||||
if (sourceUuid) {
|
||||
baseVdisBySourceUuid.set(sourceUuid, activeVdi)
|
||||
}
|
||||
// the disk is chained with the active to keep the chain linear
|
||||
baseVdisBySourceUuid.set(sourceUuid, activeVdi)
|
||||
// Track the target VM (the replicated VM to update on the next transfer).
|
||||
targetVmRef = vm.$ref
|
||||
} else {
|
||||
if (sourceUuid) {
|
||||
baseVdisBySourceUuid.set(sourceUuid, snapshot)
|
||||
}
|
||||
debug(' GOT CHANGE, source detected ? ', !!sourceUuid)
|
||||
baseVdisBySourceUuid.set(sourceUuid, snapshot)
|
||||
// there are changed block since the snapshot
|
||||
// we can reuse it to transfer a delta, but we will
|
||||
// create a new VM
|
||||
@@ -165,6 +223,7 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
)
|
||||
|
||||
if (!canChainToTargetVm) {
|
||||
debug('checkBaseVdis,NOT a valid vm target')
|
||||
// if at least one disk has new data, create a new VM
|
||||
// instead of updating it
|
||||
targetVmRef = undefined
|
||||
@@ -172,6 +231,7 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
debug('checkBaseVdis,got a valid vm target', targetVmRef)
|
||||
}
|
||||
|
||||
debug('checkBaseVdis,base vdis found : ', baseVdisBySourceUuid.size)
|
||||
return { baseVdisBySourceUuid, targetVmRef }
|
||||
}
|
||||
|
||||
@@ -269,6 +329,10 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
}
|
||||
|
||||
Object.values(backup.vdis).forEach(vdi => {
|
||||
// setVmSnapshotContentKeys sets CONTENT_KEY = the snapshot VDI's own UUID, but
|
||||
// that XenAPI write may not be visible in the xapi cache yet when exportIncrementalVm
|
||||
// reads vdi.other_config. Derive the correct value directly instead of relying on cache.
|
||||
vdi.other_config[CONTENT_KEY] = vdi.uuid
|
||||
vdi.other_config[COPY_OF] = vdi.uuid
|
||||
vdi.other_config[JOB_ID] = job.id
|
||||
vdi.other_config[SCHEDULE_ID] = scheduleId
|
||||
@@ -327,6 +391,7 @@ export class IncrementalXapiWriter extends MixinXapiWriter(AbstractIncrementalWr
|
||||
await xapi.VM_snapshot(targetVmRef, {
|
||||
name_label: `${vm.name_label} - ${job.name} / ${schedule.name} ${formatFilenameDate(timestamp)}`,
|
||||
})
|
||||
await resetVmOtherConfig(xapi, targetVmRef)
|
||||
})
|
||||
|
||||
return {
|
||||
|
||||
@@ -30,6 +30,7 @@
|
||||
"@vates/nbd-client": "^3.4.0",
|
||||
"@vates/parse-duration": "^0.1.1",
|
||||
"@vates/task": "^0.7.0",
|
||||
"@vates/types": "^1.24.0",
|
||||
"@xen-orchestra/async-map": "^0.1.3",
|
||||
"@xen-orchestra/disk-transform": "^1.2.3",
|
||||
"@xen-orchestra/fs": "^4.8.0",
|
||||
|
||||
@@ -18,10 +18,6 @@ BACKUP_REPOSITORY_NAME=Test backup QA
|
||||
# Paths must contain 'test', 'qa', or 'tmp/xo' to prevent production data deletion
|
||||
BACKUP_REPOSITORY_PATH=/tmp/xo-test-backups
|
||||
|
||||
# Production Backup Path (used for backup repository creation, NOT for cleanup)
|
||||
# This path is NOT included in automatic cleanup to protect production data
|
||||
BACKUP_PATH=/var/lib/xo/backups
|
||||
|
||||
# ===========================================
|
||||
# MIRROR BACKUP TEST CONFIGURATION
|
||||
# ===========================================
|
||||
@@ -31,6 +27,14 @@ BACKUP_PATH=/var/lib/xo/backups
|
||||
MIRROR_DESTINATION_REPOSITORY_NAME=Test mirror QA
|
||||
MIRROR_DESTINATION_REPOSITORY_PATH=/tmp/xo-test-mirror
|
||||
|
||||
# ===========================================
|
||||
# REPLICATION TEST CONFIGURATION
|
||||
# ===========================================
|
||||
|
||||
# Destination SR for incremental replication tests (required)
|
||||
# Must be a different SR from where the source VMs live
|
||||
REPLICATION_DESTINATION_SR_ID= # Required: UUID of the destination Storage Repository
|
||||
|
||||
# ===========================================
|
||||
# VHD/XVA EXPORT TEST CONFIGURATION
|
||||
# ===========================================
|
||||
|
||||
@@ -317,8 +317,13 @@ export class BackupRequest extends AbstractRequest {
|
||||
schedule: scheduleId,
|
||||
})
|
||||
} catch (wsError) {
|
||||
log.warn('Run backup job failed', { error: wsError.message })
|
||||
throw wsError
|
||||
if (wsError.code === -32000 && wsError.message.includes('task has not started yet')) {
|
||||
// sometimes we get a task not started error
|
||||
await new Promise(resolve => setTimeout(resolve, 10_000))
|
||||
} else {
|
||||
console.error('❌ Run backup job failed:', wsError.message)
|
||||
throw wsError
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for the backup to complete and get the log
|
||||
|
||||
@@ -2,7 +2,7 @@ import { createLogger } from '@xen-orchestra/log'
|
||||
import fs from 'node:fs/promises'
|
||||
import { FilterBuilder } from '../FilterBuilder.js'
|
||||
import { AbstractRequest } from './abstract.js'
|
||||
import { assertNonEmptyString } from '../../utils/index.js'
|
||||
import { assertNonEmptyString, waitUntil } from '../../utils/index.js'
|
||||
|
||||
const log = createLogger('xo:qa-test:vm')
|
||||
|
||||
@@ -101,6 +101,87 @@ export class VMRequest extends AbstractRequest {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts a virtual machine.
|
||||
*
|
||||
* Always uses JSON-RPC — the REST API does not expose the `force` parameter.
|
||||
*
|
||||
* @param {string} vmUuid - UUID of the VM to start
|
||||
* @param {Object} [options={}]
|
||||
* @param {boolean} [options.force=false] - Bypass any start-operation block set on the VM
|
||||
* @returns {Promise<void>}
|
||||
* @throws {Error} If the VM UUID is invalid or the start operation fails
|
||||
*/
|
||||
async start(vmUuid, options = {}) {
|
||||
if (!vmUuid || typeof vmUuid !== 'string') {
|
||||
throw new Error('Valid VM UUID is required')
|
||||
}
|
||||
|
||||
this._ensureConnected()
|
||||
|
||||
const { force = false } = options
|
||||
|
||||
try {
|
||||
await this.dispatchClient.xoClient.call('vm.start', { id: vmUuid, force })
|
||||
console.log(`VM started: ${vmUuid}`)
|
||||
} catch (error) {
|
||||
console.error(`❌ Failed to start VM ${vmUuid}:`, error.message)
|
||||
throw new Error(`Failed to start VM ${vmUuid}: ${error.message}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Stops a virtual machine.
|
||||
*
|
||||
* Always uses JSON-RPC — the REST API does not expose the `force` parameter.
|
||||
*
|
||||
* @param {string} vmUuid - UUID of the VM to stop
|
||||
* @param {Object} [options={}]
|
||||
* @param {boolean} [options.force=false] - Hard-stop (equivalent to power cut)
|
||||
* @returns {Promise<void>}
|
||||
* @throws {Error} If the VM UUID is invalid or the stop operation fails
|
||||
*/
|
||||
async stop(vmUuid, options = {}) {
|
||||
if (!vmUuid || typeof vmUuid !== 'string') {
|
||||
throw new Error('Valid VM UUID is required')
|
||||
}
|
||||
|
||||
this._ensureConnected()
|
||||
|
||||
const { force = false } = options
|
||||
|
||||
try {
|
||||
await this.dispatchClient.xoClient.call('vm.stop', { id: vmUuid, force })
|
||||
console.log(`VM stopped: ${vmUuid}`)
|
||||
} catch (error) {
|
||||
console.error(`❌ Failed to stop VM ${vmUuid}:`, error.message)
|
||||
throw new Error(`Failed to stop VM ${vmUuid}: ${error.message}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits for a VM to reach a specific power state.
|
||||
*
|
||||
* Polls the VM details until the expected power_state is observed or the
|
||||
* timeout is exceeded.
|
||||
*
|
||||
* @param {string} vmUuid - UUID of the VM to monitor
|
||||
* @param {'Running'|'Halted'|'Paused'|'Suspended'} targetState - Expected power state
|
||||
* @param {number} [timeout=60000] - Timeout in milliseconds
|
||||
* @returns {Promise<void>}
|
||||
* @throws {Error} If the VM does not reach the target state within the timeout
|
||||
*/
|
||||
async waitForPowerState(vmUuid, targetState, timeout = 60_000) {
|
||||
await waitUntil(
|
||||
async () => {
|
||||
const vmDetails = await this.details(vmUuid)
|
||||
return vmDetails?.power_state === targetState
|
||||
},
|
||||
2_000,
|
||||
timeout
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Deletes a virtual machine and its associated resources.
|
||||
*
|
||||
|
||||
@@ -33,6 +33,7 @@
|
||||
"qa:backup:combined:cycle": "node --env-file-if-exists=.env --test tests/backup-replication-combined.test.js --test-name-pattern='Full Cycle'",
|
||||
"qa:mirror:cr": "node --env-file-if-exists=.env --test tests/backup-replication-combined.test.js --test-name-pattern='CR Mode'",
|
||||
"qa:mirror": "node --env-file-if-exists=.env --test tests/backup-mirror.test.js",
|
||||
"qa:replication": "node --env-file-if-exists=.env --test --test-concurrency=1 tests/replication.test.js",
|
||||
"qa:export": "node --env-file-if-exists=.env --test tests/export.vhd.test.js",
|
||||
"qa:export:vhd": "node --env-file-if-exists=.env --test tests/export.vhd.test.js --test-name-pattern='VDI VHD'",
|
||||
"qa:export:xva": "node --env-file-if-exists=.env --test tests/export.vhd.test.js --test-name-pattern='XVA'",
|
||||
|
||||
@@ -0,0 +1,499 @@
|
||||
import assert from 'node:assert'
|
||||
import { after, before, describe, it } from 'node:test'
|
||||
import { createLogger } from '@xen-orchestra/log'
|
||||
|
||||
import {
|
||||
assertFullOrDeltaForSr,
|
||||
findTaskByMessage,
|
||||
generateBackupJobName,
|
||||
getBackupTransferredBytes,
|
||||
getDefaultSchedule,
|
||||
getScheduleKey,
|
||||
} from '../utils/index.js'
|
||||
import { assertBackupSuccess } from '../utils/backupUtils.js'
|
||||
import { setup, teardown } from './setup.js'
|
||||
|
||||
const log = createLogger('qa:replication')
|
||||
|
||||
/** Target SR for cross-SR replication */
|
||||
const REPLICATION_DESTINATION_SR_ID = process.env.REPLICATION_DESTINATION_SR_ID
|
||||
|
||||
describe('Incremental Replication', () => {
|
||||
/** @type {import('../client/dispatchClient.js').DispatchClient} */
|
||||
let dispatchClient
|
||||
/** @type {import('../utils/resourceTracker.js').ResourceTracker} */
|
||||
let tracker
|
||||
/** @type {{uuid: string, name_label: string, power_state: string}} */
|
||||
let vm
|
||||
/** @type {string} SR UUID where the test VM's disks live */
|
||||
let sourceVmSrUuid
|
||||
|
||||
before(async () => {
|
||||
if (!REPLICATION_DESTINATION_SR_ID) {
|
||||
throw new Error('REPLICATION_DESTINATION_SR_ID environment variable is required for replication tests')
|
||||
}
|
||||
|
||||
const setupResult = await setup()
|
||||
dispatchClient = setupResult.dispatchClient
|
||||
tracker = setupResult.tracker
|
||||
vm = setupResult.vm
|
||||
|
||||
assert.ok(vm, 'Setup should provide a test VM')
|
||||
|
||||
// Determine the SR where the test VM's disks live so we can run the
|
||||
// "same SR" scenario without hard-coding an extra env variable.
|
||||
const vdis = await dispatchClient.vdi.getVdisForVm(vm.uuid)
|
||||
assert.ok(vdis.length > 0, 'Test VM should have at least one VDI')
|
||||
sourceVmSrUuid = vdis[0].SR
|
||||
assert.ok(sourceVmSrUuid, 'Could not determine source SR from VM VDIs')
|
||||
|
||||
log.debug('Test VM', { name: vm.name_label, uuid: vm.uuid, sourceSrUuid: sourceVmSrUuid })
|
||||
|
||||
// Start the source VM so its disks are active during replication.
|
||||
// No need to wait for the OS to fully boot — the VM being in Running
|
||||
// state is enough to guarantee writes on the active VDI.
|
||||
await dispatchClient.vm.start(vm.uuid)
|
||||
log.debug('Source VM started (booting in background)', { uuid: vm.uuid })
|
||||
})
|
||||
|
||||
after(async () => {
|
||||
// Halt the source VM before teardown so the delete does not fail
|
||||
// on a running VM (started in before()).
|
||||
if (dispatchClient && vm) {
|
||||
try {
|
||||
const vmDetails = await dispatchClient.vm.details(vm.uuid)
|
||||
if (vmDetails?.power_state === 'Running') {
|
||||
await dispatchClient.vm.stop(vm.uuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(vm.uuid, 'Halted', 60_000)
|
||||
}
|
||||
} catch (error) {
|
||||
log.warn('Failed to stop source VM before teardown', { error })
|
||||
}
|
||||
}
|
||||
|
||||
if (dispatchClient && tracker) {
|
||||
await teardown(dispatchClient, tracker)
|
||||
}
|
||||
})
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Shared helpers — accessible by all nested describe blocks via closure
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Creates a delta replication job from sourceVm to targetSrUuid.
|
||||
* @param {{uuid: string, name_label: string}} sourceVm
|
||||
* @param {string} targetSrUuid
|
||||
* @returns {Promise<{jobId: string, scheduleKey: string}>}
|
||||
*/
|
||||
const createReplicationJob = async (sourceVm, targetSrUuid, baseName = '') => {
|
||||
const name = baseName + ' ' + generateBackupJobName()
|
||||
const schedule = getDefaultSchedule()
|
||||
const config = {
|
||||
name,
|
||||
mode: 'delta',
|
||||
schedules: { '': schedule },
|
||||
settings: { '': { timezone: 'Europe/Paris', copyRetention: 3, preferNbd: true, bypassVdiChainsCheck: true } },
|
||||
vms: { [sourceVm.uuid]: sourceVm },
|
||||
srs: { [targetSrUuid]: true },
|
||||
}
|
||||
|
||||
const jobId = await dispatchClient.backup.createBackupJob(config)
|
||||
const job = await dispatchClient.backup.details(jobId)
|
||||
assert.strictEqual(job.mode, 'delta', 'Replication job should be delta mode')
|
||||
|
||||
tracker.trackResource('backupJob', jobId, { name, mode: 'delta' })
|
||||
|
||||
const scheduleKey = getScheduleKey(job)
|
||||
assert.ok(scheduleKey, 'Schedule key is required')
|
||||
tracker.trackResource('schedule', scheduleKey, { name, backupJobId: jobId })
|
||||
|
||||
return { jobId, scheduleKey }
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns UUIDs of VMs that appeared since the given snapshot set.
|
||||
* @param {Set<string>} knownUuids
|
||||
* @returns {Promise<Array<string>>}
|
||||
*/
|
||||
const findNewVmUuids = async knownUuids => {
|
||||
const vmsNow = await dispatchClient.vm.list()
|
||||
return vmsNow.filter(v => !knownUuids.has(v.uuid)).map(v => v.uuid)
|
||||
}
|
||||
|
||||
/**
|
||||
* Force-stops and deletes a list of VMs, swallowing individual errors.
|
||||
* @param {Array<string>} vmUuids
|
||||
*/
|
||||
const cleanupVms = async vmUuids => {
|
||||
for (const vmUuid of vmUuids) {
|
||||
try {
|
||||
const vmDetails = await dispatchClient.vm.details(vmUuid)
|
||||
if (vmDetails?.power_state === 'Running') {
|
||||
await dispatchClient.vm.stop(vmUuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(vmUuid, 'Halted', 60_000)
|
||||
}
|
||||
await dispatchClient.vm.delete(vmUuid, { deleteDisks: true })
|
||||
log.debug('Cleaned up VM', { uuid: vmUuid })
|
||||
} catch (error) {
|
||||
log.warn('Failed to clean up VM', { uuid: vmUuid, error })
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Incremental replication — same SR and cross-SR
|
||||
// ===========================================================================
|
||||
|
||||
// Run every scenario against both "same SR" and "different SR" configurations.
|
||||
// The same-SR case is a common edge case where source and destination disks
|
||||
// live on the same storage repository.
|
||||
|
||||
for (const { label, getDestSrId } of [
|
||||
{ label: 'same SR (source = destination)', getDestSrId: () => sourceVmSrUuid },
|
||||
{ label: 'different SR', getDestSrId: () => REPLICATION_DESTINATION_SR_ID },
|
||||
]) {
|
||||
describe(label, () => {
|
||||
/** @type {{uuid: string, name_label: string}} */
|
||||
let destSr
|
||||
/** @type {Array<string>} */
|
||||
const replicatedVmUuids = []
|
||||
/** @type {string | null} Set when the destination SR is unavailable — causes tests to skip. */
|
||||
let destSrSkipReason = null
|
||||
|
||||
before(async () => {
|
||||
const destSrId = getDestSrId()
|
||||
if (!destSrId) {
|
||||
destSrSkipReason = 'REPLICATION_DESTINATION_SR_ID not configured'
|
||||
return
|
||||
}
|
||||
destSr = await dispatchClient.sr.details(destSrId)
|
||||
assert.ok(destSr, `Destination SR "${destSrId}" not found — check REPLICATION_DESTINATION_SR_ID in .env`)
|
||||
log.debug('Destination SR', { label, name: destSr.name_label, uuid: destSr.uuid })
|
||||
})
|
||||
|
||||
after(async () => cleanupVms(replicatedVmUuids))
|
||||
|
||||
/**
|
||||
* Returns user VDIs with other_config for the given VM UUID.
|
||||
* Works for both active VMs and snapshot VMs.
|
||||
*/
|
||||
const getVdisWithMeta = async vmUuid => {
|
||||
const raw = await dispatchClient.restApiClient.get(
|
||||
`/rest/v0/vms/${vmUuid}/vdis?fields=uuid,other_config,VDI_type`
|
||||
)
|
||||
return Array.isArray(raw) ? raw.filter(v => v.VDI_type === 'user') : []
|
||||
}
|
||||
|
||||
/**
|
||||
* Asserts CONTENT_KEY invariants after a replication run:
|
||||
* - source and destination snapshot VDIs share the same CONTENT_KEY value
|
||||
* - active VDIs on both sides carry no CONTENT_KEY
|
||||
*
|
||||
* @param {string} jobId
|
||||
* @param {string} replicatedVmUuid
|
||||
*/
|
||||
const assertContentKeyInvariants = async (jobId, replicatedVmUuid) => {
|
||||
const CONTENT_KEY_OC = 'xo:backup:contentKey'
|
||||
const COPY_OF_OC = 'xo:copy_of'
|
||||
|
||||
// After resetVmOtherConfig, active VDIs carry no backup metadata — COPY_OF and
|
||||
// CONTENT_KEY live exclusively on snapshot VDIs. Check each active VDI's snapshots.
|
||||
const destActiveVdis = await getVdisWithMeta(replicatedVmUuid)
|
||||
assert.ok(destActiveVdis.length > 0, 'Replicated VM must have active VDIs')
|
||||
|
||||
for (const destActiveVdi of destActiveVdis) {
|
||||
const raw = await dispatchClient.restApiClient.get(
|
||||
`/rest/v0/vdi-snapshots?${new URLSearchParams({ filter: `$snapshot_of:"${destActiveVdi.uuid}"`, fields: 'uuid,other_config' })}`
|
||||
)
|
||||
const destSnapshotVdis = Array.isArray(raw) ? raw : []
|
||||
assert.ok(
|
||||
destSnapshotVdis.length > 0,
|
||||
`Dest active VDI ${destActiveVdi.uuid} must have at least one snapshot VDI`
|
||||
)
|
||||
|
||||
for (const destSnapVdi of destSnapshotVdis) {
|
||||
const srcUuid = destSnapVdi.other_config?.[COPY_OF_OC]
|
||||
assert.ok(srcUuid, `Dest snapshot VDI ${destSnapVdi.uuid} must have ${COPY_OF_OC}`)
|
||||
|
||||
// Core invariant: CONTENT_KEY must be present and equal COPY_OF.
|
||||
// Both are set to the source snapshot VDI UUID during transfer.
|
||||
assert.strictEqual(
|
||||
destSnapVdi.other_config?.[CONTENT_KEY_OC],
|
||||
srcUuid,
|
||||
`Dest snapshot VDI ${destSnapVdi.uuid}: ${CONTENT_KEY_OC} must equal ${COPY_OF_OC} (${srcUuid})`
|
||||
)
|
||||
|
||||
// Source snapshot VDI should have CONTENT_KEY = its own UUID.
|
||||
// Skip gracefully if it was deleted by source retention on a subsequent run.
|
||||
let srcSnapVdi
|
||||
try {
|
||||
srcSnapVdi = await dispatchClient.restApiClient.get(
|
||||
`/rest/v0/vdi-snapshots/${srcUuid}?${new URLSearchParams({ fields: 'uuid,other_config' })}`
|
||||
)
|
||||
} catch {
|
||||
// Source snapshot was cleaned up by retention — skip the cross-side check
|
||||
continue
|
||||
}
|
||||
assert.strictEqual(
|
||||
srcSnapVdi.other_config?.[CONTENT_KEY_OC],
|
||||
srcUuid,
|
||||
`Source snapshot VDI ${srcUuid}: ${CONTENT_KEY_OC} must equal its own UUID`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// Active VDIs on both sides must NOT carry CONTENT_KEY
|
||||
for (const vdi of await getVdisWithMeta(vm.uuid)) {
|
||||
assert.strictEqual(
|
||||
vdi.other_config?.[CONTENT_KEY_OC],
|
||||
undefined,
|
||||
`Source active VDI ${vdi.uuid} must not have ${CONTENT_KEY_OC}`
|
||||
)
|
||||
}
|
||||
for (const vdi of destActiveVdis) {
|
||||
assert.strictEqual(
|
||||
vdi.other_config?.[CONTENT_KEY_OC],
|
||||
undefined,
|
||||
`Destination active VDI ${vdi.uuid} must not have ${CONTENT_KEY_OC}`
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Full replication lifecycle:
|
||||
// run 1 → full transfer, new VM created
|
||||
// run 2 → incremental transfer, same VM reused
|
||||
// run 3 → incremental transfer, new VM created (destination was started)
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
describe('Replication lifecycle: full → incremental → incremental after DR start', () => {
|
||||
it('should do full on first run, incremental on second, and incremental with new VM after destination started', async t => {
|
||||
if (destSrSkipReason) return t.skip(destSrSkipReason)
|
||||
const { jobId, scheduleKey } = await createReplicationJob(vm, destSr.uuid)
|
||||
const vmUuidsBefore = new Set((await dispatchClient.vm.list()).map(v => v.uuid))
|
||||
|
||||
// --- Run 1: full transfer, new VM created ---
|
||||
log.debug('Running first replication (expected full)', { label })
|
||||
const result1 = await dispatchClient.backup.runJobAndGetLog(jobId, scheduleKey)
|
||||
assertBackupSuccess(result1, 'First replication')
|
||||
assertFullOrDeltaForSr(result1, destSr.uuid, { mustBeFull: true })
|
||||
|
||||
const bytes1 = getBackupTransferredBytes(result1)
|
||||
log.debug('First run transferred (full)', { label, bytes: bytes1 })
|
||||
|
||||
const newUuids1 = await findNewVmUuids(vmUuidsBefore)
|
||||
assert.strictEqual(newUuids1.length, 1, 'First replication should create exactly one new VM')
|
||||
const replicatedVmUuid = newUuids1[0]
|
||||
assert.notStrictEqual(
|
||||
replicatedVmUuid,
|
||||
vm.uuid,
|
||||
`Replicated VM UUID must differ from source VM UUID (${vm.uuid})`
|
||||
)
|
||||
replicatedVmUuids.push(replicatedVmUuid)
|
||||
|
||||
const replicatedVm = await dispatchClient.vm.details(replicatedVmUuid)
|
||||
log.debug('Replicated VM created', { label, name: replicatedVm.name_label, uuid: replicatedVmUuid })
|
||||
|
||||
const snapshotTask1 = findTaskByMessage(result1, 'target snapshot')
|
||||
assert.ok(snapshotTask1, 'First replication should include a "target snapshot" task')
|
||||
assert.strictEqual(snapshotTask1.status, 'success', 'Target snapshot task should succeed')
|
||||
|
||||
const snapshotsAfterFirst = (await dispatchClient.vm.details(replicatedVmUuid)).snapshots?.length ?? 0
|
||||
assert.ok(
|
||||
snapshotsAfterFirst >= 1,
|
||||
`Replicated VM should have ≥1 snapshot after first run, got ${snapshotsAfterFirst}`
|
||||
)
|
||||
log.debug('Replicated VM snapshot count after first run', { label, snapshots: snapshotsAfterFirst })
|
||||
|
||||
log.debug('Checking CONTENT_KEY propagation after first run', { label })
|
||||
await assertContentKeyInvariants(jobId, replicatedVmUuid)
|
||||
log.debug('CONTENT_KEY invariants verified after first run', { label })
|
||||
|
||||
// --- Run 2: incremental transfer, same VM reused ---
|
||||
log.debug('Running second replication (expected incremental, same VM reused)', { label })
|
||||
const result2 = await dispatchClient.backup.runJobAndGetLog(jobId, scheduleKey)
|
||||
assertBackupSuccess(result2, 'Second replication')
|
||||
assertFullOrDeltaForSr(result2, destSr.uuid, { mustBeFull: false })
|
||||
|
||||
const bytes2 = getBackupTransferredBytes(result2)
|
||||
log.debug('Second run transferred (incremental)', { label, bytes: bytes2 })
|
||||
|
||||
if (bytes1 !== null && bytes2 !== null) {
|
||||
assert.ok(
|
||||
bytes2 <= bytes1,
|
||||
`Incremental transfer (${bytes2} bytes) should be ≤ full transfer (${bytes1} bytes)`
|
||||
)
|
||||
}
|
||||
|
||||
const newUuids2 = await findNewVmUuids(vmUuidsBefore)
|
||||
assert.strictEqual(
|
||||
newUuids2.length,
|
||||
1,
|
||||
`Second replication should NOT create a new VM — expected 1 total, got ${newUuids2.length}`
|
||||
)
|
||||
assert.strictEqual(
|
||||
newUuids2[0],
|
||||
replicatedVmUuid,
|
||||
`Second replication should reuse VM ${replicatedVmUuid}, got ${newUuids2[0]}`
|
||||
)
|
||||
|
||||
const snapshotsAfterSecond = (await dispatchClient.vm.details(replicatedVmUuid)).snapshots?.length ?? 0
|
||||
assert.ok(
|
||||
snapshotsAfterSecond > snapshotsAfterFirst,
|
||||
`Replicated VM should accumulate snapshots across runs (before: ${snapshotsAfterFirst}, after: ${snapshotsAfterSecond})`
|
||||
)
|
||||
log.debug('VM reused', {
|
||||
label,
|
||||
uuid: replicatedVmUuid,
|
||||
snapshotsBefore: snapshotsAfterFirst,
|
||||
snapshotsAfter: snapshotsAfterSecond,
|
||||
})
|
||||
|
||||
log.debug('Checking CONTENT_KEY propagation after second run', { label })
|
||||
await assertContentKeyInvariants(jobId, replicatedVmUuid)
|
||||
log.debug('CONTENT_KEY invariants verified after second run', { label })
|
||||
|
||||
// --- Run 3: destination VM started (DR site in use), incremental but new VM ---
|
||||
//
|
||||
// Starting and stopping the replicated VM dirtied its active disk relative to the
|
||||
// last replication snapshot. IncrementalXapiWriter detects changed blocks on the
|
||||
// destination's active VDI and cannot update it in place — a new VM is created.
|
||||
// CONTENT_KEY still matches a common snapshot, so the transfer stays incremental.
|
||||
log.debug('Starting replicated VM to simulate DR site in use', { label })
|
||||
await dispatchClient.vm.start(replicatedVmUuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(replicatedVmUuid, 'Running', 60_000)
|
||||
log.debug('Replicated VM is running', { label })
|
||||
await new Promise(resolve => setTimeout(resolve, 30_000))
|
||||
await dispatchClient.vm.stop(replicatedVmUuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(replicatedVmUuid, 'Halted', 60_000)
|
||||
|
||||
log.debug('Running third replication (delta transfer, but new VM expected)', { label })
|
||||
const result3 = await dispatchClient.backup.runJobAndGetLog(jobId, scheduleKey)
|
||||
assertBackupSuccess(result3, 'Third replication (after destination VM started)')
|
||||
assertFullOrDeltaForSr(result3, destSr.uuid, { mustBeFull: false })
|
||||
|
||||
const newUuids3 = await findNewVmUuids(vmUuidsBefore)
|
||||
assert.strictEqual(
|
||||
newUuids3.length,
|
||||
2,
|
||||
`Third replication should have created a second VM on destination (got ${newUuids3.length} new VMs total)`
|
||||
)
|
||||
const secondReplicaUuid = newUuids3.find(u => u !== replicatedVmUuid)
|
||||
assert.ok(secondReplicaUuid, 'A distinct second replica VM should exist on the destination')
|
||||
assert.notStrictEqual(
|
||||
secondReplicaUuid,
|
||||
vm.uuid,
|
||||
`Second replica UUID must differ from source VM UUID (${vm.uuid})`
|
||||
)
|
||||
replicatedVmUuids.push(secondReplicaUuid)
|
||||
|
||||
log.debug('New replica created', { label, newUuid: secondReplicaUuid, originalUuid: replicatedVmUuid })
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Bi-directional replication — planned switch to disaster site
|
||||
//
|
||||
// Scenario: prod → DR (Job A, one run), then DR → prod (Job B, first run).
|
||||
// Job B has never run before, but CONTENT_KEY matching lets it reuse the
|
||||
// common snapshot from Job A, making the reverse transfer incremental.
|
||||
// ===========================================================================
|
||||
|
||||
describe('Bi-directional — planned switch to disaster site', () => {
|
||||
/** @type {{uuid: string, name_label: string}} */
|
||||
let destSr
|
||||
/** @type {Array<string>} VMs created on destSr during this test */
|
||||
const replicatedVmUuids = []
|
||||
|
||||
before(async () => {
|
||||
destSr = await dispatchClient.sr.details(REPLICATION_DESTINATION_SR_ID)
|
||||
assert.ok(destSr, `Destination SR "${REPLICATION_DESTINATION_SR_ID}" not found`)
|
||||
log.debug('DR SR', { name: destSr.name_label, uuid: destSr.uuid })
|
||||
})
|
||||
|
||||
after(async () => cleanupVms(replicatedVmUuids))
|
||||
|
||||
it('should transfer delta from DR to prod and update the source VM in place', async () => {
|
||||
// The source VM must be halted so its active disk is clean relative to
|
||||
// the last replication snapshot — this lets Job B update it in place.
|
||||
const currentVmState = await dispatchClient.vm.details(vm.uuid)
|
||||
if (currentVmState.power_state === 'Running') {
|
||||
log.debug('Stopping source VM for planned-switch test')
|
||||
await dispatchClient.vm.stop(vm.uuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(vm.uuid, 'Halted', 60_000)
|
||||
}
|
||||
|
||||
// --- Job A: prod → DR (one run, full transfer) ---
|
||||
log.debug('Job A: replicating source VM to DR SR')
|
||||
const { jobId: jobAId, scheduleKey: scheduleKeyA } = await createReplicationJob(
|
||||
vm,
|
||||
destSr.uuid,
|
||||
'to disaster recovery site'
|
||||
)
|
||||
const vmUuidsBeforeA = new Set((await dispatchClient.vm.list()).map(v => v.uuid))
|
||||
|
||||
const resultA = await dispatchClient.backup.runJobAndGetLog(jobAId, scheduleKeyA)
|
||||
assertBackupSuccess(resultA, 'Job A (prod → DR)')
|
||||
assertFullOrDeltaForSr(resultA, destSr.uuid, { mustBeFull: true })
|
||||
|
||||
const newUuidsA = await findNewVmUuids(vmUuidsBeforeA)
|
||||
assert.strictEqual(newUuidsA.length, 1, 'Job A should create exactly one new VM on DR SR')
|
||||
const replicatedVmUuid = newUuidsA[0]
|
||||
assert.notStrictEqual(replicatedVmUuid, vm.uuid, 'Replicated VM UUID must differ from source VM UUID')
|
||||
replicatedVmUuids.push(replicatedVmUuid)
|
||||
|
||||
const replicatedVm = await dispatchClient.vm.details(replicatedVmUuid)
|
||||
log.debug('Replicated VM on DR SR', { name: replicatedVm.name_label, uuid: replicatedVmUuid })
|
||||
|
||||
// --- Job B: DR → prod ---
|
||||
log.debug('Job B: creating reverse replication job (DR → prod)')
|
||||
const { jobId: jobBId, scheduleKey: scheduleKeyB } = await createReplicationJob(
|
||||
replicatedVm,
|
||||
sourceVmSrUuid,
|
||||
'to PROD site'
|
||||
)
|
||||
|
||||
// Force-start the replicated VM to simulate the DR site being in active use.
|
||||
log.debug('Force-starting replicated VM (DR site now active)')
|
||||
await dispatchClient.vm.start(replicatedVmUuid, { force: true })
|
||||
await dispatchClient.vm.waitForPowerState(replicatedVmUuid, 'Running', 60_000)
|
||||
await new Promise(resolve => setTimeout(resolve, 30_000))
|
||||
log.debug('Replicated VM is running on DR SR')
|
||||
|
||||
const vmUuidsBeforeB = new Set((await dispatchClient.vm.list()).map(v => v.uuid))
|
||||
const snapshotsOnSourceBefore = (await dispatchClient.vm.details(vm.uuid)).snapshots?.length ?? 0
|
||||
|
||||
const resultB = await dispatchClient.backup.runJobAndGetLog(jobBId, scheduleKeyB)
|
||||
assertBackupSuccess(resultB, 'Job B (DR → prod)')
|
||||
|
||||
// Job B has never run before, but CONTENT_KEY matches the snapshot from
|
||||
// Job A → the transfer must be incremental, not a full copy.
|
||||
assertFullOrDeltaForSr(resultB, sourceVmSrUuid, { mustBeFull: false })
|
||||
|
||||
// The source VM's active disk was clean (halted, no writes since the
|
||||
// Job A snapshot) → Job B updates it in place, no new VM on the source SR.
|
||||
const newUuidsAfterB = await findNewVmUuids(vmUuidsBeforeB)
|
||||
const unexpectedNewVms = newUuidsAfterB.filter(u => u !== replicatedVmUuid)
|
||||
assert.strictEqual(
|
||||
unexpectedNewVms.length,
|
||||
0,
|
||||
`Job B should update the source VM in place, not create a new VM (got ${unexpectedNewVms.length} unexpected new VM(s))`
|
||||
)
|
||||
|
||||
// The source VM must carry a new snapshot written by Job B.
|
||||
const snapshotsOnSourceAfter = (await dispatchClient.vm.details(vm.uuid)).snapshots?.length ?? 0
|
||||
assert.ok(
|
||||
snapshotsOnSourceAfter > snapshotsOnSourceBefore,
|
||||
`Source VM should have gained a snapshot from Job B (before: ${snapshotsOnSourceBefore}, after: ${snapshotsOnSourceAfter})`
|
||||
)
|
||||
|
||||
log.debug('Planned switch completed — delta transfer confirmed, source VM updated in place', {
|
||||
snapshotsBefore: snapshotsOnSourceBefore,
|
||||
snapshotsAfter: snapshotsOnSourceAfter,
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -231,7 +231,9 @@ export class XapiDiskSource extends DiskPassthrough {
|
||||
readAhead.progressHandler = new XapiProgressHandler(xapi, `Exporting content of VDI ${label} through NBD+CBT`)
|
||||
return readAhead
|
||||
} catch (error) {
|
||||
info('Error in openNbdCBT', error)
|
||||
if (error.code !== 'CBT_DISABLED') {
|
||||
info('Error in openNbdCBT', error)
|
||||
}
|
||||
// init probaby failed, so nothing to close , but better safe than sorry
|
||||
await source?.close().catch(warn)
|
||||
|
||||
|
||||
@@ -25,6 +25,7 @@
|
||||
- [REST API] Expose `DELETE /rest/v0/srs/:id` (PR [#9464](https://github.com/vatesfr/xen-orchestra/pull/9464))
|
||||
- [xo-server-ipmi-sensors] Migrated the old ipmi feature to a new plugin and added user defined sensors (PR [#9724](https://github.com/vatesfr/xen-orchestra/pull/9724))
|
||||
- [MCP] Read `HTTP_PROXY`/`HTTPS_PROXY`/`NO_PROXY` so an internal XOA stays reachable when the AI assistant goes through a corporate proxy (PR [#9820](https://github.com/vatesfr/xen-orchestra/pull/9820))
|
||||
- [Backups] bidirectional replication, reuse snapshot when possible (PR [#9806](https://github.com/vatesfr/xen-orchestra/pull/9806))
|
||||
|
||||
### Bug fixes
|
||||
|
||||
|
||||
Reference in New Issue
Block a user