Compare commits

..

1 Commits

Author SHA1 Message Date
Julien Fontanet
d274c34b6b fix(xapi): support ISO 8601 datetime format
Fixes #6082
2022-01-13 15:23:59 +01:00
662 changed files with 11142 additions and 24057 deletions

View File

@@ -1,7 +1,5 @@
'use strict'
module.exports = {
extends: ['plugin:eslint-comments/recommended', 'plugin:n/recommended', 'standard', 'standard-jsx', 'prettier'],
extends: ['plugin:eslint-comments/recommended', 'standard', 'standard-jsx', 'prettier'],
globals: {
__DEV__: true,
$Dict: true,
@@ -17,40 +15,11 @@ module.exports = {
{
files: ['cli.{,c,m}js', '*-cli.{,c,m}js', '**/*cli*/**/*.{,c,m}js'],
rules: {
'n/no-process-exit': 'off',
'no-console': 'off',
},
},
{
files: ['*.mjs'],
parserOptions: {
sourceType: 'module',
},
},
{
files: ['*.spec.{,c,m}js'],
rules: {
'n/no-unsupported-features/node-builtins': [
'error',
{
version: '>=16',
},
],
'n/no-unsupported-features/es-syntax': [
'error',
{
version: '>=16',
},
],
},
},
],
parserOptions: {
ecmaVersion: 13,
sourceType: 'script',
},
rules: {
// disabled because XAPI objects are using camel case
camelcase: ['off'],
@@ -65,7 +34,5 @@ module.exports = {
'lines-between-class-members': 'off',
'no-console': ['error', { allow: ['warn', 'error'] }],
strict: 'error',
},
}

16
.flowconfig Normal file
View File

@@ -0,0 +1,16 @@
[ignore]
<PROJECT_ROOT>/node_modules/.*
[include]
[libs]
[lints]
[options]
esproposal.decorators=ignore
esproposal.optional_chaining=enable
include_warnings=true
module.use_strict=true
[strict]

View File

@@ -6,18 +6,6 @@ labels: 'status: triaging :triangular_flag_on_post:, type: bug :bug:'
assignees: ''
---
**XOA or XO from the sources?**
If XOA:
- which release channel? (`stable` vs `latest`)
- please consider creating a support ticket in [your dedicated support area](https://xen-orchestra.com/#!/member/support)
If XO from the sources:
- Don't forget to [read this first](https://xen-orchestra.com/docs/community.html)
- As well as follow [this guide](https://xen-orchestra.com/docs/community.html#report-a-bug)
**Describe the bug**
A clear and concise description of what the bug is.
@@ -35,7 +23,7 @@ A clear and concise description of what you expected to happen.
**Screenshots**
If applicable, add screenshots to help explain your problem.
**Environment (please provide the following information):**
**Desktop (please complete the following information):**
- Node: [e.g. 16.12.1]
- xo-server: [e.g. 5.82.3]

View File

@@ -4,6 +4,7 @@ about: Suggest an idea for this project
title: ''
labels: ''
assignees: ''
---
**Is your feature request related to a problem? Please describe.**

View File

@@ -1,13 +0,0 @@
name: CI
on: [push]
jobs:
build:
name: Test
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: satackey/action-docker-layer-caching@v0.0.11
# Ignore the failure of a step and avoid terminating the job.
continue-on-error: true
- run: docker-compose -f docker/docker-compose.dev.yml build
- run: docker-compose -f docker/docker-compose.dev.yml up

9
.gitignore vendored
View File

@@ -1,4 +1,5 @@
/_book/
/coverage/
/node_modules/
/lerna-debug.log
/lerna-debug.log.*
@@ -10,6 +11,10 @@
/packages/*/dist/
/packages/*/node_modules/
/@xen-orchestra/proxy/src/app/mixins/index.mjs
/packages/vhd-cli/src/commands/index.js
/packages/xen-api/examples/node_modules/
/packages/xen-api/plot.dat
@@ -30,7 +35,3 @@ pnpm-debug.log.*
yarn-error.log
yarn-error.log.*
.env
# code coverage
.nyc_output/
coverage/

23
.travis.yml Normal file
View File

@@ -0,0 +1,23 @@
language: node_js
node_js:
- 14
# Use containers.
# http://docs.travis-ci.com/user/workers/container-based-infrastructure/
sudo: false
addons:
apt:
packages:
- qemu-utils
- blktap-utils
- vmdk-stream-converter
before_install:
- curl -o- -L https://yarnpkg.com/install.sh | bash
- export PATH="$HOME/.yarn/bin:$PATH"
cache:
yarn: true
script:
- yarn run travis-tests

View File

@@ -32,7 +32,7 @@ Returns a promise wich rejects as soon as a call to `iteratee` throws or a promi
`opts` is an object that can contains the following options:
- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `10`. The value `0` means no concurrency limit.
- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `1`
- `signal`: an abort signal to stop the iteration
- `stopOnError`: wether to stop iteration of first error, or wait for all calls to finish and throw an `AggregateError`, defaults to `true`

View File

@@ -14,7 +14,7 @@ Returns a promise wich rejects as soon as a call to `iteratee` throws or a promi
`opts` is an object that can contains the following options:
- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `10`. The value `0` means no concurrency limit.
- `concurrency`: a number which indicates the maximum number of parallel call to `iteratee`, defaults to `1`
- `signal`: an abort signal to stop the iteration
- `stopOnError`: wether to stop iteration of first error, or wait for all calls to finish and throw an `AggregateError`, defaults to `true`

View File

@@ -9,16 +9,7 @@ class AggregateError extends Error {
}
}
/**
* @template Item
* @param {Iterable<Item>} iterable
* @param {(item: Item, index: number, iterable: Iterable<Item>) => Promise<void>} iteratee
* @returns {Promise<void>}
*/
exports.asyncEach = function asyncEach(iterable, iteratee, { concurrency = 10, signal, stopOnError = true } = {}) {
if (concurrency === 0) {
concurrency = Infinity
}
exports.asyncEach = function asyncEach(iterable, iteratee, { concurrency = 1, signal, stopOnError = true } = {}) {
return new Promise((resolve, reject) => {
const it = (iterable[Symbol.iterator] || iterable[Symbol.asyncIterator]).call(iterable)
const errors = []

View File

@@ -36,7 +36,7 @@ describe('asyncEach', () => {
it('works', async () => {
const iteratee = jest.fn(async () => {})
await asyncEach.call(thisArg, iterable, iteratee, { concurrency: 1 })
await asyncEach.call(thisArg, iterable, iteratee)
expect(iteratee.mock.instances).toEqual(Array.from(values, () => thisArg))
expect(iteratee.mock.calls).toEqual(Array.from(values, (value, index) => [value, index, iterable]))
@@ -66,7 +66,7 @@ describe('asyncEach', () => {
}
})
expect(await rejectionOf(asyncEach(iterable, iteratee, { concurrency: 1, stopOnError: true }))).toBe(error)
expect(await rejectionOf(asyncEach(iterable, iteratee, { stopOnError: true }))).toBe(error)
expect(iteratee).toHaveBeenCalledTimes(2)
})
@@ -91,9 +91,7 @@ describe('asyncEach', () => {
}
})
await expect(asyncEach(iterable, iteratee, { concurrency: 1, signal: ac.signal })).rejects.toThrow(
'asyncEach aborted'
)
await expect(asyncEach(iterable, iteratee, { signal: ac.signal })).rejects.toThrow('asyncEach aborted')
expect(iteratee).toHaveBeenCalledTimes(2)
})
})

View File

@@ -24,7 +24,7 @@
"url": "https://vates.fr"
},
"license": "ISC",
"version": "1.0.0",
"version": "0.1.0",
"engines": {
"node": ">=8.10"
},

View File

@@ -1,30 +0,0 @@
Node does not cache queries to `dns.lookup`, which can lead application doing a lot of connections to have perf issues and to saturate Node threads pool.
This library attempts to mitigate these problems by providing a version of this function with a version short cache, applied on both errors and results.
> Limitation: `verbatim: false` option is not supported.
It has exactly the same API as the native method and can be used directly:
```js
import { createCachedLookup } from '@vates/cached-dns.lookup'
const lookup = createCachedLookup()
lookup('example.net', { all: true, family: 0 }, (error, result) => {
if (error != null) {
return console.warn(error)
}
console.log(result)
})
```
Or it can be used to replace the native implementation and speed up the whole app:
```js
// assign our cached implementation to dns.lookup
const restore = createCachedLookup().patchGlobal()
// to restore the previous implementation
restore()
```

View File

@@ -1 +0,0 @@
../../scripts/npmignore

View File

@@ -1,63 +0,0 @@
<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
# @vates/cached-dns.lookup
[![Package Version](https://badgen.net/npm/v/@vates/cached-dns.lookup)](https://npmjs.org/package/@vates/cached-dns.lookup) ![License](https://badgen.net/npm/license/@vates/cached-dns.lookup) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/cached-dns.lookup)](https://bundlephobia.com/result?p=@vates/cached-dns.lookup) [![Node compatibility](https://badgen.net/npm/node/@vates/cached-dns.lookup)](https://npmjs.org/package/@vates/cached-dns.lookup)
> Cached implementation of dns.lookup
## Install
Installation of the [npm package](https://npmjs.org/package/@vates/cached-dns.lookup):
```
> npm install --save @vates/cached-dns.lookup
```
## Usage
Node does not cache queries to `dns.lookup`, which can lead application doing a lot of connections to have perf issues and to saturate Node threads pool.
This library attempts to mitigate these problems by providing a version of this function with a version short cache, applied on both errors and results.
> Limitation: `verbatim: false` option is not supported.
It has exactly the same API as the native method and can be used directly:
```js
import { createCachedLookup } from '@vates/cached-dns.lookup'
const lookup = createCachedLookup()
lookup('example.net', { all: true, family: 0 }, (error, result) => {
if (error != null) {
return console.warn(error)
}
console.log(result)
})
```
Or it can be used to replace the native implementation and speed up the whole app:
```js
// assign our cached implementation to dns.lookup
const restore = createCachedLookup().patchGlobal()
// to restore the previous implementation
restore()
```
## Contributions
Contributions are _very_ welcomed, either on the documentation or on
the code.
You may:
- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
you've encountered;
- fork and create a pull request.
## License
[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

View File

@@ -1,72 +0,0 @@
'use strict'
const assert = require('assert')
const dns = require('dns')
const LRU = require('lru-cache')
function reportResults(all, results, callback) {
if (all) {
callback(null, results)
} else {
const first = results[0]
callback(null, first.address, first.family)
}
}
exports.createCachedLookup = function createCachedLookup({ lookup = dns.lookup } = {}) {
const cache = new LRU({
max: 500,
// 1 minute: long enough to be effective, short enough so there is no need to bother with DNS TTLs
ttl: 60e3,
})
function cachedLookup(hostname, options, callback) {
let all = false
let family = 0
if (typeof options === 'function') {
callback = options
} else if (typeof options === 'number') {
family = options
} else if (options != null) {
assert.notStrictEqual(options.verbatim, false, 'not supported by this implementation')
;({ all = all, family = family } = options)
}
// cache by family option because there will be an error if there is no
// entries for the requestion family so we cannot easily cache all families
// and filter on reporting back
const key = hostname + '/' + family
const result = cache.get(key)
if (result !== undefined) {
setImmediate(reportResults, all, result, callback)
} else {
lookup(hostname, { all: true, family, verbatim: true }, function onLookup(error, results) {
// errors are not cached because this will delay recovery after DNS/network issues
//
// there are no reliable way to detect if the error is real or simply
// that there are no results for the requested hostname
//
// there should be much fewer errors than success, therefore it should
// not be a big deal to not cache them
if (error != null) {
return callback(error)
}
cache.set(key, results)
reportResults(all, results, callback)
})
}
}
cachedLookup.patchGlobal = function patchGlobal() {
const previous = dns.lookup
dns.lookup = cachedLookup
return function restoreGlobal() {
assert.strictEqual(dns.lookup, cachedLookup)
dns.lookup = previous
}
}
return cachedLookup
}

View File

@@ -1,32 +0,0 @@
{
"engines": {
"node": ">=8"
},
"dependencies": {
"lru-cache": "^7.0.4"
},
"private": false,
"name": "@vates/cached-dns.lookup",
"description": "Cached implementation of dns.lookup",
"keywords": [
"cache",
"dns",
"lookup"
],
"homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/cached-dns.lookup",
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
"repository": {
"directory": "@vates/cached-dns.lookup",
"type": "git",
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"author": {
"name": "Vates SAS",
"url": "https://vates.fr"
},
"license": "ISC",
"version": "1.0.0",
"scripts": {
"postversion": "npm publish --access public"
}
}

View File

@@ -1,5 +1,3 @@
'use strict'
exports.coalesceCalls = function (fn) {
let promise
const clean = () => {

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { coalesceCalls } = require('./')

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { compose } = require('./')

View File

@@ -31,19 +31,15 @@ class Foo {
}
```
### `decorateClass(class, map)`
### `decorateMethodsWith(class, map)`
Decorates a number of accessors and methods directly, without using the decorator syntax:
Decorates a number of methods directly, without using the decorator syntax:
```js
import { decorateClass } from '@vates/decorate-with'
import { decorateMethodsWith } from '@vates/decorate-with'
class Foo {
get bar() {
// body
}
set bar(value) {
bar() {
// body
}
@@ -52,28 +48,22 @@ class Foo {
}
}
decorateClass(Foo, {
// getter and/or setter
bar: {
// without arguments
get: lodash.memoize,
decorateMethodsWith(Foo, {
// without arguments
bar: lodash.curry,
// with arguments
set: [lodash.debounce, 150],
},
// method (with or without arguments)
baz: lodash.curry,
// with arguments
baz: [lodash.debounce, 150],
})
```
The decorated class is returned, so you can export it directly.
To apply multiple transforms to an accessor/method, you can either call `decorateClass` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
To apply multiple transforms to a method, you can either call `decorateMethodsWith` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
```js
decorateClass(Foo, {
baz: compose([
decorateMethodsWith(Foo, {
bar: compose([
[lodash.debounce, 150]
lodash.curry,
])
@@ -97,11 +87,7 @@ class Foo {
}
```
Because it's a normal function, it can also be used with `decorateClass`, with `compose` or even by itself.
### `decorateMethodsWith(class, map)`
> Deprecated alias for [`decorateClass(class, map)`](#decorateclassclass-map).
Because it's a normal function, it can also be used with `decorateMethodsWith`, with `compose` or even by itself.
## Contributions

View File

@@ -13,19 +13,15 @@ class Foo {
}
```
### `decorateClass(class, map)`
### `decorateMethodsWith(class, map)`
Decorates a number of accessors and methods directly, without using the decorator syntax:
Decorates a number of methods directly, without using the decorator syntax:
```js
import { decorateClass } from '@vates/decorate-with'
import { decorateMethodsWith } from '@vates/decorate-with'
class Foo {
get bar() {
// body
}
set bar(value) {
bar() {
// body
}
@@ -34,28 +30,22 @@ class Foo {
}
}
decorateClass(Foo, {
// getter and/or setter
bar: {
// without arguments
get: lodash.memoize,
decorateMethodsWith(Foo, {
// without arguments
bar: lodash.curry,
// with arguments
set: [lodash.debounce, 150],
},
// method (with or without arguments)
baz: lodash.curry,
// with arguments
baz: [lodash.debounce, 150],
})
```
The decorated class is returned, so you can export it directly.
To apply multiple transforms to an accessor/method, you can either call `decorateClass` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
To apply multiple transforms to a method, you can either call `decorateMethodsWith` multiple times or use [`@vates/compose`](https://www.npmjs.com/package/@vates/compose):
```js
decorateClass(Foo, {
baz: compose([
decorateMethodsWith(Foo, {
bar: compose([
[lodash.debounce, 150]
lodash.curry,
])
@@ -79,8 +69,4 @@ class Foo {
}
```
Because it's a normal function, it can also be used with `decorateClass`, with `compose` or even by itself.
### `decorateMethodsWith(class, map)`
> Deprecated alias for [`decorateClass(class, map)`](#decorateclassclass-map).
Because it's a normal function, it can also be used with `decorateMethodsWith`, with `compose` or even by itself.

View File

@@ -1,5 +1,3 @@
'use strict'
exports.decorateWith = function decorateWith(fn, ...args) {
return (target, name, descriptor) => ({
...descriptor,
@@ -9,27 +7,14 @@ exports.decorateWith = function decorateWith(fn, ...args) {
const { getOwnPropertyDescriptor, defineProperty } = Object
function applyDecorator(decorator, value) {
return typeof decorator === 'function' ? decorator(value) : decorator[0](value, ...decorator.slice(1))
}
exports.decorateClass = exports.decorateMethodsWith = function decorateClass(klass, map) {
exports.decorateMethodsWith = function decorateMethodsWith(klass, map) {
const { prototype } = klass
for (const name of Object.keys(map)) {
const decorator = map[name]
const descriptor = getOwnPropertyDescriptor(prototype, name)
if (typeof decorator === 'function' || Array.isArray(decorator)) {
descriptor.value = applyDecorator(decorator, descriptor.value)
} else {
const { get, set } = decorator
if (get !== undefined) {
descriptor.get = applyDecorator(get, descriptor.get)
}
if (set !== undefined) {
descriptor.set = applyDecorator(set, descriptor.set)
}
}
const { value } = descriptor
const decorator = map[name]
descriptor.value = typeof decorator === 'function' ? decorator(value) : decorator[0](value, ...decorator.slice(1))
defineProperty(prototype, name, descriptor)
}
return klass

View File

@@ -1,152 +0,0 @@
'use strict'
const assert = require('assert')
const { describe, it } = require('tap').mocha
const { decorateClass, decorateWith, decorateMethodsWith, perInstance } = require('./')
const identity = _ => _
describe('decorateWith', () => {
it('works', () => {
const expectedArgs = [Math.random(), Math.random()]
const expectedFn = Function.prototype
const newFn = () => {}
const decorator = decorateWith(function wrapper(fn, ...args) {
assert.deepStrictEqual(fn, expectedFn)
assert.deepStrictEqual(args, expectedArgs)
return newFn
}, ...expectedArgs)
const descriptor = {
configurable: true,
enumerable: false,
value: expectedFn,
writable: true,
}
assert.deepStrictEqual(decorator({}, 'foo', descriptor), {
...descriptor,
value: newFn,
})
})
})
describe('decorateClass', () => {
it('works', () => {
class C {
foo() {}
bar() {}
get baz() {}
// eslint-disable-next-line accessor-pairs
set qux(_) {}
}
const expectedArgs = [Math.random(), Math.random()]
const P = C.prototype
const descriptors = Object.getOwnPropertyDescriptors(P)
const newFoo = () => {}
const newBar = () => {}
const newGetBaz = () => {}
const newSetQux = _ => {}
decorateClass(C, {
foo(fn) {
assert.strictEqual(arguments.length, 1)
assert.strictEqual(fn, P.foo)
return newFoo
},
bar: [
function (fn, ...args) {
assert.strictEqual(fn, P.bar)
assert.deepStrictEqual(args, expectedArgs)
return newBar
},
...expectedArgs,
],
baz: {
get(fn) {
assert.strictEqual(arguments.length, 1)
assert.strictEqual(fn, descriptors.baz.get)
return newGetBaz
},
},
qux: {
set: [
function (fn, ...args) {
assert.strictEqual(fn, descriptors.qux.set)
assert.deepStrictEqual(args, expectedArgs)
return newSetQux
},
...expectedArgs,
],
},
})
const newDescriptors = Object.getOwnPropertyDescriptors(P)
assert.deepStrictEqual(newDescriptors.foo, { ...descriptors.foo, value: newFoo })
assert.deepStrictEqual(newDescriptors.bar, { ...descriptors.bar, value: newBar })
assert.deepStrictEqual(newDescriptors.baz, { ...descriptors.baz, get: newGetBaz })
assert.deepStrictEqual(newDescriptors.qux, { ...descriptors.qux, set: newSetQux })
})
it('throws if using an accessor decorator for a method', function () {
assert.throws(() =>
decorateClass(
class {
foo() {}
},
{ foo: { get: identity, set: identity } }
)
)
})
it('throws if using a method decorator for an accessor', function () {
assert.throws(() =>
decorateClass(
class {
get foo() {}
},
{ foo: identity }
)
)
})
})
it('decorateMethodsWith is an alias of decorateClass', function () {
assert.strictEqual(decorateMethodsWith, decorateClass)
})
describe('perInstance', () => {
it('works', () => {
let calls = 0
const expectedArgs = [Math.random(), Math.random()]
const expectedFn = Function.prototype
function wrapper(fn, ...args) {
assert.strictEqual(fn, expectedFn)
assert.deepStrictEqual(args, expectedArgs)
const i = ++calls
return () => i
}
const wrapped = perInstance(expectedFn, wrapper, ...expectedArgs)
// decorator is not called before decorated called
assert.strictEqual(calls, 0)
const o1 = {}
const o2 = {}
assert.strictEqual(wrapped.call(o1), 1)
// the same decorated function is returned for the same instance
assert.strictEqual(wrapped.call(o1), 1)
// a new decorated function is returned for another instance
assert.strictEqual(wrapped.call(o2), 2)
})
})

View File

@@ -20,15 +20,11 @@
"url": "https://vates.fr"
},
"license": "ISC",
"version": "2.0.0",
"version": "1.0.0",
"engines": {
"node": ">=8.10"
},
"scripts": {
"postversion": "npm publish --access public",
"test": "tap"
},
"devDependencies": {
"tap": "^16.0.1"
"postversion": "npm publish --access public"
}
}

View File

@@ -1,5 +1,3 @@
'use strict'
const { asyncMap } = require('@xen-orchestra/async-map')
const { createLogger } = require('@xen-orchestra/log')

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { createDebounceResource } = require('./debounceResource')

View File

@@ -1,5 +1,3 @@
'use strict'
const ensureArray = require('ensure-array')
const { MultiKeyMap } = require('@vates/multi-key-map')

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { deduped } = require('./deduped')

View File

@@ -1,50 +0,0 @@
> This library is compatible with Node's `EventEmitter` and web browsers' `EventTarget` APIs.
### API
```js
import { EventListenersManager } from '@vates/event-listeners-manager'
const events = new EventListenersManager(emitter)
// adding listeners
events.add('foo', onFoo).add('bar', onBar).on('baz', onBaz)
// removing a specific listener
events.remove('foo', onFoo)
// removing all listeners for a specific event
events.removeAll('foo')
// removing all listeners
events.removeAll()
```
### Typical use case
> Removing all listeners when no longer necessary.
Manually:
```js
const onFoo = () => {}
const onBar = () => {}
const onBaz = () => {}
emitter.on('foo', onFoo).on('bar', onBar).on('baz', onBaz)
// CODE LOGIC
emitter.off('foo', onFoo).off('bar', onBar).off('baz', onBaz)
```
With this library:
```js
const events = new EventListenersManager(emitter)
events.add('foo', () => {})).add('bar', () => {})).add('baz', () => {}))
// CODE LOGIC
events.removeAll()
```

View File

@@ -1 +0,0 @@
../../scripts/npmignore

View File

@@ -1,81 +0,0 @@
<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
# @vates/event-listeners-manager
[![Package Version](https://badgen.net/npm/v/@vates/event-listeners-manager)](https://npmjs.org/package/@vates/event-listeners-manager) ![License](https://badgen.net/npm/license/@vates/event-listeners-manager) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/event-listeners-manager)](https://bundlephobia.com/result?p=@vates/event-listeners-manager) [![Node compatibility](https://badgen.net/npm/node/@vates/event-listeners-manager)](https://npmjs.org/package/@vates/event-listeners-manager)
## Install
Installation of the [npm package](https://npmjs.org/package/@vates/event-listeners-manager):
```
> npm install --save @vates/event-listeners-manager
```
## Usage
> This library is compatible with Node's `EventEmitter` and web browsers' `EventTarget` APIs.
### API
```js
import { EventListenersManager } from '@vates/event-listeners-manager'
const events = new EventListenersManager(emitter)
// adding listeners
events.add('foo', onFoo).add('bar', onBar).on('baz', onBaz)
// removing a specific listener
events.remove('foo', onFoo)
// removing all listeners for a specific event
events.removeAll('foo')
// removing all listeners
events.removeAll()
```
### Typical use case
> Removing all listeners when no longer necessary.
Manually:
```js
const onFoo = () => {}
const onBar = () => {}
const onBaz = () => {}
emitter.on('foo', onFoo).on('bar', onBar).on('baz', onBaz)
// CODE LOGIC
emitter.off('foo', onFoo).off('bar', onBar).off('baz', onBaz)
```
With this library:
```js
const events = new EventListenersManager(emitter)
events.add('foo', () => {})).add('bar', () => {})).add('baz', () => {}))
// CODE LOGIC
events.removeAll()
```
## Contributions
Contributions are _very_ welcomed, either on the documentation or on
the code.
You may:
- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
you've encountered;
- fork and create a pull request.
## License
[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

View File

@@ -1,56 +0,0 @@
'use strict'
exports.EventListenersManager = class EventListenersManager {
constructor(emitter) {
this._listeners = new Map()
this._add = (emitter.addListener || emitter.addEventListener).bind(emitter)
this._remove = (emitter.removeListener || emitter.removeEventListener).bind(emitter)
}
add(type, listener) {
let listeners = this._listeners.get(type)
if (listeners === undefined) {
listeners = new Set()
this._listeners.set(type, listeners)
}
// don't add the same listener multiple times (allowed on Node.js)
if (!listeners.has(listener)) {
listeners.add(listener)
this._add(type, listener)
}
return this
}
remove(type, listener) {
const allListeners = this._listeners
const listeners = allListeners.get(type)
if (listeners !== undefined && listeners.delete(listener)) {
this._remove(type, listener)
if (listeners.size === 0) {
allListeners.delete(type)
}
}
return this
}
removeAll(type) {
const allListeners = this._listeners
const remove = this._remove
const types = type !== undefined ? [type] : allListeners.keys()
for (const type of types) {
const listeners = allListeners.get(type)
if (listeners !== undefined) {
allListeners.delete(type)
for (const listener of listeners) {
remove(type, listener)
}
}
}
return this
}
}

View File

@@ -1,67 +0,0 @@
'use strict'
const t = require('tap')
const { EventEmitter } = require('events')
const { EventListenersManager } = require('./')
const noop = Function.prototype
// function spy (impl = Function.prototype) {
// function spy() {
// spy.calls.push([Array.from(arguments), this])
// }
// spy.calls = []
// return spy
// }
function assertListeners(t, event, listeners) {
t.strictSame(t.context.ee.listeners(event), listeners)
}
t.beforeEach(function (t) {
t.context.ee = new EventEmitter()
t.context.em = new EventListenersManager(t.context.ee)
})
t.test('.add adds a listener', function (t) {
t.context.em.add('foo', noop)
assertListeners(t, 'foo', [noop])
t.end()
})
t.test('.add does not add a duplicate listener', function (t) {
t.context.em.add('foo', noop).add('foo', noop)
assertListeners(t, 'foo', [noop])
t.end()
})
t.test('.remove removes a listener', function (t) {
t.context.em.add('foo', noop).remove('foo', noop)
assertListeners(t, 'foo', [])
t.end()
})
t.test('.removeAll removes all listeners of a given type', function (t) {
t.context.em.add('foo', noop).add('bar', noop).removeAll('foo')
assertListeners(t, 'foo', [])
assertListeners(t, 'bar', [noop])
t.end()
})
t.test('.removeAll removes all listeners', function (t) {
t.context.em.add('foo', noop).add('bar', noop).removeAll()
assertListeners(t, 'foo', [])
assertListeners(t, 'bar', [])
t.end()
})

View File

@@ -1,46 +0,0 @@
{
"engines": {
"node": ">=6"
},
"private": false,
"name": "@vates/event-listeners-manager",
"descriptions": "Easy way to clean up event listeners",
"keywords": [
"add",
"addEventListener",
"addListener",
"browser",
"clear",
"DOM",
"emitter",
"event",
"EventEmitter",
"EventTarget",
"management",
"manager",
"node",
"remove",
"removeEventListener",
"removeListener"
],
"homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/event-listeners-manager",
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
"repository": {
"directory": "@vates/event-listeners-manager",
"type": "git",
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"author": {
"name": "Vates SAS",
"url": "https://vates.fr"
},
"license": "ISC",
"version": "1.0.1",
"scripts": {
"postversion": "npm publish --access public",
"test": "tap --branches=72"
},
"devDependencies": {
"tap": "^16.2.0"
}
}

View File

@@ -1,5 +1,3 @@
'use strict'
class Node {
constructor(value) {
this.children = new Map()

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { MultiKeyMap } = require('./')

View File

@@ -1,5 +1,3 @@
'use strict'
const ms = require('ms')
exports.parseDuration = value => {

View File

@@ -1,57 +0,0 @@
`undefined` predicates are ignored and `undefined` is returned if all predicates are `undefined`, this permits the most efficient composition:
```js
const compositePredicate = every(undefined, some(predicate2, undefined))
// ends up as
const compositePredicate = predicate2
```
Predicates can also be passed wrapped in an array:
```js
const compositePredicate = every([predicate1, some([predicate2, predicate3])])
```
`this` and all arguments are passed to the nested predicates.
### `every(predicates)`
> Returns a predicate that returns `true` iff every predicate returns `true`.
```js
const isBetween3And7 = every(
n => n >= 3,
n => n <= 7
)
isBetween3And10(0)
// → false
isBetween3And10(5)
// → true
isBetween3And10(10)
// → false
```
### `some(predicates)`
> Returns a predicate that returns `true` iff some predicate returns `true`.
```js
const isAliceOrBob = some(
name => name === 'Alice',
name => name === 'Bob'
)
isAliceOrBob('Alice')
// → true
isAliceOrBob('Bob')
// → true
isAliceOrBob('Oscar')
// → false
```

View File

@@ -1 +0,0 @@
../../scripts/npmignore

View File

@@ -1,90 +0,0 @@
<!-- DO NOT EDIT MANUALLY, THIS FILE HAS BEEN GENERATED -->
# @vates/predicates
[![Package Version](https://badgen.net/npm/v/@vates/predicates)](https://npmjs.org/package/@vates/predicates) ![License](https://badgen.net/npm/license/@vates/predicates) [![PackagePhobia](https://badgen.net/bundlephobia/minzip/@vates/predicates)](https://bundlephobia.com/result?p=@vates/predicates) [![Node compatibility](https://badgen.net/npm/node/@vates/predicates)](https://npmjs.org/package/@vates/predicates)
> Utilities to compose predicates
## Install
Installation of the [npm package](https://npmjs.org/package/@vates/predicates):
```
> npm install --save @vates/predicates
```
## Usage
`undefined` predicates are ignored and `undefined` is returned if all predicates are `undefined`, this permits the most efficient composition:
```js
const compositePredicate = every(undefined, some(predicate2, undefined))
// ends up as
const compositePredicate = predicate2
```
Predicates can also be passed wrapped in an array:
```js
const compositePredicate = every([predicate1, some([predicate2, predicate3])])
```
`this` and all arguments are passed to the nested predicates.
### `every(predicates)`
> Returns a predicate that returns `true` iff every predicate returns `true`.
```js
const isBetween3And7 = every(
n => n >= 3,
n => n <= 7
)
isBetween3And10(0)
// → false
isBetween3And10(5)
// → true
isBetween3And10(10)
// → false
```
### `some(predicates)`
> Returns a predicate that returns `true` iff some predicate returns `true`.
```js
const isAliceOrBob = some(
name => name === 'Alice',
name => name === 'Bob'
)
isAliceOrBob('Alice')
// → true
isAliceOrBob('Bob')
// → true
isAliceOrBob('Oscar')
// → false
```
## Contributions
Contributions are _very_ welcomed, either on the documentation or on
the code.
You may:
- report any [issue](https://github.com/vatesfr/xen-orchestra/issues)
you've encountered;
- fork and create a pull request.
## License
[ISC](https://spdx.org/licenses/ISC) © [Vates SAS](https://vates.fr)

View File

@@ -1,71 +0,0 @@
'use strict'
const {
isArray,
prototype: { filter },
} = Array
class InvalidPredicate extends TypeError {
constructor(value) {
super('not a valid predicate')
this.value = value
}
}
function isDefinedPredicate(value) {
if (value === undefined) {
return false
}
if (typeof value !== 'function') {
throw new InvalidPredicate(value)
}
return true
}
function handleArgs() {
let predicates
if (!(arguments.length === 1 && isArray((predicates = arguments[0])))) {
predicates = arguments
}
return filter.call(predicates, isDefinedPredicate)
}
exports.every = function every() {
const predicates = handleArgs.apply(this, arguments)
const n = predicates.length
if (n === 0) {
return
}
if (n === 1) {
return predicates[0]
}
return function everyPredicate() {
for (let i = 0; i < n; ++i) {
if (!predicates[i].apply(this, arguments)) {
return false
}
}
return true
}
}
exports.some = function some() {
const predicates = handleArgs.apply(this, arguments)
const n = predicates.length
if (n === 0) {
return
}
if (n === 1) {
return predicates[0]
}
return function somePredicate() {
for (let i = 0; i < n; ++i) {
if (predicates[i].apply(this, arguments)) {
return true
}
}
return false
}
}

View File

@@ -1,65 +0,0 @@
'use strict'
const assert = require('assert/strict')
const { describe, it } = require('tap').mocha
const { every, some } = require('./')
const T = () => true
const F = () => false
const testArgsHandling = fn => {
it('returns undefined if all predicates are undefined', () => {
assert.equal(fn(undefined), undefined)
assert.equal(fn([undefined]), undefined)
})
it('returns the predicate if only a single one is passed', () => {
assert.equal(fn(undefined, T), T)
assert.equal(fn([undefined, T]), T)
})
it('throws if it receives a non-predicate', () => {
const error = new TypeError('not a valid predicate')
error.value = 3
assert.throws(() => fn(3), error)
})
it('forwards this and arguments to predicates', () => {
const thisArg = 'qux'
const args = ['foo', 'bar', 'baz']
const predicate = function () {
assert.equal(this, thisArg)
assert.deepEqual(Array.from(arguments), args)
}
fn(predicate, predicate).apply(thisArg, args)
})
}
const runTests = (fn, truthTable) =>
it('works', () => {
truthTable.forEach(([result, ...predicates]) => {
assert.equal(fn(...predicates)(), result)
assert.equal(fn(predicates)(), result)
})
})
describe('every', () => {
testArgsHandling(every)
runTests(every, [
[true, T, T],
[false, T, F],
[false, F, T],
[false, F, F],
])
})
describe('some', () => {
testArgsHandling(some)
runTests(some, [
[true, T, T],
[true, T, F],
[true, F, T],
[false, F, F],
])
})

View File

@@ -1,40 +0,0 @@
{
"private": false,
"name": "@vates/predicates",
"description": "Utilities to compose predicates",
"keywords": [
"and",
"combine",
"compose",
"every",
"function",
"functions",
"or",
"predicate",
"predicates",
"some"
],
"homepage": "https://github.com/vatesfr/xen-orchestra/tree/master/@vates/predicates",
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
"repository": {
"directory": "@vates/predicates",
"type": "git",
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"author": {
"name": "Vates SAS",
"url": "https://vates.fr"
},
"license": "ISC",
"version": "1.0.0",
"engines": {
"node": ">=6"
},
"scripts": {
"postversion": "npm publish --access public",
"test": "tap"
},
"devDependencies": {
"tap": "^16.0.1"
}
}

View File

@@ -1,26 +0,0 @@
### `readChunk(stream, [size])`
- returns the next available chunk of data
- like `stream.read()`, a number of bytes can be specified
- returns with less data than expected if stream has ended
- returns `null` if the stream has ended and no data has been read
```js
import { readChunk } from '@vates/read-chunk'
;(async () => {
let chunk
while ((chunk = await readChunk(stream, 1024)) !== null) {
// do something with chunk
}
})()
```
### `readChunkStrict(stream, [size])`
Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
```js
import { readChunkStrict } from '@vates/read-chunk'
const chunk = await readChunkStrict(stream, 1024)
```

View File

@@ -16,12 +16,9 @@ Installation of the [npm package](https://npmjs.org/package/@vates/read-chunk):
## Usage
### `readChunk(stream, [size])`
- returns the next available chunk of data
- like `stream.read()`, a number of bytes can be specified
- returns with less data than expected if stream has ended
- returns `null` if the stream has ended and no data has been read
- returns `null` if the stream has ended
```js
import { readChunk } from '@vates/read-chunk'
@@ -33,16 +30,6 @@ import { readChunk } from '@vates/read-chunk'
})()
```
### `readChunkStrict(stream, [size])`
Similar behavior to `readChunk` but throws if the stream ended before the requested data could be read.
```js
import { readChunkStrict } from '@vates/read-chunk'
const chunk = await readChunkStrict(stream, 1024)
```
## Contributions
Contributions are _very_ welcomed, either on the documentation or on

View File

@@ -0,0 +1,13 @@
- returns the next available chunk of data
- like `stream.read()`, a number of bytes can be specified
- returns `null` if the stream has ended
```js
import { readChunk } from '@vates/read-chunk'
;(async () => {
let chunk
while ((chunk = await readChunk(stream, 1024)) !== null) {
// do something with chunk
}
})()
```

View File

@@ -1,5 +1,3 @@
'use strict'
const readChunk = (stream, size) =>
size === 0
? Promise.resolve(Buffer.alloc(0))
@@ -30,22 +28,3 @@ const readChunk = (stream, size) =>
onReadable()
})
exports.readChunk = readChunk
exports.readChunkStrict = async function readChunkStrict(stream, size) {
const chunk = await readChunk(stream, size)
if (chunk === null) {
throw new Error('stream has ended without data')
}
if (size !== undefined && chunk.length !== size) {
const error = new Error('stream has ended with not enough data')
Object.defineProperties(error, {
chunk: {
value: chunk,
},
})
throw error
}
return chunk
}

View File

@@ -1,10 +1,8 @@
'use strict'
/* eslint-env jest */
const { Readable } = require('stream')
const { readChunk, readChunkStrict } = require('./')
const { readChunk } = require('./')
const makeStream = it => Readable.from(it, { objectMode: false })
makeStream.obj = Readable.from
@@ -43,27 +41,3 @@ describe('readChunk', () => {
})
})
})
const rejectionOf = promise =>
promise.then(
value => {
throw value
},
error => error
)
describe('readChunkStrict', function () {
it('throws if stream is empty', async () => {
const error = await rejectionOf(readChunkStrict(makeStream([])))
expect(error).toBeInstanceOf(Error)
expect(error.message).toBe('stream has ended without data')
expect(error.chunk).toEqual(undefined)
})
it('throws if stream ends with not enough data', async () => {
const error = await rejectionOf(readChunkStrict(makeStream(['foo', 'bar']), 10))
expect(error).toBeInstanceOf(Error)
expect(error.message).toBe('stream has ended with not enough data')
expect(error.chunk).toEqual(Buffer.from('foobar'))
})
})

View File

@@ -19,7 +19,7 @@
"type": "git",
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"version": "1.0.0",
"version": "0.1.2",
"engines": {
"node": ">=8.10"
},

View File

@@ -1,7 +1,5 @@
#!/usr/bin/env node
'use strict'
const fs = require('fs')
const mapKeys = (object, iteratee) => {

View File

@@ -1,5 +1,3 @@
'use strict'
const wrapCall = (fn, arg, thisArg) => {
try {
return Promise.resolve(fn.call(thisArg, arg))

View File

@@ -1,5 +1,3 @@
'use strict'
/* eslint-env jest */
const { asyncMapSettled } = require('./')

View File

@@ -1,5 +1,3 @@
'use strict'
// type MaybePromise<T> = Promise<T> | T
//
// declare export function asyncMap<T1, T2>(

View File

@@ -0,0 +1 @@
module.exports = require('../../@xen-orchestra/babel-config')(require('./package.json'))

View File

@@ -0,0 +1 @@
../../scripts/babel-eslintrc.js

View File

@@ -9,14 +9,28 @@
},
"version": "0.2.0",
"engines": {
"node": ">=14"
"node": ">=10"
},
"main": "dist/",
"scripts": {
"build": "cross-env NODE_ENV=production babel --source-maps --out-dir=dist/ src/",
"dev": "cross-env NODE_ENV=development babel --watch --source-maps --out-dir=dist/ src/",
"postversion": "npm publish --access public",
"test": "tap --lines 67 --functions 92 --branches 52 --statements 67"
"prebuild": "rimraf dist/",
"predev": "yarn run prebuild",
"prepublishOnly": "yarn run build"
},
"devDependencies": {
"@babel/cli": "^7.7.4",
"@babel/core": "^7.7.4",
"@babel/plugin-proposal-decorators": "^7.8.0",
"@babel/plugin-proposal-nullish-coalescing-operator": "^7.8.0",
"@babel/preset-env": "^7.7.4",
"cross-env": "^7.0.2",
"rimraf": "^3.0.0"
},
"dependencies": {
"@vates/decorate-with": "^2.0.0",
"@vates/decorate-with": "^1.0.0",
"@xen-orchestra/log": "^0.3.0",
"golike-defer": "^0.5.1",
"object-hash": "^2.0.1"
@@ -26,8 +40,5 @@
"author": {
"name": "Vates SAS",
"url": "https://vates.fr"
},
"devDependencies": {
"tap": "^16.0.1"
}
}

View File

@@ -1,14 +1,12 @@
'use strict'
const assert = require('assert')
const hash = require('object-hash')
const { createLogger } = require('@xen-orchestra/log')
const { decorateClass } = require('@vates/decorate-with')
const { defer } = require('golike-defer')
import assert from 'assert'
import hash from 'object-hash'
import { createLogger } from '@xen-orchestra/log'
import { decorateWith } from '@vates/decorate-with'
import { defer } from 'golike-defer'
const log = createLogger('xo:audit-core')
exports.Storage = class Storage {
export class Storage {
constructor() {
this._lock = Promise.resolve()
}
@@ -31,7 +29,7 @@ const ID_TO_ALGORITHM = {
5: 'sha256',
}
class AlteredRecordError extends Error {
export class AlteredRecordError extends Error {
constructor(id, nValid, record) {
super('altered record')
@@ -40,9 +38,8 @@ class AlteredRecordError extends Error {
this.record = record
}
}
exports.AlteredRecordError = AlteredRecordError
class MissingRecordError extends Error {
export class MissingRecordError extends Error {
constructor(id, nValid) {
super('missing record')
@@ -50,10 +47,8 @@ class MissingRecordError extends Error {
this.nValid = nValid
}
}
exports.MissingRecordError = MissingRecordError
const NULL_ID = 'nullId'
exports.NULL_ID = NULL_ID
export const NULL_ID = 'nullId'
const HASH_ALGORITHM_ID = '5'
const createHash = (data, algorithmId = HASH_ALGORITHM_ID) =>
@@ -62,12 +57,13 @@ const createHash = (data, algorithmId = HASH_ALGORITHM_ID) =>
excludeKeys: key => key === 'id',
})}`
class AuditCore {
export class AuditCore {
constructor(storage) {
assert.notStrictEqual(storage, undefined)
this._storage = storage
}
@decorateWith(defer)
async add($defer, subject, event, data) {
const time = Date.now()
$defer(await this._storage.acquireLock())
@@ -152,6 +148,7 @@ class AuditCore {
}
}
@decorateWith(defer)
async deleteRangeAndRewrite($defer, newest, oldest) {
assert.notStrictEqual(newest, undefined)
assert.notStrictEqual(oldest, undefined)
@@ -192,9 +189,3 @@ class AuditCore {
}
}
}
exports.AuditCore = AuditCore
decorateClass(AuditCore, {
add: defer,
deleteRangeAndRewrite: defer,
})

View File

@@ -1,9 +1,6 @@
'use strict'
/* eslint-env jest */
const assert = require('assert/strict')
const { afterEach, describe, it } = require('tap').mocha
const { AlteredRecordError, AuditCore, MissingRecordError, NULL_ID, Storage } = require('.')
import { AlteredRecordError, AuditCore, MissingRecordError, NULL_ID, Storage } from '.'
const asyncIteratorToArray = async asyncIterator => {
const array = []
@@ -75,7 +72,7 @@ const auditCore = new AuditCore(db)
const storeAuditRecords = async () => {
await Promise.all(DATA.map(data => auditCore.add(...data)))
const records = await asyncIteratorToArray(auditCore.getFrom())
assert.equal(records.length, DATA.length)
expect(records.length).toBe(DATA.length)
return records
}
@@ -86,11 +83,10 @@ describe('auditCore', () => {
const [newestRecord, deletedRecord] = await storeAuditRecords()
const nValidRecords = await auditCore.checkIntegrity(NULL_ID, newestRecord.id)
assert.equal(nValidRecords, DATA.length)
expect(nValidRecords).toBe(DATA.length)
await db.del(deletedRecord.id)
await assert.rejects(
auditCore.checkIntegrity(NULL_ID, newestRecord.id),
await expect(auditCore.checkIntegrity(NULL_ID, newestRecord.id)).rejects.toEqual(
new MissingRecordError(deletedRecord.id, 1)
)
})
@@ -101,8 +97,7 @@ describe('auditCore', () => {
alteredRecord.event = ''
await db.put(alteredRecord)
await assert.rejects(
auditCore.checkIntegrity(NULL_ID, newestRecord.id),
await expect(auditCore.checkIntegrity(NULL_ID, newestRecord.id)).rejects.toEqual(
new AlteredRecordError(alteredRecord.id, 1, alteredRecord)
)
})
@@ -112,8 +107,8 @@ describe('auditCore', () => {
await auditCore.deleteFrom(secondRecord.id)
assert.equal(await db.get(firstRecord.id), undefined)
assert.equal(await db.get(secondRecord.id), undefined)
expect(await db.get(firstRecord.id)).toBe(undefined)
expect(await db.get(secondRecord.id)).toBe(undefined)
await auditCore.checkIntegrity(secondRecord.id, thirdRecord.id)
})

View File

@@ -10,7 +10,7 @@
"url": "https://github.com/vatesfr/xen-orchestra.git"
},
"engines": {
"node": ">=8.3"
"node": ">=6"
},
"license": "AGPL-3.0-or-later",
"author": {

View File

@@ -1,5 +1,3 @@
'use strict'
const getopts = require('getopts')
const { version } = require('./package.json')

View File

@@ -1,5 +1,3 @@
'use strict'
const { dirname } = require('path')
const fs = require('promise-toolbox/promisifyAll')(require('fs'))

View File

@@ -1,4 +1,4 @@
'use strict'
#!/usr/bin/env node
// -----------------------------------------------------------------------------
@@ -26,13 +26,7 @@ module.exports = async function main(args) {
await asyncMap(_, async vmDir => {
vmDir = resolve(vmDir)
try {
await adapter.cleanVm(vmDir, {
fixMetadata: fix,
remove,
merge,
logInfo: (...args) => console.log(...args),
logWarn: (...args) => console.warn(...args),
})
await adapter.cleanVm(vmDir, { fixMetadata: fix, remove, merge, onLog: (...args) => console.warn(...args) })
} catch (error) {
console.error('adapter.cleanVm', vmDir, error)
}

View File

@@ -1,5 +1,3 @@
'use strict'
const filenamify = require('filenamify')
const get = require('lodash/get')
const { asyncMap } = require('@xen-orchestra/async-map')

View File

@@ -1,5 +1,3 @@
'use strict'
const groupBy = require('lodash/groupBy')
const { asyncMap } = require('@xen-orchestra/async-map')
const { createHash } = require('crypto')

View File

@@ -1,7 +1,5 @@
#!/usr/bin/env node
'use strict'
require('./_composeCommands')({
'clean-vms': {
get main() {

View File

@@ -7,12 +7,12 @@
"bugs": "https://github.com/vatesfr/xen-orchestra/issues",
"dependencies": {
"@xen-orchestra/async-map": "^0.1.2",
"@xen-orchestra/backups": "^0.27.4",
"@xen-orchestra/fs": "^3.0.0",
"@xen-orchestra/backups": "^0.18.3",
"@xen-orchestra/fs": "^0.19.3",
"filenamify": "^4.1.0",
"getopts": "^2.2.5",
"lodash": "^4.17.15",
"promise-toolbox": "^0.21.0"
"promise-toolbox": "^0.20.0"
},
"engines": {
"node": ">=7.10.1"
@@ -27,7 +27,7 @@
"scripts": {
"postversion": "npm publish --access public"
},
"version": "0.7.7",
"version": "0.6.1",
"license": "AGPL-3.0-or-later",
"author": {
"name": "Vates SAS",

View File

@@ -1,12 +1,10 @@
'use strict'
const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
const Disposable = require('promise-toolbox/Disposable')
const ignoreErrors = require('promise-toolbox/ignoreErrors')
const Disposable = require('promise-toolbox/Disposable.js')
const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
const { compileTemplate } = require('@xen-orchestra/template')
const { limitConcurrency } = require('limit-concurrency-decorator')
const { extractIdsFromSimplePattern } = require('./extractIdsFromSimplePattern.js')
const { extractIdsFromSimplePattern } = require('./_extractIdsFromSimplePattern.js')
const { PoolMetadataBackup } = require('./_PoolMetadataBackup.js')
const { Task } = require('./Task.js')
const { VmBackup } = require('./_VmBackup.js')
@@ -24,34 +22,6 @@ const getAdaptersByRemote = adapters => {
const runTask = (...args) => Task.run(...args).catch(noop) // errors are handled by logs
const DEFAULT_SETTINGS = {
reportWhen: 'failure',
}
const DEFAULT_VM_SETTINGS = {
bypassVdiChainsCheck: false,
checkpointSnapshot: false,
concurrency: 2,
copyRetention: 0,
deleteFirst: false,
exportRetention: 0,
fullInterval: 0,
healthCheckSr: undefined,
healthCheckVmsWithTags: [],
maxMergedDeltasPerRun: 2,
offlineBackup: false,
offlineSnapshot: false,
snapshotRetention: 0,
timeout: 0,
unconditionalSnapshot: false,
vmTimeout: 0,
}
const DEFAULT_METADATA_SETTINGS = {
retentionPoolMetadata: 0,
retentionXoMetadata: 0,
}
exports.Backup = class Backup {
constructor({ config, getAdapter, getConnectedRecord, job, schedule }) {
this._config = config
@@ -70,22 +40,17 @@ exports.Backup = class Backup {
'{job.name}': job.name,
'{vm.name_label}': vm => vm.name_label,
})
}
const { type } = job
const baseSettings = { ...DEFAULT_SETTINGS }
run() {
const type = this._job.type
if (type === 'backup') {
Object.assign(baseSettings, DEFAULT_VM_SETTINGS, config.defaultSettings, config.vm?.defaultSettings)
this.run = this._runVmBackup
return this._runVmBackup()
} else if (type === 'metadataBackup') {
Object.assign(baseSettings, DEFAULT_METADATA_SETTINGS, config.defaultSettings, config.metadata?.defaultSettings)
this.run = this._runMetadataBackup
return this._runMetadataBackup()
} else {
throw new Error(`No runner for the backup type ${type}`)
}
Object.assign(baseSettings, job.settings[''])
this._baseSettings = baseSettings
this._settings = { ...baseSettings, ...job.settings[schedule.id] }
}
async _runMetadataBackup() {
@@ -97,6 +62,13 @@ exports.Backup = class Backup {
}
const config = this._config
const settings = {
...config.defaultSettings,
...config.metadata.defaultSettings,
...job.settings[''],
...job.settings[schedule.id],
}
const poolIds = extractIdsFromSimplePattern(job.pools)
const isEmptyPools = poolIds.length === 0
const isXoMetadata = job.xoMetadata !== undefined
@@ -104,8 +76,6 @@ exports.Backup = class Backup {
throw new Error('no metadata mode found')
}
const settings = this._settings
const { retentionPoolMetadata, retentionXoMetadata } = settings
if (
@@ -217,7 +187,14 @@ exports.Backup = class Backup {
const schedule = this._schedule
const config = this._config
const settings = this._settings
const { settings } = job
const scheduleSettings = {
...config.defaultSettings,
...config.vm.defaultSettings,
...settings[''],
...settings[schedule.id],
}
await Disposable.use(
Disposable.all(
extractIdsFromSimplePattern(job.srs).map(id =>
@@ -245,15 +222,14 @@ exports.Backup = class Backup {
})
)
),
() => (settings.healthCheckSr !== undefined ? this._getRecord('SR', settings.healthCheckSr) : undefined),
async (srs, remoteAdapters, healthCheckSr) => {
async (srs, remoteAdapters) => {
// remove adapters that failed (already handled)
remoteAdapters = remoteAdapters.filter(_ => _ !== undefined)
// remove srs that failed (already handled)
srs = srs.filter(_ => _ !== undefined)
if (remoteAdapters.length === 0 && srs.length === 0 && settings.snapshotRetention === 0) {
if (remoteAdapters.length === 0 && srs.length === 0 && scheduleSettings.snapshotRetention === 0) {
return
}
@@ -263,27 +239,23 @@ exports.Backup = class Backup {
remoteAdapters = getAdaptersByRemote(remoteAdapters)
const allSettings = this._job.settings
const baseSettings = this._baseSettings
const handleVm = vmUuid =>
runTask({ name: 'backup VM', data: { type: 'VM', id: vmUuid } }, () =>
Disposable.use(this._getRecord('VM', vmUuid), vm =>
new VmBackup({
baseSettings,
config,
getSnapshotNameLabel,
healthCheckSr,
job,
// remotes,
remoteAdapters,
schedule,
settings: { ...settings, ...allSettings[vm.uuid] },
settings: { ...scheduleSettings, ...settings[vmUuid] },
srs,
vm,
}).run()
)
)
const { concurrency } = settings
const { concurrency } = scheduleSettings
await asyncMapSettled(vmIds, concurrency === 0 ? handleVm : limitConcurrency(concurrency)(handleVm))
}
)

View File

@@ -1,5 +1,3 @@
'use strict'
const { asyncMap } = require('@xen-orchestra/async-map')
exports.DurablePartition = class DurablePartition {

View File

@@ -1,64 +0,0 @@
'use strict'
const { Task } = require('./Task')
exports.HealthCheckVmBackup = class HealthCheckVmBackup {
#xapi
#restoredVm
constructor({ restoredVm, xapi }) {
this.#restoredVm = restoredVm
this.#xapi = xapi
}
async run() {
return Task.run(
{
name: 'vmstart',
},
async () => {
let restoredVm = this.#restoredVm
const xapi = this.#xapi
const restoredId = restoredVm.uuid
// remove vifs
await Promise.all(restoredVm.$VIFs.map(vif => xapi.callAsync('VIF.destroy', vif.$ref)))
const start = new Date()
// start Vm
await xapi.callAsync(
'VM.start',
restoredVm.$ref,
false, // Start paused?
false // Skip pre-boot checks?
)
const started = new Date()
const timeout = 10 * 60 * 1000
const startDuration = started - start
let remainingTimeout = timeout - startDuration
if (remainingTimeout < 0) {
throw new Error(`VM ${restoredId} not started after ${timeout / 1000} second`)
}
// wait for the 'Running' event to be really stored in local xapi object cache
restoredVm = await xapi.waitObjectState(restoredVm.$ref, vm => vm.power_state === 'Running', {
timeout: remainingTimeout,
})
const running = new Date()
remainingTimeout -= running - started
if (remainingTimeout < 0) {
throw new Error(`local xapi did not get Runnig state for VM ${restoredId} after ${timeout / 1000} second`)
}
// wait for the guest tool version to be defined
await xapi.waitObjectState(restoredVm.guest_metrics, gm => gm?.PV_drivers_version?.major !== undefined, {
timeout: remainingTimeout,
})
}
)
}
}

View File

@@ -1,5 +1,3 @@
'use strict'
const assert = require('assert')
const { formatFilenameDate } = require('./_filenameDate.js')
@@ -8,9 +6,9 @@ const { Task } = require('./Task.js')
const { watchStreamSize } = require('./_watchStreamSize.js')
exports.ImportVmBackup = class ImportVmBackup {
constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses, mapVdisSrs = {} } = {} }) {
constructor({ adapter, metadata, srUuid, xapi, settings: { newMacAddresses } = {} }) {
this._adapter = adapter
this._importDeltaVmSettings = { newMacAddresses, mapVdisSrs }
this._importDeltaVmSettings = { newMacAddresses }
this._metadata = metadata
this._srUuid = srUuid
this._xapi = xapi
@@ -30,12 +28,7 @@ exports.ImportVmBackup = class ImportVmBackup {
} else {
assert.strictEqual(metadata.mode, 'delta')
const ignoredVdis = new Set(
Object.entries(this._importDeltaVmSettings.mapVdisSrs)
.filter(([_, srUuid]) => srUuid === null)
.map(([vdiUuid]) => vdiUuid)
)
backup = await adapter.readDeltaVmBackup(metadata, ignoredVdis)
backup = await adapter.readDeltaVmBackup(metadata)
Object.values(backup.streams).forEach(stream => watchStreamSize(stream, sizeContainer))
}

View File

@@ -1,24 +1,17 @@
'use strict'
const { asyncMap, asyncMapSettled } = require('@xen-orchestra/async-map')
const { synchronized } = require('decorator-synchronized')
const Disposable = require('promise-toolbox/Disposable')
const fromCallback = require('promise-toolbox/fromCallback')
const fromEvent = require('promise-toolbox/fromEvent')
const pDefer = require('promise-toolbox/defer')
const Disposable = require('promise-toolbox/Disposable.js')
const fromCallback = require('promise-toolbox/fromCallback.js')
const fromEvent = require('promise-toolbox/fromEvent.js')
const pDefer = require('promise-toolbox/defer.js')
const groupBy = require('lodash/groupBy.js')
const pickBy = require('lodash/pickBy.js')
const { dirname, join, normalize, resolve } = require('path')
const { createLogger } = require('@xen-orchestra/log')
const { createVhdDirectoryFromStream, openVhd, VhdAbstract, VhdDirectory, VhdSynthetic } = require('vhd-lib')
const { Constants, createVhdDirectoryFromStream, openVhd, VhdAbstract, VhdDirectory, VhdSynthetic } = require('vhd-lib')
const { deduped } = require('@vates/disposable/deduped.js')
const { decorateMethodsWith } = require('@vates/decorate-with')
const { compose } = require('@vates/compose')
const { execFile } = require('child_process')
const { readdir, lstat } = require('fs-extra')
const { readdir, stat } = require('fs-extra')
const { v4: uuidv4 } = require('uuid')
const { ZipFile } = require('yazl')
const zlib = require('zlib')
const { BACKUP_DIR } = require('./_getVmBackupDir.js')
const { cleanVm } = require('./_cleanVm.js')
@@ -47,12 +40,13 @@ const resolveSubpath = (root, path) => resolve(root, `.${resolve('/', path)}`)
const RE_VHDI = /^vhdi(\d+)$/
async function addDirectory(files, realPath, metadataPath) {
const stats = await lstat(realPath)
if (stats.isDirectory()) {
await asyncMap(await readdir(realPath), file =>
addDirectory(files, realPath + '/' + file, metadataPath + '/' + file)
)
} else if (stats.isFile()) {
try {
const subFiles = await readdir(realPath)
await asyncMap(subFiles, file => addDirectory(files, realPath + '/' + file, metadataPath + '/' + file))
} catch (error) {
if (error == null || error.code !== 'ENOTDIR') {
throw error
}
files.push({
realPath,
metadataPath,
@@ -79,7 +73,6 @@ class RemoteAdapter {
this._dirMode = dirMode
this._handler = handler
this._vhdDirectoryCompression = vhdDirectoryCompression
this._readCacheListVmBackups = synchronized.withKey()(this._readCacheListVmBackups)
}
get handler() {
@@ -95,6 +88,9 @@ class RemoteAdapter {
return partition
}
_getLvmLogicalVolumes = Disposable.factory(this._getLvmLogicalVolumes)
_getLvmLogicalVolumes = deduped(this._getLvmLogicalVolumes, (devicePath, pvId, vgName) => [devicePath, pvId, vgName])
_getLvmLogicalVolumes = debounceResourceFactory(this._getLvmLogicalVolumes)
async *_getLvmLogicalVolumes(devicePath, pvId, vgName) {
yield this._getLvmPhysicalVolume(devicePath, pvId && (await this._findPartition(devicePath, pvId)))
@@ -106,6 +102,9 @@ class RemoteAdapter {
}
}
_getLvmPhysicalVolume = Disposable.factory(this._getLvmPhysicalVolume)
_getLvmPhysicalVolume = deduped(this._getLvmPhysicalVolume, (devicePath, partition) => [devicePath, partition?.id])
_getLvmPhysicalVolume = debounceResourceFactory(this._getLvmPhysicalVolume)
async *_getLvmPhysicalVolume(devicePath, partition) {
const args = []
if (partition !== undefined) {
@@ -126,6 +125,9 @@ class RemoteAdapter {
}
}
_getPartition = Disposable.factory(this._getPartition)
_getPartition = deduped(this._getPartition, (devicePath, partition) => [devicePath, partition?.id])
_getPartition = debounceResourceFactory(this._getPartition)
async *_getPartition(devicePath, partition) {
const options = ['loop', 'ro']
@@ -178,6 +180,7 @@ class RemoteAdapter {
})
}
_usePartitionFiles = Disposable.factory(this._usePartitionFiles)
async *_usePartitionFiles(diskId, partitionId, paths) {
const path = yield this.getPartition(diskId, partitionId)
@@ -227,8 +230,8 @@ class RemoteAdapter {
async deleteDeltaVmBackups(backups) {
const handler = this._handler
// this will delete the json, unused VHDs will be detected by `cleanVm`
await asyncMapSettled(backups, ({ _filename }) => handler.unlink(_filename))
// unused VHDs will be detected by `cleanVm`
await asyncMapSettled(backups, ({ _filename }) => VhdAbstract.unlink(handler, _filename))
}
async deleteMetadataBackup(backupId) {
@@ -263,8 +266,7 @@ class RemoteAdapter {
}
async deleteVmBackups(files) {
const metadatas = await asyncMap(files, file => this.readVmBackupMetadata(file))
const { delta, full, ...others } = groupBy(metadatas, 'mode')
const { delta, full, ...others } = groupBy(await asyncMap(files, file => this.readVmBackupMetadata(file)), 'mode')
const unsupportedModes = Object.keys(others)
if (unsupportedModes.length !== 0) {
@@ -275,15 +277,6 @@ class RemoteAdapter {
delta !== undefined && this.deleteDeltaVmBackups(delta),
full !== undefined && this.deleteFullVmBackups(full),
])
const dirs = new Set(files.map(file => dirname(file)))
for (const dir of dirs) {
// don't merge in main process, unused VHDs will be merged in the next backup run
await this.cleanVm(dir, { remove: true, logWarn: warn })
}
const dedupedVmUuid = new Set(metadatas.map(_ => _.vm.uuid))
await asyncMap(dedupedVmUuid, vmUuid => this.invalidateVmBackupListCache(vmUuid))
}
#getCompressionType() {
@@ -291,13 +284,16 @@ class RemoteAdapter {
}
#useVhdDirectory() {
return this.handler.useVhdDirectory()
return this.handler.type === 's3'
}
#useAlias() {
return this.#useVhdDirectory()
}
getDisk = Disposable.factory(this.getDisk)
getDisk = deduped(this.getDisk, diskId => [diskId])
getDisk = debounceResourceFactory(this.getDisk)
async *getDisk(diskId) {
const handler = this._handler
@@ -334,6 +330,7 @@ class RemoteAdapter {
// - `<partitionId>`: partitioned disk
// - `<pvId>/<vgName>/<lvName>`: LVM on a partitioned disk
// - `/<vgName>/lvName>`: LVM on a raw disk
getPartition = Disposable.factory(this.getPartition)
async *getPartition(diskId, partitionId) {
const devicePath = yield this.getDisk(diskId)
if (partitionId === undefined) {
@@ -362,14 +359,9 @@ class RemoteAdapter {
const handler = this._handler
const backups = { __proto__: null }
await asyncMap(await handler.list(BACKUP_DIR), async entry => {
// ignore hidden and lock files
if (entry[0] !== '.' && !entry.endsWith('.lock')) {
const vmBackups = await this.listVmBackups(entry)
if (vmBackups.length !== 0) {
backups[entry] = vmBackups
}
}
await asyncMap(await handler.list(BACKUP_DIR), async vmUuid => {
const vmBackups = await this.listVmBackups(vmUuid)
backups[vmUuid] = vmBackups
})
return backups
@@ -382,12 +374,8 @@ class RemoteAdapter {
const entriesMap = {}
await asyncMap(await readdir(path), async name => {
try {
const stats = await lstat(`${path}/${name}`)
if (stats.isDirectory()) {
entriesMap[name + '/'] = {}
} else if (stats.isFile()) {
entriesMap[name] = {}
}
const stats = await stat(`${path}/${name}`)
entriesMap[stats.isDirectory() ? `${name}/` : name] = {}
} catch (error) {
if (error == null || error.code !== 'ENOENT') {
throw error
@@ -458,94 +446,34 @@ class RemoteAdapter {
return backupsByPool
}
async invalidateVmBackupListCache(vmUuid) {
await this.handler.unlink(`${BACKUP_DIR}/${vmUuid}/cache.json.gz`)
}
async #getCachabledDataListVmBackups(dir) {
async listVmBackups(vmUuid, predicate) {
const handler = this._handler
const backups = {}
const backups = []
try {
const files = await handler.list(dir, {
const files = await handler.list(`${BACKUP_DIR}/${vmUuid}`, {
filter: isMetadataFile,
prependDir: true,
})
await asyncMap(files, async file => {
try {
const metadata = await this.readVmBackupMetadata(file)
// inject an id usable by importVmBackupNg()
metadata.id = metadata._filename
backups[file] = metadata
if (predicate === undefined || predicate(metadata)) {
// inject an id usable by importVmBackupNg()
metadata.id = metadata._filename
backups.push(metadata)
}
} catch (error) {
warn(`can't read vm backup metadata`, { error, file, dir })
warn(`listVmBackups ${file}`, { error })
}
})
return backups
} catch (error) {
let code
if (error == null || ((code = error.code) !== 'ENOENT' && code !== 'ENOTDIR')) {
throw error
}
}
}
// use _ to mark this method as private by convention
// since we decorate it with synchronized.withKey in the constructor
// and # function are not writeable.
//
// read the list of backup of a Vm from cache
// if cache is missing or broken => regenerate it and return
async _readCacheListVmBackups(vmUuid) {
const dir = `${BACKUP_DIR}/${vmUuid}`
const path = `${dir}/cache.json.gz`
try {
const gzipped = await this.handler.readFile(path)
const text = await fromCallback(zlib.gunzip, gzipped)
return JSON.parse(text)
} catch (error) {
if (error.code !== 'ENOENT') {
warn('Cache file was unreadable', { vmUuid, error })
}
}
// nothing cached, or cache unreadable => regenerate it
const backups = await this.#getCachabledDataListVmBackups(dir)
if (backups === undefined) {
return
}
// detached async action, will not reject
this.#writeVmBackupsCache(path, backups)
return backups
}
async #writeVmBackupsCache(cacheFile, backups) {
try {
const text = JSON.stringify(backups)
const zipped = await fromCallback(zlib.gzip, text)
await this.handler.writeFile(cacheFile, zipped, { flags: 'w' })
} catch (error) {
warn('writeVmBackupsCache', { cacheFile, error })
}
}
async listVmBackups(vmUuid, predicate) {
const backups = []
const cached = await this._readCacheListVmBackups(vmUuid)
if (cached === undefined) {
return []
}
Object.values(cached).forEach(metadata => {
if (predicate === undefined || predicate(metadata)) {
backups.push(metadata)
}
})
return backups.sort(compareTimestamp)
}
@@ -601,42 +529,60 @@ class RemoteAdapter {
})
}
// open the hierarchy of ancestors until we find a full one
async _createSyntheticStream(handler, path) {
const disposableSynthetic = await VhdSynthetic.fromVhdChain(handler, path)
async _createSyntheticStream(handler, paths) {
let disposableVhds = []
// if it's a path : open all hierarchy of parent
if (typeof paths === 'string') {
let vhd,
vhdPath = paths
do {
const disposable = await openVhd(handler, vhdPath)
vhd = disposable.value
disposableVhds.push(disposable)
vhdPath = resolveRelativeFromFile(vhdPath, vhd.header.parentUnicodeName)
} while (vhd.footer.diskType !== Constants.DISK_TYPES.DYNAMIC)
} else {
// only open the list of path given
disposableVhds = paths.map(path => openVhd(handler, path))
}
// I don't want the vhds to be disposed on return
// but only when the stream is done ( or failed )
const disposables = await Disposable.all(disposableVhds)
const vhds = disposables.value
let disposed = false
const disposeOnce = async () => {
if (!disposed) {
disposed = true
try {
await disposableSynthetic.dispose()
await disposables.dispose()
} catch (error) {
warn('openVhd: failed to dispose VHDs', { error })
warn('_createSyntheticStream: failed to dispose VHDs', { error })
}
}
}
const synthetic = disposableSynthetic.value
const synthetic = new VhdSynthetic(vhds)
await synthetic.readHeaderAndFooter()
await synthetic.readBlockAllocationTable()
const stream = await synthetic.stream()
stream.on('end', disposeOnce)
stream.on('close', disposeOnce)
stream.on('error', disposeOnce)
return stream
}
async readDeltaVmBackup(metadata, ignoredVdis) {
async readDeltaVmBackup(metadata) {
const handler = this._handler
const { vbds, vhds, vifs, vm } = metadata
const { vbds, vdis, vhds, vifs, vm } = metadata
const dir = dirname(metadata._filename)
const vdis = ignoredVdis === undefined ? metadata.vdis : pickBy(metadata.vdis, vdi => !ignoredVdis.has(vdi.uuid))
const streams = {}
await asyncMapSettled(Object.keys(vdis), async ref => {
streams[`${ref}.vhd`] = await this._createSyntheticStream(handler, join(dir, vhds[ref]))
await asyncMapSettled(Object.keys(vdis), async id => {
streams[`${id}.vhd`] = await this._createSyntheticStream(handler, join(dir, vhds[id]))
})
return {
@@ -654,10 +600,7 @@ class RemoteAdapter {
}
async readVmBackupMetadata(path) {
// _filename is a private field used to compute the backup id
//
// it's enumerable to make it cacheable
return { ...JSON.parse(await this._handler.readFile(path)), _filename: path }
return Object.defineProperty(JSON.parse(await this._handler.readFile(path)), '_filename', { value: path })
}
}
@@ -672,30 +615,4 @@ Object.assign(RemoteAdapter.prototype, {
isValidXva,
})
decorateMethodsWith(RemoteAdapter, {
_getLvmLogicalVolumes: compose([
Disposable.factory,
[deduped, (devicePath, pvId, vgName) => [devicePath, pvId, vgName]],
debounceResourceFactory,
]),
_getLvmPhysicalVolume: compose([
Disposable.factory,
[deduped, (devicePath, partition) => [devicePath, partition?.id]],
debounceResourceFactory,
]),
_getPartition: compose([
Disposable.factory,
[deduped, (devicePath, partition) => [devicePath, partition?.id]],
debounceResourceFactory,
]),
_usePartitionFiles: Disposable.factory,
getDisk: compose([Disposable.factory, [deduped, diskId => [diskId]], debounceResourceFactory]),
getPartition: Disposable.factory,
})
exports.RemoteAdapter = RemoteAdapter

View File

@@ -1,5 +1,3 @@
'use strict'
const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')
const { PATH_DB_DUMP } = require('./_PoolMetadataBackup.js')

View File

@@ -1,18 +1,12 @@
'use strict'
const CancelToken = require('promise-toolbox/CancelToken')
const CancelToken = require('promise-toolbox/CancelToken.js')
const Zone = require('node-zone')
const logAfterEnd = log => {
const error = new Error('task has already ended')
error.log = log
throw error
const logAfterEnd = () => {
throw new Error('task has already ended')
}
const noop = Function.prototype
const serializeErrors = errors => (Array.isArray(errors) ? errors.map(serializeError) : errors)
// Create a serializable object from an error.
//
// Otherwise some fields might be non-enumerable and missing from logs.
@@ -21,7 +15,6 @@ const serializeError = error =>
? {
...error, // Copy enumerable properties.
code: error.code,
errors: serializeErrors(error.errors), // supports AggregateError
message: error.message,
name: error.name,
stack: error.stack,

View File

@@ -1,5 +1,3 @@
'use strict'
const { asyncMap } = require('@xen-orchestra/async-map')
const { DIR_XO_POOL_METADATA_BACKUPS } = require('./RemoteAdapter.js')

View File

@@ -1,14 +1,11 @@
'use strict'
const assert = require('assert')
const findLast = require('lodash/findLast.js')
const groupBy = require('lodash/groupBy.js')
const ignoreErrors = require('promise-toolbox/ignoreErrors')
const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
const keyBy = require('lodash/keyBy.js')
const mapValues = require('lodash/mapValues.js')
const { asyncMap } = require('@xen-orchestra/async-map')
const { createLogger } = require('@xen-orchestra/log')
const { decorateMethodsWith } = require('@vates/decorate-with')
const { defer } = require('golike-defer')
const { formatDateTime } = require('@xen-orchestra/xapi')
@@ -24,13 +21,6 @@ const { watchStreamSize } = require('./_watchStreamSize.js')
const { debug, warn } = createLogger('xo:backups:VmBackup')
class AggregateError extends Error {
constructor(errors, message) {
super(message)
this.errors = errors
}
}
const asyncEach = async (iterable, fn, thisArg = iterable) => {
for (const item of iterable) {
await fn.call(thisArg, item)
@@ -44,28 +34,17 @@ const forkDeltaExport = deltaExport =>
},
})
class VmBackup {
constructor({
config,
getSnapshotNameLabel,
healthCheckSr,
job,
remoteAdapters,
remotes,
schedule,
settings,
srs,
vm,
}) {
if (vm.other_config['xo:backup:job'] === job.id && 'start' in vm.blocked_operations) {
// don't match replicated VMs created by this very job otherwise they
// will be replicated again and again
exports.VmBackup = class VmBackup {
constructor({ config, getSnapshotNameLabel, job, remoteAdapters, remotes, schedule, settings, srs, vm }) {
if (vm.other_config['xo:backup:job'] === job.id) {
// otherwise replicated VMs would be matched and replicated again and again
throw new Error('cannot backup a VM created by this very job')
}
this.config = config
this.job = job
this.remoteAdapters = remoteAdapters
this.remotes = remotes
this.scheduleId = schedule.id
this.timestamp = undefined
@@ -79,7 +58,6 @@ class VmBackup {
this._fullVdisRequired = undefined
this._getSnapshotNameLabel = getSnapshotNameLabel
this._isDelta = job.mode === 'delta'
this._healthCheckSr = healthCheckSr
this._jobId = job.id
this._jobSnapshots = undefined
this._xapi = vm.$xapi
@@ -106,6 +84,7 @@ class VmBackup {
: [FullBackupWriter, FullReplicationWriter]
const allSettings = job.settings
Object.keys(remoteAdapters).forEach(remoteId => {
const targetSettings = {
...settings,
@@ -128,49 +107,33 @@ class VmBackup {
}
// calls fn for each function, warns of any errors, and throws only if there are no writers left
async _callWriters(fn, step, parallel = true) {
async _callWriters(fn, warnMessage, parallel = true) {
const writers = this._writers
const n = writers.size
if (n === 0) {
return
}
async function callWriter(writer) {
const { name } = writer.constructor
try {
debug('writer step starting', { step, writer: name })
await fn(writer)
debug('writer step succeeded', { duration: step, writer: name })
} catch (error) {
writers.delete(writer)
warn('writer step failed', { error, step, writer: name })
// these two steps are the only one that are not already in their own sub tasks
if (step === 'writer.checkBaseVdis()' || step === 'writer.beforeBackup()') {
Task.warning(
`the writer ${name} has failed the step ${step} with error ${error.message}. It won't be used anymore in this job execution.`
)
}
throw error
}
}
if (n === 1) {
const [writer] = writers
return callWriter(writer)
try {
await fn(writer)
} catch (error) {
writers.delete(writer)
throw error
}
return
}
const errors = []
await (parallel ? asyncMap : asyncEach)(writers, async function (writer) {
try {
await callWriter(writer)
await fn(writer)
} catch (error) {
errors.push(error)
this.delete(writer)
warn(warnMessage, { error, writer: writer.constructor.name })
}
})
if (writers.size === 0) {
throw new AggregateError(errors, 'all targets have failed, step: ' + step)
throw new Error('all targets have failed, step: ' + warnMessage)
}
}
@@ -197,10 +160,7 @@ class VmBackup {
const settings = this._settings
const doSnapshot =
settings.unconditionalSnapshot ||
this._isDelta ||
(!settings.offlineBackup && vm.power_state === 'Running') ||
settings.snapshotRetention !== 0
this._isDelta || (!settings.offlineBackup && vm.power_state === 'Running') || settings.snapshotRetention !== 0
if (doSnapshot) {
await Task.run({ name: 'snapshot' }, async () => {
if (!settings.bypassVdiChainsCheck) {
@@ -208,9 +168,7 @@ class VmBackup {
}
const snapshotRef = await vm[settings.checkpointSnapshot ? '$checkpoint' : '$snapshot']({
ignoreNobakVdis: true,
name_label: this._getSnapshotNameLabel(vm),
unplugVusbs: true,
})
this.timestamp = Date.now()
@@ -332,17 +290,22 @@ class VmBackup {
}
async _removeUnusedSnapshots() {
const allSettings = this.job.settings
const baseSettings = this._baseSettings
const jobSettings = this.job.settings
const baseVmRef = this._baseVm?.$ref
const { config } = this
const baseSettings = {
...config.defaultSettings,
...config.metadata.defaultSettings,
...jobSettings[''],
}
const snapshotsPerSchedule = groupBy(this._jobSnapshots, _ => _.other_config['xo:backup:schedule'])
const xapi = this._xapi
await asyncMap(Object.entries(snapshotsPerSchedule), ([scheduleId, snapshots]) => {
const settings = {
...baseSettings,
...allSettings[scheduleId],
...allSettings[this.vm.uuid],
...jobSettings[scheduleId],
...jobSettings[this.vm.uuid],
}
return asyncMap(getOldEntries(settings.snapshotRetention, snapshots), ({ $ref }) => {
if ($ref !== baseVmRef) {
@@ -421,24 +384,7 @@ class VmBackup {
this._fullVdisRequired = fullVdisRequired
}
async _healthCheck() {
const settings = this._settings
if (this._healthCheckSr === undefined) {
return
}
// check if current VM has tags
const { tags } = this.vm
const intersect = settings.healthCheckVmsWithTags.some(t => tags.includes(t))
if (settings.healthCheckVmsWithTags.length !== 0 && !intersect) {
return
}
await this._callWriters(writer => writer.healthCheck(this._healthCheckSr), 'writer.healthCheck()')
}
run = defer(this.run)
async run($defer) {
const settings = this._settings
assert(
@@ -448,9 +394,7 @@ class VmBackup {
await this._callWriters(async writer => {
await writer.beforeBackup()
$defer(async () => {
await writer.afterBackup()
})
$defer(() => writer.afterBackup())
}, 'writer.beforeBackup()')
await this._fetchJobSnapshots()
@@ -486,11 +430,5 @@ class VmBackup {
await this._fetchJobSnapshots()
await this._removeUnusedSnapshots()
}
await this._healthCheck()
}
}
exports.VmBackup = VmBackup
decorateMethodsWith(VmBackup, {
run: defer,
})

View File

@@ -1,5 +1,3 @@
'use strict'
const { asyncMap } = require('@xen-orchestra/async-map')
const { DIR_XO_CONFIG_BACKUPS } = require('./RemoteAdapter.js')

View File

@@ -1,6 +1,4 @@
'use strict'
exports.isMetadataFile = filename => filename.endsWith('.json')
exports.isVhdFile = filename => filename.endsWith('.vhd')
exports.isXvaFile = filename => filename.endsWith('.xva')
exports.isXvaSumFile = filename => filename.endsWith('.xva.checksum')
exports.isXvaSumFile = filename => filename.endsWith('.xva.cheksum')

View File

@@ -1,16 +1,11 @@
'use strict'
require('@xen-orchestra/log/configure.js').catchGlobalErrors(
require('@xen-orchestra/log').createLogger('xo:backups:worker')
)
require('@vates/cached-dns.lookup').createCachedLookup().patchGlobal()
const Disposable = require('promise-toolbox/Disposable')
const ignoreErrors = require('promise-toolbox/ignoreErrors')
const Disposable = require('promise-toolbox/Disposable.js')
const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
const { compose } = require('@vates/compose')
const { createDebounceResource } = require('@vates/disposable/debounceResource.js')
const { decorateMethodsWith } = require('@vates/decorate-with')
const { deduped } = require('@vates/disposable/deduped.js')
const { getHandler } = require('@xen-orchestra/fs')
const { parseDuration } = require('@vates/parse-duration')
@@ -63,6 +58,11 @@ class BackupWorker {
}).run()
}
getAdapter = Disposable.factory(this.getAdapter)
getAdapter = deduped(this.getAdapter, remote => [remote.url])
getAdapter = compose(this.getAdapter, function (resource) {
return this.debounceResource(resource)
})
async *getAdapter(remote) {
const handler = getHandler(remote, this.#remoteOptions)
await handler.sync()
@@ -77,6 +77,11 @@ class BackupWorker {
}
}
getXapi = Disposable.factory(this.getXapi)
getXapi = deduped(this.getXapi, ({ url }) => [url])
getXapi = compose(this.getXapi, function (resource) {
return this.debounceResource(resource)
})
async *getXapi({ credentials: { username: user, password }, ...opts }) {
const xapi = new Xapi({
...this.#xapiOptions,
@@ -98,30 +103,6 @@ class BackupWorker {
}
}
decorateMethodsWith(BackupWorker, {
getAdapter: compose([
Disposable.factory,
[deduped, remote => [remote.url]],
[
compose,
function (resource) {
return this.debounceResource(resource)
},
],
]),
getXapi: compose([
Disposable.factory,
[deduped, xapi => [xapi.url]],
[
compose,
function (resource) {
return this.debounceResource(resource)
},
],
]),
})
// Received message:
//
// Message {

View File

@@ -1,7 +1,5 @@
'use strict'
const cancelable = require('promise-toolbox/cancelable')
const CancelToken = require('promise-toolbox/CancelToken')
const cancelable = require('promise-toolbox/cancelable.js')
const CancelToken = require('promise-toolbox/CancelToken.js')
// Similar to `Promise.all` + `map` but pass a cancel token to the callback
//

View File

@@ -1,18 +1,14 @@
'use strict'
/* eslint-env jest */
const rimraf = require('rimraf')
const tmp = require('tmp')
const fs = require('fs-extra')
const uuid = require('uuid')
const { getHandler } = require('@xen-orchestra/fs')
const { pFromCallback } = require('promise-toolbox')
const crypto = require('crypto')
const { RemoteAdapter } = require('./RemoteAdapter')
const { VHDFOOTER, VHDHEADER } = require('./tests.fixtures.js')
const { VhdFile, Constants, VhdDirectory, VhdAbstract } = require('vhd-lib')
const { checkAliases } = require('./_cleanVm')
const { dirname, basename } = require('path')
let tempDir, adapter, handler, jobId, vdiId, basePath
@@ -34,17 +30,12 @@ afterEach(async () => {
await handler.forget()
})
const uniqueId = () => uuid.v1()
const uniqueIdBuffer = () => uuid.v1({}, Buffer.alloc(16))
const uniqueId = () => crypto.randomBytes(16).toString('hex')
async function generateVhd(path, opts = {}) {
let vhd
let dataPath = path
if (opts.useAlias) {
await handler.mkdir(dirname(path) + '/data/')
dataPath = dirname(path) + '/data/' + basename(path)
}
const dataPath = opts.useAlias ? path + '.data' : path
if (opts.mode === 'directory') {
await handler.mkdir(dataPath)
vhd = new VhdDirectory(handler, dataPath)
@@ -54,9 +45,10 @@ async function generateVhd(path, opts = {}) {
}
vhd.header = { ...VHDHEADER, ...opts.header }
vhd.footer = { ...VHDFOOTER, ...opts.footer, uuid: uniqueIdBuffer() }
vhd.footer = { ...VHDFOOTER, ...opts.footer }
vhd.footer.uuid = Buffer.from(crypto.randomBytes(16))
if (vhd.header.parentUuid) {
if (vhd.header.parentUnicodeName) {
vhd.footer.diskType = Constants.DISK_TYPES.DIFFERENCING
} else {
vhd.footer.diskType = Constants.DISK_TYPES.DYNAMIC
@@ -78,53 +70,48 @@ test('It remove broken vhd', async () => {
await handler.writeFile(`${basePath}/notReallyAVhd.vhd`, 'I AM NOT A VHD')
expect((await handler.list(basePath)).length).toEqual(1)
let loggued = ''
const logInfo = message => {
const onLog = message => {
loggued += message
}
await adapter.cleanVm('/', { remove: false, logInfo, logWarn: logInfo, lock: false })
expect(loggued).toEqual(`VHD check error`)
await adapter.cleanVm('/', { remove: false, onLog })
expect(loggued).toEqual(`error while checking the VHD with path /${basePath}/notReallyAVhd.vhd`)
// not removed
expect((await handler.list(basePath)).length).toEqual(1)
// really remove it
await adapter.cleanVm('/', { remove: true, logInfo, logWarn: () => {}, lock: false })
await adapter.cleanVm('/', { remove: true, onLog })
expect((await handler.list(basePath)).length).toEqual(0)
})
test('it remove vhd with missing or multiple ancestors', async () => {
// one with a broken parent, should be deleted
// one with a broken parent
await generateVhd(`${basePath}/abandonned.vhd`, {
header: {
parentUnicodeName: 'gone.vhd',
parentUuid: uniqueIdBuffer(),
parentUid: Buffer.from(crypto.randomBytes(16)),
},
})
// one orphan, which is a full vhd, no parent : should stay
// one orphan, which is a full vhd, no parent
const orphan = await generateVhd(`${basePath}/orphan.vhd`)
// a child to the orphan in the metadata : should stay
// a child to the orphan
await generateVhd(`${basePath}/child.vhd`, {
header: {
parentUnicodeName: 'orphan.vhd',
parentUuid: orphan.footer.uuid,
parentUid: orphan.footer.uuid,
},
})
await handler.writeFile(
`metadata.json`,
JSON.stringify({
mode: 'delta',
vhds: [`${basePath}/child.vhd`, `${basePath}/abandonned.vhd`],
}),
{ flags: 'w' }
)
// clean
let loggued = ''
const logInfo = message => {
const onLog = message => {
loggued += message + '\n'
}
await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
await adapter.cleanVm('/', { remove: true, onLog })
const deletedOrphanVhd = loggued.match(/deleting orphan VHD/g) || []
expect(deletedOrphanVhd.length).toEqual(1) // only one vhd should have been deleted
const deletedAbandonnedVhd = loggued.match(/abandonned.vhd is missing/g) || []
expect(deletedAbandonnedVhd.length).toEqual(1) // and it must be abandonned.vhd
// we don't test the filew on disk, since they will all be marker as unused and deleted without a metadata.json file
})
@@ -152,17 +139,19 @@ test('it remove backup meta data referencing a missing vhd in delta backup', asy
await generateVhd(`${basePath}/child.vhd`, {
header: {
parentUnicodeName: 'orphan.vhd',
parentUuid: orphan.footer.uuid,
parentUid: orphan.footer.uuid,
},
})
let loggued = ''
const logInfo = message => {
const onLog = message => {
loggued += message + '\n'
}
await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
let matched = loggued.match(/deleting unused VHD/g) || []
await adapter.cleanVm('/', { remove: true, onLog })
let matched = loggued.match(/deleting unused VHD /g) || []
expect(matched.length).toEqual(1) // only one vhd should have been deleted
matched = loggued.match(/abandonned.vhd is unused/g) || []
expect(matched.length).toEqual(1) // and it must be abandonned.vhd
// a missing vhd cause clean to remove all vhds
await handler.writeFile(
@@ -173,14 +162,14 @@ test('it remove backup meta data referencing a missing vhd in delta backup', asy
`${basePath}/deleted.vhd`, // in metadata but not in vhds
`${basePath}/orphan.vhd`,
`${basePath}/child.vhd`,
// abandonned.vhd is not here anymore
// abandonned.json is not here
],
}),
{ flags: 'w' }
)
loggued = ''
await adapter.cleanVm('/', { remove: true, logInfo, logWarn: () => {}, lock: false })
matched = loggued.match(/deleting unused VHD/g) || []
await adapter.cleanVm('/', { remove: true, onLog })
matched = loggued.match(/deleting unused VHD /g) || []
expect(matched.length).toEqual(2) // all vhds (orphan and child ) should have been deleted
})
@@ -204,28 +193,30 @@ test('it merges delta of non destroyed chain', async () => {
const child = await generateVhd(`${basePath}/child.vhd`, {
header: {
parentUnicodeName: 'orphan.vhd',
parentUuid: orphan.footer.uuid,
parentUid: orphan.footer.uuid,
},
})
// a grand child
await generateVhd(`${basePath}/grandchild.vhd`, {
header: {
parentUnicodeName: 'child.vhd',
parentUuid: child.footer.uuid,
parentUid: child.footer.uuid,
},
})
let loggued = []
const logInfo = message => {
const onLog = message => {
loggued.push(message)
}
await adapter.cleanVm('/', { remove: true, logInfo, logWarn: logInfo, lock: false })
expect(loggued[0]).toEqual(`incorrect backup size in metadata`)
await adapter.cleanVm('/', { remove: true, onLog })
expect(loggued[0]).toEqual(`the parent /${basePath}/orphan.vhd of the child /${basePath}/child.vhd is unused`)
expect(loggued[1]).toEqual(`incorrect size in metadata: 12000 instead of 209920`)
loggued = []
await adapter.cleanVm('/', { remove: true, merge: true, logInfo, logWarn: () => {}, lock: false })
const [merging] = loggued
expect(merging).toEqual(`merging VHD chain`)
await adapter.cleanVm('/', { remove: true, merge: true, onLog })
const [unused, merging] = loggued
expect(unused).toEqual(`the parent /${basePath}/orphan.vhd of the child /${basePath}/child.vhd is unused`)
expect(merging).toEqual(`merging /${basePath}/child.vhd into /${basePath}/orphan.vhd`)
const metadata = JSON.parse(await handler.readFile(`metadata.json`))
// size should be the size of children + grand children after the merge
@@ -244,8 +235,12 @@ test('it finish unterminated merge ', async () => {
`metadata.json`,
JSON.stringify({
mode: 'delta',
size: 209920,
vhds: [`${basePath}/orphan.vhd`, `${basePath}/child.vhd`],
size: undefined,
vhds: [
`${basePath}/orphan.vhd`, // grand child should not be merged
`${basePath}/child.vhd`,
// orphan is not here, he should be merged in child
],
})
)
@@ -255,7 +250,7 @@ test('it finish unterminated merge ', async () => {
const child = await generateVhd(`${basePath}/child.vhd`, {
header: {
parentUnicodeName: 'orphan.vhd',
parentUuid: orphan.footer.uuid,
parentUid: orphan.footer.uuid,
},
})
// a merge in progress file
@@ -271,7 +266,8 @@ test('it finish unterminated merge ', async () => {
})
)
await adapter.cleanVm('/', { remove: true, merge: true, logWarn: () => {}, lock: false })
// a unfinished merging
await adapter.cleanVm('/', { remove: true, merge: true })
// merging is already tested in vhd-lib, don't retest it here (and theses vhd are as empty as my stomach at 12h12)
// only check deletion
@@ -283,17 +279,12 @@ test('it finish unterminated merge ', async () => {
// each of the vhd can be a file, a directory, an alias to a file or an alias to a directory
// the message an resulting files should be identical to the output with vhd files which is tested independantly
describe('tests multiple combination ', () => {
describe('tests mulitple combination ', () => {
for (const useAlias of [true, false]) {
for (const vhdMode of ['file', 'directory']) {
test(`alias : ${useAlias}, mode: ${vhdMode}`, async () => {
// a broken VHD
if (useAlias) {
await handler.mkdir(basePath + '/data')
}
const brokenVhdDataPath = basePath + (useAlias ? '/data/broken.vhd' : '/broken.vhd')
const brokenVhdDataPath = basePath + useAlias ? 'broken.data' : 'broken.vhd'
if (vhdMode === 'directory') {
await handler.mkdir(brokenVhdDataPath)
} else {
@@ -311,10 +302,9 @@ describe('tests multiple combination ', () => {
mode: vhdMode,
header: {
parentUnicodeName: 'gone.vhd',
parentUuid: uniqueIdBuffer(),
parentUid: crypto.randomBytes(16),
},
})
// an ancestor of a vhd present in metadata
const ancestor = await generateVhd(`${basePath}/ancestor.vhd`, {
useAlias,
@@ -325,7 +315,7 @@ describe('tests multiple combination ', () => {
mode: vhdMode,
header: {
parentUnicodeName: 'ancestor.vhd' + (useAlias ? '.alias.vhd' : ''),
parentUuid: ancestor.footer.uuid,
parentUid: ancestor.footer.uuid,
},
})
// a grand child vhd in metadata
@@ -334,7 +324,7 @@ describe('tests multiple combination ', () => {
mode: vhdMode,
header: {
parentUnicodeName: 'child.vhd' + (useAlias ? '.alias.vhd' : ''),
parentUuid: child.footer.uuid,
parentUid: child.footer.uuid,
},
})
@@ -349,7 +339,7 @@ describe('tests multiple combination ', () => {
mode: vhdMode,
header: {
parentUnicodeName: 'cleanAncestor.vhd' + (useAlias ? '.alias.vhd' : ''),
parentUuid: cleanAncestor.footer.uuid,
parentUid: cleanAncestor.footer.uuid,
},
})
@@ -377,8 +367,7 @@ describe('tests multiple combination ', () => {
],
})
)
await adapter.cleanVm('/', { remove: true, merge: true, logWarn: () => {}, lock: false })
await adapter.cleanVm('/', { remove: true, merge: true })
const metadata = JSON.parse(await handler.readFile(`metadata.json`))
// size should be the size of children + grand children + clean after the merge
@@ -390,16 +379,14 @@ describe('tests multiple combination ', () => {
const survivors = await handler.list(basePath)
// console.log(survivors)
if (useAlias) {
const dataSurvivors = await handler.list(basePath + '/data')
// the goal of the alias : do not move a full folder
expect(dataSurvivors).toContain('ancestor.vhd')
expect(dataSurvivors).toContain('grandchild.vhd')
expect(dataSurvivors).toContain('cleanAncestor.vhd')
expect(survivors).toContain('ancestor.vhd.data')
expect(survivors).toContain('grandchild.vhd.data')
expect(survivors).toContain('cleanAncestor.vhd.data')
expect(survivors).toContain('clean.vhd.alias.vhd')
expect(survivors).toContain('child.vhd.alias.vhd')
expect(survivors).toContain('grandchild.vhd.alias.vhd')
expect(survivors.length).toEqual(4) // the 3 ok + data
expect(dataSurvivors.length).toEqual(3) // the 3 ok + data
expect(survivors.length).toEqual(6)
} else {
expect(survivors).toContain('clean.vhd')
expect(survivors).toContain('child.vhd')
@@ -414,31 +401,7 @@ describe('tests multiple combination ', () => {
test('it cleans orphan merge states ', async () => {
await handler.writeFile(`${basePath}/.orphan.vhd.merge.json`, '')
await adapter.cleanVm('/', { remove: true, logWarn: () => {}, lock: false })
await adapter.cleanVm('/', { remove: true })
expect(await handler.list(basePath)).toEqual([])
})
test('check Aliases should work alone', async () => {
await handler.mkdir('vhds')
await handler.mkdir('vhds/data')
await generateVhd(`vhds/data/ok.vhd`)
await VhdAbstract.createAlias(handler, 'vhds/ok.alias.vhd', 'vhds/data/ok.vhd')
await VhdAbstract.createAlias(handler, 'vhds/missingData.alias.vhd', 'vhds/data/nonexistent.vhd')
await generateVhd(`vhds/data/missingalias.vhd`)
await checkAliases(['vhds/missingData.alias.vhd', 'vhds/ok.alias.vhd'], 'vhds/data', {
remove: true,
handler,
logWarn: () => {},
})
// only ok have suvived
const alias = (await handler.list('vhds')).filter(f => f.endsWith('.vhd'))
expect(alias.length).toEqual(1)
const data = await handler.list('vhds/data')
expect(data.length).toEqual(1)
})

View File

@@ -1,27 +1,19 @@
'use strict'
const assert = require('assert')
const sum = require('lodash/sum')
const UUID = require('uuid')
const { asyncMap } = require('@xen-orchestra/async-map')
const { Constants, openVhd, VhdAbstract, VhdFile } = require('vhd-lib')
const { isVhdAlias, resolveVhdAlias } = require('vhd-lib/aliases')
const { Constants, mergeVhd, openVhd, VhdAbstract, VhdFile } = require('vhd-lib')
const { dirname, resolve } = require('path')
const { DISK_TYPES } = Constants
const { isMetadataFile, isVhdFile, isXvaFile, isXvaSumFile } = require('./_backupType.js')
const { limitConcurrency } = require('limit-concurrency-decorator')
const { mergeVhdChain } = require('vhd-lib/merge')
const { Task } = require('./Task.js')
const { Disposable } = require('promise-toolbox')
const handlerPath = require('@xen-orchestra/fs/path')
// checking the size of a vhd directory is costly
// 1 Http Query per 1000 blocks
// we only check size of all the vhd are VhdFiles
function shouldComputeVhdsSize(handler, vhds) {
if (handler.isEncrypted) {
return false
}
function shouldComputeVhdsSize(vhds) {
return vhds.every(vhd => vhd instanceof VhdFile)
}
@@ -29,50 +21,88 @@ const computeVhdsSize = (handler, vhdPaths) =>
Disposable.use(
vhdPaths.map(vhdPath => openVhd(handler, vhdPath)),
async vhds => {
if (shouldComputeVhdsSize(handler, vhds)) {
if (shouldComputeVhdsSize(vhds)) {
const sizes = await asyncMap(vhds, vhd => vhd.getSize())
return sum(sizes)
}
}
)
// chain is [ ancestor, child_1, ..., child_n ]
async function _mergeVhdChain(handler, chain, { logInfo, remove, merge }) {
// chain is an array of VHDs from child to parent
//
// the whole chain will be merged into parent, parent will be renamed to child
// and all the others will deleted
async function mergeVhdChain(chain, { handler, onLog, remove, merge }) {
assert(chain.length >= 2)
let child = chain[0]
const parent = chain[chain.length - 1]
const children = chain.slice(0, -1).reverse()
chain
.slice(1)
.reverse()
.forEach(parent => {
onLog(`the parent ${parent} of the child ${child} is unused`)
})
if (merge) {
logInfo(`merging VHD chain`, { chain })
// `mergeVhd` does not work with a stream, either
// - make it accept a stream
// - or create synthetic VHD which is not a stream
if (children.length !== 1) {
// TODO: implement merging multiple children
children.length = 1
child = children[0]
}
onLog(`merging ${child} into ${parent}`)
let done, total
const handle = setInterval(() => {
if (done !== undefined) {
logInfo('merge in progress', {
done,
parent: chain[0],
progress: Math.round((100 * done) / total),
total,
})
onLog(`merging ${child}: ${done}/${total}`)
}
}, 10e3)
try {
return await mergeVhdChain(handler, chain, {
logInfo,
const mergedSize = await mergeVhd(
handler,
parent,
handler,
child,
// children.length === 1
// ? child
// : await createSyntheticStream(handler, children),
{
onProgress({ done: d, total: t }) {
done = d
total = t
},
removeUnused: remove,
})
} finally {
clearInterval(handle)
}
}
)
clearInterval(handle)
await Promise.all([
VhdAbstract.rename(handler, parent, child),
asyncMap(children.slice(0, -1), child => {
onLog(`the VHD ${child} is unused`)
if (remove) {
onLog(`deleting unused VHD ${child}`)
return VhdAbstract.unlink(handler, child)
}
}),
])
return mergedSize
}
}
const noop = Function.prototype
const INTERRUPTED_VHDS_REG = /^\.(.+)\.merge.json$/
const listVhds = async (handler, vmDir, logWarn) => {
const listVhds = async (handler, vmDir) => {
const vhds = new Set()
const aliases = {}
const interruptedVhds = new Map()
await asyncMap(
@@ -89,110 +119,37 @@ const listVhds = async (handler, vmDir, logWarn) => {
const list = await handler.list(vdiDir, {
filter: file => isVhdFile(file) || INTERRUPTED_VHDS_REG.test(file),
})
aliases[vdiDir] = list.filter(vhd => isVhdAlias(vhd)).map(file => `${vdiDir}/${file}`)
await asyncMap(list, async file => {
list.forEach(file => {
const res = INTERRUPTED_VHDS_REG.exec(file)
if (res === null) {
vhds.add(`${vdiDir}/${file}`)
} else {
try {
const mergeState = JSON.parse(await handler.readFile(`${vdiDir}/${file}`))
interruptedVhds.set(`${vdiDir}/${res[1]}`, {
statePath: `${vdiDir}/${file}`,
chain: mergeState.chain,
})
} catch (error) {
// fall back to a non resuming merge
vhds.add(`${vdiDir}/${file}`)
logWarn('failed to read existing merge state', { path: file, error })
}
interruptedVhds.set(`${vdiDir}/${res[1]}`, `${vdiDir}/${file}`)
}
})
}
)
)
return { vhds, interruptedVhds, aliases }
return { vhds, interruptedVhds }
}
async function checkAliases(
aliasPaths,
targetDataRepository,
{ handler, logInfo = noop, logWarn = console.warn, remove = false }
) {
const aliasFound = []
for (const alias of aliasPaths) {
const target = await resolveVhdAlias(handler, alias)
if (!isVhdFile(target)) {
logWarn('alias references non VHD target', { alias, target })
if (remove) {
logInfo('removing alias and non VHD target', { alias, target })
await handler.unlink(target)
await handler.unlink(alias)
}
continue
}
try {
const { dispose } = await openVhd(handler, target)
try {
await dispose()
} catch (e) {
// error during dispose should not trigger a deletion
}
} catch (error) {
logWarn('missing or broken alias target', { alias, target, error })
if (remove) {
try {
await VhdAbstract.unlink(handler, alias)
} catch (error) {
if (error.code !== 'ENOENT') {
logWarn('error deleting alias target', { alias, target, error })
}
}
}
continue
}
aliasFound.push(resolve('/', target))
}
const vhds = await handler.list(targetDataRepository, {
ignoreMissing: true,
prependDir: true,
})
await asyncMap(vhds, async path => {
if (!aliasFound.includes(path)) {
logWarn('no alias references VHD', { path })
if (remove) {
logInfo('deleting unused VHD', { path })
await VhdAbstract.unlink(handler, path)
}
}
})
}
exports.checkAliases = checkAliases
const defaultMergeLimiter = limitConcurrency(1)
exports.cleanVm = async function cleanVm(
vmDir,
{ fixMetadata, remove, merge, mergeLimiter = defaultMergeLimiter, logInfo = noop, logWarn = console.warn }
{ fixMetadata, remove, merge, mergeLimiter = defaultMergeLimiter, onLog = noop }
) {
const limitedMergeVhdChain = mergeLimiter(_mergeVhdChain)
const limitedMergeVhdChain = mergeLimiter(mergeVhdChain)
const handler = this._handler
const vhdsToJSons = new Set()
const vhdById = new Map()
const vhdParents = { __proto__: null }
const vhdChildren = { __proto__: null }
const { vhds, interruptedVhds, aliases } = await listVhds(handler, vmDir, logWarn)
const { vhds, interruptedVhds } = await listVhds(handler, vmDir)
// remove broken VHDs
await asyncMap(vhds, async path => {
@@ -210,31 +167,12 @@ exports.cleanVm = async function cleanVm(
}
vhdChildren[parent] = path
}
// Detect VHDs with the same UUIDs
//
// Due to a bug introduced in a1bcd35e2
const duplicate = vhdById.get(UUID.stringify(vhd.footer.uuid))
let vhdKept = vhd
if (duplicate !== undefined) {
logWarn('uuid is duplicated', { uuid: UUID.stringify(vhd.footer.uuid) })
if (duplicate.containsAllDataOf(vhd)) {
logWarn(`should delete ${path}`)
vhdKept = duplicate
vhds.delete(path)
} else if (vhd.containsAllDataOf(duplicate)) {
logWarn(`should delete ${duplicate._path}`)
vhds.delete(duplicate._path)
} else {
logWarn('same ids but different content')
}
}
vhdById.set(UUID.stringify(vhdKept.footer.uuid), vhdKept)
})
} catch (error) {
vhds.delete(path)
logWarn('VHD check error', { path, error })
onLog(`error while checking the VHD with path ${path}`, { error })
if (error?.code === 'ERR_ASSERTION' && remove) {
logInfo('deleting broken VHD', { path })
onLog(`deleting broken ${path}`)
return VhdAbstract.unlink(handler, path)
}
}
@@ -243,25 +181,21 @@ exports.cleanVm = async function cleanVm(
// remove interrupted merge states for missing VHDs
for (const interruptedVhd of interruptedVhds.keys()) {
if (!vhds.has(interruptedVhd)) {
const { statePath } = interruptedVhds.get(interruptedVhd)
const statePath = interruptedVhds.get(interruptedVhd)
interruptedVhds.delete(interruptedVhd)
logWarn('orphan merge state', {
onLog('orphan merge state', {
mergeStatePath: statePath,
missingVhdPath: interruptedVhd,
})
if (remove) {
logInfo('deleting orphan merge state', { statePath })
onLog(`deleting orphan merge state ${statePath}`)
await handler.unlink(statePath)
}
}
}
// check if alias are correct
// check if all vhd in data subfolder have a corresponding alias
await asyncMap(Object.keys(aliases), async dir => {
await checkAliases(aliases[dir], `${dir}/data`, { handler, logInfo, logWarn, remove })
})
// @todo : add check for data folder of alias not referenced in a valid alias
// remove VHDs with missing ancestors
{
@@ -282,9 +216,9 @@ exports.cleanVm = async function cleanVm(
if (!vhds.has(parent)) {
vhds.delete(vhdPath)
logWarn('parent VHD is missing', { parent, child: vhdPath })
onLog(`the parent ${parent} of the VHD ${vhdPath} is missing`)
if (remove) {
logInfo('deleting orphan VHD', { path: vhdPath })
onLog(`deleting orphan VHD ${vhdPath}`)
deletions.push(VhdAbstract.unlink(handler, vhdPath))
}
}
@@ -321,7 +255,7 @@ exports.cleanVm = async function cleanVm(
// check is not good enough to delete the file, the best we can do is report
// it
if (!(await this.isValidXva(path))) {
logWarn('XVA might be broken', { path })
onLog(`the XVA with path ${path} is potentially broken`)
}
})
@@ -335,7 +269,7 @@ exports.cleanVm = async function cleanVm(
try {
metadata = JSON.parse(await handler.readFile(json))
} catch (error) {
logWarn('failed to read backup metadata', { path: json, error })
onLog(`failed to read metadata file ${json}`, { error })
jsons.delete(json)
return
}
@@ -346,9 +280,9 @@ exports.cleanVm = async function cleanVm(
if (xvas.has(linkedXva)) {
unusedXvas.delete(linkedXva)
} else {
logWarn('the XVA linked to the backup is missing', { backup: json, xva: linkedXva })
onLog(`the XVA linked to the metadata ${json} is missing`)
if (remove) {
logInfo('deleting incomplete backup', { path: json })
onLog(`deleting incomplete backup ${json}`)
jsons.delete(json)
await handler.unlink(json)
}
@@ -369,9 +303,9 @@ exports.cleanVm = async function cleanVm(
vhdsToJSons[path] = json
})
} else {
logWarn('some VHDs linked to the backup are missing', { backup: json, missingVhds })
onLog(`Some VHDs linked to the metadata ${json} are missing`, { missingVhds })
if (remove) {
logInfo('deleting incomplete backup', { path: json })
onLog(`deleting incomplete backup ${json}`)
jsons.delete(json)
await handler.unlink(json)
}
@@ -383,7 +317,7 @@ exports.cleanVm = async function cleanVm(
const unusedVhdsDeletion = []
const toMerge = []
{
// VHD chains (as list from oldest to most recent) to merge indexed by most recent
// VHD chains (as list from child to ancestor) to merge indexed by last
// ancestor
const vhdChainsToMerge = { __proto__: null }
@@ -407,14 +341,14 @@ exports.cleanVm = async function cleanVm(
if (child !== undefined) {
const chain = getUsedChildChainOrDelete(child)
if (chain !== undefined) {
chain.unshift(vhd)
chain.push(vhd)
return chain
}
}
logWarn('unused VHD', { path: vhd })
onLog(`the VHD ${vhd} is unused`)
if (remove) {
logInfo('deleting unused VHD', { path: vhd })
onLog(`deleting unused VHD ${vhd}`)
unusedVhdsDeletion.push(VhdAbstract.unlink(handler, vhd))
}
}
@@ -425,13 +359,7 @@ exports.cleanVm = async function cleanVm(
// merge interrupted VHDs
for (const parent of interruptedVhds.keys()) {
// before #6349 the chain wasn't in the mergeState
const { chain, statePath } = interruptedVhds.get(parent)
if (chain === undefined) {
vhdChainsToMerge[parent] = [parent, vhdChildren[parent]]
} else {
vhdChainsToMerge[parent] = chain.map(vhdPath => handlerPath.resolveFromFile(statePath, vhdPath))
}
vhdChainsToMerge[parent] = [vhdChildren[parent], parent]
}
Object.values(vhdChainsToMerge).forEach(chain => {
@@ -444,9 +372,9 @@ exports.cleanVm = async function cleanVm(
const metadataWithMergedVhd = {}
const doMerge = async () => {
await asyncMap(toMerge, async chain => {
const merged = await limitedMergeVhdChain(handler, chain, { logInfo, logWarn, remove, merge })
const merged = await limitedMergeVhdChain(chain, { handler, onLog, remove, merge })
if (merged !== undefined) {
const metadataPath = vhdsToJSons[chain[chain.length - 1]] // all the chain should have the same metada file
const metadataPath = vhdsToJSons[chain[0]] // all the chain should have the same metada file
metadataWithMergedVhd[metadataPath] = true
}
})
@@ -456,18 +384,18 @@ exports.cleanVm = async function cleanVm(
...unusedVhdsDeletion,
toMerge.length !== 0 && (merge ? Task.run({ name: 'merge' }, doMerge) : doMerge()),
asyncMap(unusedXvas, path => {
logWarn('unused XVA', { path })
onLog(`the XVA ${path} is unused`)
if (remove) {
logInfo('deleting unused XVA', { path })
onLog(`deleting unused XVA ${path}`)
return handler.unlink(path)
}
}),
asyncMap(xvaSums, path => {
// no need to handle checksums for XVAs deleted by the script, they will be handled by `unlink()`
if (!xvas.has(path.slice(0, -'.checksum'.length))) {
logInfo('unused XVA checksum', { path })
onLog(`the XVA checksum ${path} is unused`)
if (remove) {
logInfo('deleting unused XVA checksum', { path })
onLog(`deleting unused XVA checksum ${path}`)
return handler.unlink(path)
}
}
@@ -489,11 +417,7 @@ exports.cleanVm = async function cleanVm(
if (mode === 'full') {
// a full backup : check size
const linkedXva = resolve('/', vmDir, xva)
try {
fileSystemSize = await handler.getSize(linkedXva)
} catch (error) {
// can fail with encrypted remote
}
fileSystemSize = await handler.getSize(linkedXva)
} else if (mode === 'delta') {
const linkedVhds = Object.keys(vhds).map(key => resolve('/', vmDir, vhds[key]))
fileSystemSize = await computeVhdsSize(handler, linkedVhds)
@@ -505,15 +429,11 @@ exports.cleanVm = async function cleanVm(
// don't warn if the size has changed after a merge
if (!merged && fileSystemSize !== size) {
logWarn('incorrect backup size in metadata', {
path: metadataPath,
actual: size ?? 'none',
expected: fileSystemSize,
})
onLog(`incorrect size in metadata: ${size ?? 'none'} instead of ${fileSystemSize}`)
}
}
} catch (error) {
logWarn('failed to get backup size', { backup: metadataPath, error })
onLog(`failed to get size of ${metadataPath}`, { error })
return
}
@@ -523,7 +443,7 @@ exports.cleanVm = async function cleanVm(
try {
await handler.writeFile(metadataPath, JSON.stringify(metadata), { flags: 'w' })
} catch (error) {
logWarn('failed to update backup size in metadata', { path: metadataPath, error })
onLog(`failed to update size in backup metadata ${metadataPath} after merge`, { error })
}
}
})

View File

@@ -1,9 +1,7 @@
'use strict'
const compareVersions = require('compare-versions')
const find = require('lodash/find.js')
const groupBy = require('lodash/groupBy.js')
const ignoreErrors = require('promise-toolbox/ignoreErrors')
const ignoreErrors = require('promise-toolbox/ignoreErrors.js')
const omit = require('lodash/omit.js')
const { asyncMap } = require('@xen-orchestra/async-map')
const { CancelToken } = require('promise-toolbox')
@@ -11,8 +9,6 @@ const { createVhdStreamWithLength } = require('vhd-lib')
const { defer } = require('golike-defer')
const { cancelableMap } = require('./_cancelableMap.js')
const { Task } = require('./Task.js')
const { pick } = require('lodash')
const TAG_BASE_DELTA = 'xo:base_delta'
exports.TAG_BASE_DELTA = TAG_BASE_DELTA
@@ -21,17 +17,6 @@ const TAG_COPY_SRC = 'xo:copy_of'
exports.TAG_COPY_SRC = TAG_COPY_SRC
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
}
exports.exportDeltaVm = async function exportDeltaVm(
vm,
@@ -65,6 +50,17 @@ exports.exportDeltaVm = async function exportDeltaVm(
return
}
// If the VDI name start with `[NOBAK]`, do not export it.
if (vdi.name_label.startsWith('[NOBAK]')) {
// FIXME: find a way to not create the VDI snapshot in the
// first time.
//
// The snapshot must not exist otherwise it could break the
// next export.
ignoreErrors.call(vdi.$destroy())
return
}
vbds[vbd.$ref] = vbd
const vdiRef = vdi.$ref
@@ -169,12 +165,6 @@ exports.importDeltaVm = defer(async function importDeltaVm(
}
}
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 => {
@@ -189,25 +179,19 @@ exports.importDeltaVm = defer(async function importDeltaVm(
let suspendVdi
if (vmRecord.power_state === 'Suspended') {
const vdi = vdiRecords[vmRecord.suspend_VDI]
if (vdi === undefined) {
Task.warning('Suspend VDI not available for this suspended VM', {
vm: pick(vmRecord, 'uuid', 'name_label'),
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: mapVdisSrs[vdi.uuid] ?? sr.$ref,
})
} 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,
})
)
$defer.onFailure(() => suspendVdi.$destroy())
}
)
$defer.onFailure(() => suspendVdi.$destroy())
}
// 1. Create the VM.
@@ -271,7 +255,7 @@ exports.importDeltaVm = defer(async function importDeltaVm(
[TAG_BASE_DELTA]: undefined,
[TAG_COPY_SRC]: vdi.uuid,
},
SR: mapVdisSrRefs[vdi.uuid] ?? sr.$ref,
SR: mapVdisSrs[vdi.uuid] ?? sr.$ref,
})
)
$defer.onFailure(() => newVdi.$destroy())

View File

@@ -1,5 +1,3 @@
'use strict'
exports.extractIdsFromSimplePattern = function extractIdsFromSimplePattern(pattern) {
if (pattern === undefined) {
return []

View File

@@ -1,5 +1,3 @@
'use strict'
const { utcFormat, utcParse } = require('d3-time-format')
// Format a date in ISO 8601 in a safe way to be used in filenames

View File

@@ -1,10 +1,6 @@
'use strict'
const eos = require('end-of-stream')
const { PassThrough } = require('stream')
const { debug } = require('@xen-orchestra/log').createLogger('xo:backups:forkStreamUnpipe')
// create a new readable stream from an existing one which may be piped later
//
// in case of error in the new readable stream, it will simply be unpiped
@@ -13,23 +9,18 @@ exports.forkStreamUnpipe = function forkStreamUnpipe(stream) {
const { forks = 0 } = stream
stream.forks = forks + 1
debug('forking', { forks: stream.forks })
const proxy = new PassThrough()
stream.pipe(proxy)
eos(stream, error => {
if (error !== undefined) {
debug('error on original stream, destroying fork', { error })
proxy.destroy(error)
}
})
eos(proxy, error => {
debug('end of stream, unpiping', { error, forks: --stream.forks })
eos(proxy, _ => {
stream.forks--
stream.unpipe(proxy)
if (stream.forks === 0) {
debug('no more forks, destroying original stream')
stream.destroy(new Error('no more consumers for this stream'))
}
})

View File

@@ -1,5 +1,3 @@
'use strict'
// returns all entries but the last retention-th
exports.getOldEntries = function getOldEntries(retention, entries) {
return entries === undefined ? [] : retention > 0 ? entries.slice(0, -retention) : entries

View File

@@ -1,6 +1,4 @@
'use strict'
const Disposable = require('promise-toolbox/Disposable')
const Disposable = require('promise-toolbox/Disposable.js')
const { join } = require('path')
const { mkdir, rmdir } = require('fs-extra')
const { tmpdir } = require('os')

Some files were not shown because too many files have changed in this diff Show More