Merge pull request #23920 from overleaf/bg-backup-queue

initial history backup queue worker app

GitOrigin-RevId: e9647a2ec3caeb1fff843cead12164ba89fdb1f8

services/history-v1/backup-worker-app.mjs

@@ -0,0 +1,70 @@
+// @ts-check
+// Metrics must be initialized before importing anything else
+import '@overleaf/metrics/initialize.js'
+import http from 'node:http'
+import { fileURLToPath } from 'node:url'
+import { promisify } from 'node:util'
+import express from 'express'
+import logger from '@overleaf/logger'
+import Metrics from '@overleaf/metrics'
+import { expressify } from '@overleaf/promise-utils'
+import { drainQueue, healthCheck } from './storage/scripts/backup_worker.mjs'
+const app = express()
+
+logger.initialize('history-v1-backup-worker')
+Metrics.open_sockets.monitor()
+Metrics.injectMetricsRoute(app)
+app.use(Metrics.http.monitor(logger))
+Metrics.leaked_sockets.monitor(logger)
+Metrics.event_loop.monitor(logger)
+Metrics.memory.monitor(logger)
+
+app.get('/status', (req, res) => {
+  res.send('history-v1-backup-worker is up')
+})
+
+app.get(
+  '/health_check',
+  expressify(async (req, res) => {
+    await healthCheck()
+    res.sendStatus(200)
+  })
+)
+
+app.use((err, req, res, next) => {
+  req.logger.addFields({ err })
+  req.logger.setLevel('error')
+  next(err)
+})
+
+async function triggerGracefulShutdown(server, signal) {
+  logger.warn({ signal }, 'graceful shutdown: started shutdown sequence')
+  await drainQueue()
+  server.close(function () {
+    logger.warn({ signal }, 'graceful shutdown: closed server')
+    setTimeout(() => {
+      process.exit(0)
+    }, 1000)
+  })
+}
+
+/**
+ * @param {number} port
+ * @return {Promise<http.Server>}
+ */
+export async function startApp(port) {
+  await healthCheck()
+  const server = http.createServer(app)
+  await promisify(server.listen.bind(server, port))()
+  const signals = ['SIGINT', 'SIGTERM']
+  signals.forEach(signal => {
+    process.on(signal, () => triggerGracefulShutdown(server, signal))
+  })
+  return server
+}
+
+// Run this if we're called directly
+if (process.argv[1] === fileURLToPath(import.meta.url)) {
+  const PORT = parseInt(process.env.PORT || '3103', 10)
+  await startApp(PORT)
+}

services/history-v1/package.json

@@ -20,6 +20,7 @@
     "basic-auth": "^2.0.1",
     "bluebird": "^3.7.2",
     "body-parser": "^1.20.3",
+    "bull": "^4.16.5",
     "bunyan": "^1.8.12",
     "check-types": "^11.1.2",
     "command-line-args": "^3.0.3",

services/history-v1/storage/scripts/backup_scheduler.mjs

