feat(rest-api): expose backup-log in the SSE (#10563)

* feat(rest-api): expose backup-log in the SSE

* noop instead undefined

* fix: handle sequance

* changelog

* fix

* final fix

* emit remove event from the gc

* Kilian's feedback

---------

Co-authored-by: Killian <killian.allegrain@vates.tech>

packages/xo-server/src/xo-mixins/backups-ng-logs.mjs

@@ -83,7 +83,7 @@ function adaptTask(task) {
   task.tasks?.forEach(adaptTask)
 }
 
-function taskFormatAdapter(log) {
+export function taskFormatAdapter(log) {
   const isXoTask = log.tasks?.length > 0 && !!log.tasks[0].properties
   if (isXoTask) {
     adaptTask(log.tasks[0])

packages/xo-server/src/xo-mixins/jobs/index.mjs

@@ -10,10 +10,13 @@ import { noSuchObject } from 'xo-common/api-errors.js'
 
 import Collection from '../../collection/redis.mjs'
 import patch from '../../patch.mjs'
-import { serializeError } from '../../utils.mjs'
+import { noop, serializeError } from '../../utils.mjs'
 
 import executeCall from './execute-call.mjs'
 import backupGuard from '../../api/_backupGuard.mjs'
+import { REMOVE_CACHE_ENTRY } from '../../_pDebounceWithKey.mjs'
+import { EventEmitter } from 'node:events'
+import { taskFormatAdapter } from '../backups-ng-logs.mjs'
 
 // ===================================================================
 
@@ -52,11 +55,17 @@ class JobsDb extends Collection {
 // -----------------------------------------------------------------------------
 
 export default class Jobs {
+  #backupLogsEe = new EventEmitter()
   get runningJobs() {
     return this._runningJobs
   }
 
+  get backupLogsEe() {
+    return this.#backupLogsEe
+  }
+
   constructor(app) {
+    /** @type {import('@vates/types').XoApp} */
     this._app = app
     const executors = (this._executors = { __proto__: null })
     this._logger = undefined
@@ -178,6 +187,8 @@ export default class Jobs {
   @decorateWith(defer)
   async runJob($defer, job, schedule, data_) {
     const logger = this._logger
+    let unwatchBackupTask = noop
+
     const { id, type } = job
 
     const jobData = {
@@ -199,12 +210,46 @@ export default class Jobs {
       key: job.key,
       type,
     }
-    const runJobId = logger.notice(`Starting execution of ${id}.`, jobData)
+    const runJobId = await logger.notice(`Starting execution of ${id}.`, jobData, true)
+    const app = this._app
     // Links the backup log to the job run
     // We keep the jobs for this because of some mechanism related to jobs, like preventing double execution.
     jobData.runJobId = runJobId
 
-    const app = this._app
+    let previousBackupLog
+    let startBackupLog
+    // ensure we don't stop a backup job due to an issue when trying to fetch backup logs
+    async function safeGetBackupLog(runJobId, { bypassCache = false } = {}) {
+      try {
+        if (bypassCache) {
+          app.getBackupNgLogs(REMOVE_CACHE_ENTRY, runJobId)
+        }
+        return await app.getBackupNgLogs(runJobId)
+      } catch (error) {
+        log.error('cannot get backup-log', { error, runJobId })
+        return undefined
+      }
+    }
+
+    if (type === 'backup' || job.key === undefined) {
+      const safeBackupLog = await safeGetBackupLog(runJobId)
+      if (safeBackupLog === undefined) {
+        log.warn('SSE ignored to not interrupt the backup-job')
+      } else {
+        startBackupLog = {
+          ...safeBackupLog,
+          status: 'pending', // overwrite the status, because `getBackupNgLogs` return a `interrupted` status here. see `handleLog`
+        }
+        app.backupLogsEe.emit('add', startBackupLog)
+        previousBackupLog = startBackupLog
+      }
+    }
+
+    function emitBackupLogUpdate(backupLog) {
+      app.backupLogsEe.emit('update', backupLog, previousBackupLog)
+      previousBackupLog = backupLog
+    }
+
     try {
       let executor = this._executors[type]
       if (executor === undefined) {
@@ -295,13 +340,23 @@ export default class Jobs {
 
       // Links the job run to its backup log
       const jobUpdateFct = async backupTaskId => {
+        if (startBackupLog !== undefined) {
+          function onBackupTaskUpdate(task) {
+            const backupLog = { ...startBackupLog, tasks: [structuredClone(task)] }
+            taskFormatAdapter(backupLog)
+            emitBackupLogUpdate(backupLog)
+          }
+
+          app.tasks.on(backupTaskId, onBackupTaskUpdate)
+          unwatchBackupTask = () => app.tasks.off(backupTaskId, onBackupTaskUpdate)
+        }
+
         await logger.notice(`Adding backupTaskId to job run ${runJobId}`, {
           backupTaskId,
           event: 'job.backupTaskStart',
           runJobId,
         })
       }
-
       await executor({
         app,
         cancelToken: token,
@@ -337,6 +392,16 @@ export default class Jobs {
       )
       app.emit('job:terminated', runJobId, { type })
       throw error
+    } finally {
+      if (startBackupLog !== undefined) {
+        unwatchBackupTask()
+        const safeBackupLog = await safeGetBackupLog(runJobId, { bypassCache: true })
+        if (safeBackupLog === undefined) {
+          log.warn('skip SSE event due to a not found backup-log')
+        } else {
+          emitBackupLogUpdate(safeBackupLog)
+        }
+      }
     }
   }
 

packages/xo-server/src/xo-mixins/logs/index.mjs

@@ -5,12 +5,16 @@ import { dedupe } from '@xen-orchestra/log/dedupe'
 import { defer, fromEvent } from 'promise-toolbox'
 
 import LevelDbLogger from './loggers/leveldb.mjs'
+import { REMOVE_CACHE_ENTRY } from '../../_pDebounceWithKey.mjs'
 
 const { DEBUG } = process.env
 
 const { warn } = createLogger('xo:mixins:logs')
 export default class Logs {
   constructor(app) {
+    /**
+     * @type {import('@vates/types').XoApp}
+     */
     this._app = app
 
     app.hooks.on('clean', () => this._gc())
@@ -59,7 +63,14 @@ export default class Logs {
   }
 
   async _gc(keep = 2e4) {
-    const db = await this._app.getStore('logs')
+    const [db, backupLogs] = await Promise.all([
+      this._app.getStore('logs'),
+      this._app.getBackupNgLogs().catch(error => {
+        warn('cannot snapshot backup logs, no remove events for this GC', { error })
+        return {}
+      }),
+    ])
+    const removedBackupLogs = []
 
     let count = 1
     const { promise, resolve } = defer()
@@ -75,7 +86,13 @@ export default class Logs {
 
     const deleteEntry = key => {
       ++count
-      db.del(key, cb)
+      const backupLog = backupLogs[key]
+      db.del(key, error => {
+        if (error == null && backupLog?.message === 'backup') {
+          removedBackupLogs.push(backupLog)
+        }
+        cb()
+      })
     }
 
     const onData =
@@ -91,6 +108,15 @@ export default class Logs {
 
     await fromEvent(stream, 'end')
     cb()
+    await promise
+
+    if (removedBackupLogs.length > 0) {
+      this._app.getBackupNgLogs(REMOVE_CACHE_ENTRY)
+      removedBackupLogs.forEach(backupLog => {
+        this._app.getBackupNgLogs(REMOVE_CACHE_ENTRY, backupLog.id)
+        this._app.backupLogsEe.emit('remove', undefined, backupLog)
+      })
+    }
 
     return promise
   }