feat(backups): implement speed limit at job level

@xen-orchestra/backups/Backup.js

@@ -12,6 +12,7 @@ const { PoolMetadataBackup } = require('./_PoolMetadataBackup.js')
 const { Task } = require('./Task.js')
 const { VmBackup } = require('./_VmBackup.js')
 const { XoMetadataBackup } = require('./_XoMetadataBackup.js')
+const createStreamThrottle = require('./_createStreamThrottle.js')
 
 const noop = Function.prototype
 
@@ -40,6 +41,7 @@ const DEFAULT_VM_SETTINGS = {
   fullInterval: 0,
   healthCheckSr: undefined,
   healthCheckVmsWithTags: [],
+  maxExportRate: 0,
   maxMergedDeltasPerRun: Infinity,
   offlineBackup: false,
   offlineSnapshot: false,
@@ -226,9 +228,11 @@ exports.Backup = class Backup {
     // FIXME: proper SimpleIdPattern handling
     const getSnapshotNameLabel = this._getSnapshotNameLabel
     const schedule = this._schedule
+    const settings = this._settings
+
+    const throttleStream = createStreamThrottle(settings.maxExportRate)
 
     const config = this._config
-    const settings = this._settings
     await Disposable.use(
       Disposable.all(
         extractIdsFromSimplePattern(job.srs).map(id =>
@@ -278,6 +282,7 @@ exports.Backup = class Backup {
                 schedule,
                 settings: { ...settings, ...allSettings[vm.uuid] },
                 srs,
+                throttleStream,
                 vm,
               }).run()
             )

@xen-orchestra/backups/_VmBackup.js

@@ -55,6 +55,7 @@ class VmBackup {
     schedule,
     settings,
     srs,
+    throttleStream,
     vm,
   }) {
     if (vm.other_config['xo:backup:job'] === job.id && 'start' in vm.blocked_operations) {
@@ -82,6 +83,7 @@ class VmBackup {
     this._healthCheckSr = healthCheckSr
     this._jobId = job.id
     this._jobSnapshots = undefined
+    this._throttleStream = throttleStream
     this._xapi = vm.$xapi
 
     // Base VM for the export
@@ -244,6 +246,7 @@ class VmBackup {
       fullVdisRequired,
     })
     const sizeContainers = mapValues(deltaExport.streams, stream => watchStreamSize(stream))
+    deltaExport.streams = mapValues(deltaExport.streams, this._throttleStream)
 
     const timestamp = Date.now()
 
@@ -285,10 +288,12 @@ class VmBackup {
 
   async _copyFull() {
     const { compression } = this.job
-    const stream = await this._xapi.VM_export(this.exportedVm.$ref, {
-      compress: Boolean(compression) && (compression === 'native' ? 'gzip' : 'zstd'),
-      useSnapshot: false,
-    })
+    const stream = this._throttleStream(
+      await this._xapi.VM_export(this.exportedVm.$ref, {
+        compress: Boolean(compression) && (compression === 'native' ? 'gzip' : 'zstd'),
+        useSnapshot: false,
+      })
+    )
     const sizeContainer = watchStreamSize(stream)
 
     const timestamp = Date.now()

@xen-orchestra/backups/_createStreamThrottle.js

@@ -0,0 +1,17 @@
+'use strict'
+
+const { pipeline } = require('node:stream')
+const { ThrottleGroup } = require('@kldzj/stream-throttle')
+const identity = require('lodash/identity.js')
+
+const noop = Function.prototype
+
+module.exports = function createStreamThrottle(rate) {
+  if (rate === 0) {
+    return identity
+  }
+  const group = new ThrottleGroup({ rate })
+  return function throttleStream(stream) {
+    return pipeline(stream, group.createThrottle(), noop)
+  }
+}

@xen-orchestra/backups/package.json

@@ -17,6 +17,7 @@
     "test": "node--test"
   },
   "dependencies": {
+    "@kldzj/stream-throttle": "^1.1.1",
     "@vates/async-each": "^1.0.0",
     "@vates/cached-dns.lookup": "^1.0.0",
     "@vates/compose": "^2.1.0",

yarn.lock

@@ -2738,6 +2738,13 @@
     "@jridgewell/resolve-uri" "3.1.0"
     "@jridgewell/sourcemap-codec" "1.4.14"
 
+"@kldzj/stream-throttle@^1.1.1":
+  version "1.1.1"
+  resolved "https://registry.yarnpkg.com/@kldzj/stream-throttle/-/stream-throttle-1.1.1.tgz#16d18d03fe97a02890724c183dddb17d25bf2ad8"
+  integrity sha512-/x0r2yUUMNe9SuHyaYtNV2b4LnVayEkCOFbP+5kr3sgYce6MyTAc2EEm7hV6lIWgXIT6GWUWzlDhQt6kwsxRUA==
+  dependencies:
+    limiter "^2.1.0"
+
 "@koa/router@^12.0.0":
   version "12.0.0"
   resolved "https://registry.yarnpkg.com/@koa/router/-/router-12.0.0.tgz#2ae7937093fd392761c0e5833c368379d4a35737"
@@ -12867,6 +12874,11 @@ just-extend@^4.0.2:
   resolved "https://registry.yarnpkg.com/just-extend/-/just-extend-4.2.1.tgz#ef5e589afb61e5d66b24eca749409a8939a8c744"
   integrity sha512-g3UB796vUFIY90VIv/WX3L2c8CS2MdWUww3CNrYmqza1Fg0DURc2K/O4YrnklBdQarSJ/y8JnJYDGc+1iumQjg==
 
+just-performance@4.3.0:
+  version "4.3.0"
+  resolved "https://registry.yarnpkg.com/just-performance/-/just-performance-4.3.0.tgz#cc2bc8c9227f09e97b6b1df4cd0de2df7ae16db1"
+  integrity sha512-L7RjvtJsL0QO8xFs5wEoDDzzJwoiowRw6Rn/GnvldlchS2JQr9wFYPiwZcDfrbbujEKqKN0tvENdbjXdYhDp5Q==
+
 just-reduce-object@^1.0.3:
   version "1.2.1"
   resolved "https://registry.yarnpkg.com/just-reduce-object/-/just-reduce-object-1.2.1.tgz#92845dedc4c5da34df5e5ad6a4bf62f21fdc37f5"
@@ -13240,6 +13252,13 @@ limit-concurrency-decorator@^0.5.0:
   resolved "https://registry.yarnpkg.com/limit-concurrency-decorator/-/limit-concurrency-decorator-0.5.0.tgz#7455fc7c8d12e93ce725cb98dc18b861397dd726"
   integrity sha512-s5HqdnTpRJhvK/vleMY3qJ3yEfIQ1BUCUqbBJwtXCKngMSc+qpS1Rl6/rxdhr1Z/oQz3keYho6G4XCFSHb7nbA==
 
+limiter@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/limiter/-/limiter-2.1.0.tgz#d38d7c5b63729bb84fb0c4d8594b7e955a5182a2"
+  integrity sha512-361TYz6iay6n+9KvUUImqdLuFigK+K79qrUtBsXhJTLdH4rIt/r1y8r1iozwh8KbZNpujbFTSh74mJ7bwbAMOw==
+  dependencies:
+    just-performance "4.3.0"
+
 lines-and-columns@^1.1.6:
   version "1.2.4"
   resolved "https://registry.yarnpkg.com/lines-and-columns/-/lines-and-columns-1.2.4.tgz#eca284f75d2965079309dc0ad9255abb2ebc1632"