@@ -0,0 +1,137 @@
+import Queue from 'bull'
+import config from 'config'
+import commandLineArgs from 'command-line-args'
+import logger from '@overleaf/logger'
+
+logger.initialize('backup-queue')
+
+// Use the same redis config as backup_worker
+const redisOptions = config.get('redis.queue')
+
+// Create a Bull queue named 'backup'
+const backupQueue = new Queue('backup', {
+  redis: redisOptions,
+  defaultJobOptions: {
+    removeOnComplete: true,
+    removeOnFail: true,
+  },
+})
+
+// Define command-line options
+const optionDefinitions = [
+  { name: 'clean', type: Boolean },
+  { name: 'status', type: Boolean },
+  { name: 'add', type: String, multiple: true },
+  { name: 'monitor', type: Boolean },
+]
+
+// Parse command line arguments
+const options = commandLineArgs(optionDefinitions)
+
+// Setup queue event listeners
+function setupMonitoring() {
+  console.log('Starting queue monitoring. Press Ctrl+C to exit.')
+
+  backupQueue.on('global:error', error => {
+    logger.info({ error }, 'Queue error')
+  })
+
+  backupQueue.on('global:waiting', jobId => {
+    logger.info({ jobId }, 'job is waiting')
+  })
+
+  backupQueue.on('global:active', jobId => {
+    logger.info({ jobId }, 'job is now active')
+  })
+
+  backupQueue.on('global:stalled', jobId => {
+    logger.info({ jobId }, 'job has stalled')
+  })
+
+  backupQueue.on('global:progress', (jobId, progress) => {
+    logger.info({ jobId, progress }, 'job progress')
+  })
+
+  backupQueue.on('global:completed', (jobId, result) => {
+    logger.info({ jobId, result }, 'job completed')
+  })
+
+  backupQueue.on('global:failed', (jobId, err) => {
+    logger.info({ jobId, err }, 'job failed')
+  })
+
+  backupQueue.on('global:paused', () => {
+    logger.info({}, 'Queue paused')
+  })
+
+  backupQueue.on('global:resumed', () => {
+    logger.info({}, 'Queue resumed')
+  })
+
+  backupQueue.on('global:cleaned', (jobs, type) => {
+    logger.info({ jobsCount: jobs.length, type }, 'Jobs cleaned')
+  })
+
+  backupQueue.on('global:drained', () => {
+    logger.info({}, 'Queue drained')
+  })
+
+  backupQueue.on('global:removed', jobId => {
+    logger.info({ jobId }, 'Job removed')
+  })
+}
+
+// Main execution block
+async function run() {
+  const optionCount = [
+    options.clean,
+    options.status,
+    options.add,
+    options.monitor,
+  ].filter(Boolean).length
+  if (optionCount > 1) {
+    console.error('Only one option can be specified')
+    process.exit(1)
+  }
+
+  if (options.clean) {
+    const beforeCounts = await backupQueue.getJobCounts()
+    console.log('Current queue state:', JSON.stringify(beforeCounts))
+    console.log('Cleaning completed and failed jobs...')
+    await backupQueue.clean(1, 'completed')
+    await backupQueue.clean(1, 'failed')
+    const afterCounts = await backupQueue.getJobCounts()
+    console.log('Current queue state:', JSON.stringify(afterCounts))
+    console.log('Queue cleaned successfully')
+  } else if (options.status) {
+    const counts = await backupQueue.getJobCounts()
+    console.log('Current queue state:', JSON.stringify(counts))
+  } else if (options.add) {
+    const projectIds = Array.isArray(options.add) ? options.add : [options.add]
+    for (const projectId of projectIds) {
+      const job = await backupQueue.add({ projectId }, { jobId: projectId })
+      console.log(`Added job for project: ${projectId}, job ID: ${job.id}`)
+    }
+  } else if (options.monitor) {
+    setupMonitoring()
+  } else {
+    console.log('Usage:')
+    console.log('  --clean              Clean up completed and failed jobs')
+    console.log('  --status             Show current job counts')
+    console.log('  --add [projectId]    Add a job for the specified projectId')
+    console.log('  --monitor            Monitor queue events')
+  }
+}
+
+// Run and handle errors
+run()
+  .catch(err => {
+    console.error('Error:', err)
+    process.exit(1)
+  })
+  .then(result => {
+    // Only exit if not in monitor mode
+    if (!options.monitor) {
+      process.exit(0)
+    }
+  })

services/history-v1/storage/scripts/backup_worker.mjs

@@ -0,0 +1,63 @@
+import Queue from 'bull'
+import logger from '@overleaf/logger'
+import config from 'config'
+import metrics from '@overleaf/metrics'
+
+const CONCURRENCY = 10
+const redisOptions = config.get('redis.queue')
+const TIME_BUCKETS = [10, 100, 500, 1000, 5000, 10000, 30000, 60000]
+
+// Create a Bull queue named 'backup'
+const backupQueue = new Queue('backup', {
+  redis: redisOptions,
+})
+
+// Log queue events
+backupQueue.on('active', job => {
+  logger.info({ job }, 'job  is now active')
+})
+
+backupQueue.on('completed', (job, result) => {
+  metrics.inc('backup_worker_job', 1, { status: 'completed' })
+  logger.info({ job, result }, 'job completed')
+})
+
+backupQueue.on('failed', (job, err) => {
+  metrics.inc('backup_worker_job', 1, { status: 'failed' })
+  logger.error({ job, err }, 'job failed')
+})
+
+backupQueue.on('waiting', jobId => {
+  logger.info({ jobId }, 'job is waiting')
+})
+
+backupQueue.on('error', error => {
+  logger.error({ error }, 'queue error')
+})
+
+// Process jobs
+backupQueue.process(CONCURRENCY, async job => {
+  const { projectId } = job.data
+  const timer = new metrics.Timer(
+    'backup_worker_job_duration',
+    1,
+    {},
+    TIME_BUCKETS
+  )
+  logger.info({ projectId }, 'processing backup for project')
+  await new Promise(resolve => setTimeout(resolve, 5000 + Math.random() * 5000))
+  timer.done()
+  return `backup completed ${projectId}`
+})
+
+export async function drainQueue() {
+  logger.info({ queue: backupQueue.name }, 'pausing queue')
+  await backupQueue.pause(true) // pause this worker and wait for jobs to finish
+  logger.info({ queue: backupQueue.name }, 'closing queue')
+  await backupQueue.close()
+}
+
+export async function healthCheck() {
+  const count = await backupQueue.count()
+  metrics.gauge('backup_worker_queue_length', count)
+}

services/history-v1/tsconfig.json

@@ -6,6 +6,7 @@
     "app/js/**/*",
     "backup-deletion-app.mjs",
     "backup-verifier-app.mjs",
+    "backup-worker-app.mjs",
     "benchmarks/**/*",
     "config/**/*",
     "migrations/**/*",
@@ -14,4 +15,4 @@
     "test/**/*",
     "types"
   ]
-}
+}