feat: add ops HTTP server with /healthz, /readyz, and Prometheus /metrics (#24)

* feat: add ops HTTP server with /healthz, /readyz, and Prometheus /metrics

Kubernetes has been operating this pod blind — no liveness or
readiness signal means a stuck consumer looks healthy from the
cluster's perspective. Observability beyond the structured log
stream was also missing, so dashboards and alerts had to be
log-derived.

- New minimal HTTP server (node:http, no framework) on HTTP_PORT
  (default 8080). Starts before RabbitMQ so /healthz is answerable
  during init and stops last during shutdown.
- /healthz: always 200 while the process is alive.
- /readyz: 200 only when RabbitMQ is connected AND not shutting
  down. Flips to 503 at the start of graceful shutdown so load
  balancers drain the pod before migrations stop.
- /metrics: Prometheus text exposition via prom-client. Includes
  default Node runtime metrics plus:
  - nextcloud_migration_active (gauge)
  - nextcloud_migration_started_total (counter)
  - nextcloud_migration_finished_total{outcome}
  - nextcloud_migration_files_total{outcome}
  - nextcloud_migration_file_transfer_duration_seconds (histogram)
  - nextcloud_migration_cloudery_token_total{outcome}
  - nextcloud_migration_rabbitmq_connected (gauge)
- Metrics live on a shared registry in src/runtime/metrics.ts so
  call sites (migration.ts, migration-runner.ts, cloudery-client.ts,
  index.ts) increment without threading a registry through every
  factory.
- HTTP_PORT config + validation, docs (configuration, operations,
  development) updated.

* refactor: lazy default metrics + pull-based active-migrations gauge

- collectDefaultMetrics() moved behind an enableDefaultMetrics()
  function called once from main(). Test runs that import metrics.ts
  no longer kick off the default runtime timers at module load.
- active-migrations is now a pull-model gauge: the runner holds the
  count, metrics.ts exposes bindActiveMigrationsSource() which main()
  wires once. The runner no longer imports the metrics module at
  all, so the coupling runs one direction only.
- Small clarifying note on the http-server test's ephemeral-port
  helper.

Author: rezk2ll

docs/configuration.md

@@ -64,6 +64,14 @@ Lower values give the Settings UI more frequent progress updates at the cost of
 FLUSH_INTERVAL="50"
 ```
 
+### `HTTP_PORT`
+
+TCP port the ops HTTP server binds on. Defaults to `8080`. The server exposes `/healthz`, `/readyz`, and `/metrics` — see [Operations](operations.md#health-and-metrics) for what each returns.
+
+```
+HTTP_PORT="8080"
+```
+
 ### `MAX_CONCURRENT_MIGRATIONS`
 
 Hard cap on the number of migrations the consumer will run at the same time. Defaults to `10`, matching the RabbitMQ prefetch.

docs/development.md

@@ -47,6 +47,8 @@ src/
   runtime/                  Process wiring. How domain + clients get started and fed.
     consumer.ts             Message handler — validation, idempotency, quota check, early ACK
     migration-runner.ts     Concurrency cap + in-flight tracking for graceful shutdown
+    http-server.ts          Ops HTTP server — /healthz, /readyz, /metrics
+    metrics.ts              Prometheus registry + metric definitions
     config.ts               Environment variable parsing + Config type
 
 test/

docs/operations.md

@@ -2,15 +2,41 @@
 
 Runbook for operating the service in production. Covers how to tell when something is wrong, what the service does automatically, and when a human needs to step in.
 
-## Health signal
+## Health and metrics
 
-The service has no HTTP health endpoint. What you can observe externally:
+The service exposes a small HTTP server on `HTTP_PORT` (default `8080`) with three endpoints:
+
+| Path | Purpose |
+|---|---|
+| `/healthz` | Liveness probe. Always returns 200 while the process is alive. Deliberately cheap — a stuck event loop can still return 200, which is the point of separating it from readiness. |
+| `/readyz` | Readiness probe. Returns 200 only when RabbitMQ is connected and the process is not shutting down. Flips to 503 as soon as SIGTERM is received so load balancers and `PodDisruptionBudget` logic stop routing to the pod before the drain begins. |
+| `/metrics` | Prometheus text exposition. Scrape interval of 15–30 s is plenty; the metrics below are counters and gauges, not high-cardinality. |
+
+In Kubernetes, point `livenessProbe` at `/healthz` and `readinessProbe` at `/readyz`. The default Helm chart values do this for you.
+
+### Metrics worth alerting on
+
+| Metric | Type | What it tells you |
+|---|---|---|
+| `nextcloud_migration_active` | gauge | How many migrations are in-flight right now. Sustained at the concurrency cap means you're queue-bound. |
+| `nextcloud_migration_started_total` | counter | Migration arrival rate. Alert on a drop to zero while the RabbitMQ queue has items — indicates the consumer has stopped picking work up. |
+| `nextcloud_migration_finished_total{outcome}` | counter | `outcome="completed"` vs `outcome="failed"`. Alert on a sustained failure ratio above your tolerance (e.g. > 10%). |
+| `nextcloud_migration_files_total{outcome}` | counter | Per-file outcomes (`transferred`, `skipped`, `failed`). `skipped` is normal during resumes; `failed` is not. |
+| `nextcloud_migration_file_transfer_duration_seconds` | histogram | Per-file transfer latency. A shift in the upper buckets is an early signal of Stack or Nextcloud slowdown. |
+| `nextcloud_migration_cloudery_token_total{outcome}` | counter | `outcome="success"` vs `"failed"`. A spike in failed tokens points at Cloudery, not at this service. |
+| `nextcloud_migration_rabbitmq_connected` | gauge | `1` when connected, `0` otherwise. Complements the `/readyz` probe for dashboards. |
+
+Default Node.js runtime metrics (event loop lag, heap, GC) are included automatically by `prom-client`.
+
+## Log-based signals
+
+Even with metrics, the event log remains the source of truth for per-migration diagnosis:
 
 - **Process uptime.** Log lines with `event: service.starting`, `service.shutting_down`, `service.stopped` delimit lifetimes.
-- **RabbitMQ connection.** An `event: rabbitmq.connected` on startup means the broker is reachable. Silent thereafter; connection drops surface as errors from the underlying library.
-- **Consumption rate.** Count `event: consumer.message_received` over time. A healthy instance emits one per incoming migration request.
+- **RabbitMQ connection.** An `event: rabbitmq.connected` on startup means the broker is reachable.
+- **Consumption rate.** Count `event: consumer.message_received` over time.
 
-If the process is alive and RabbitMQ is drained but the queue keeps growing, look at the concurrency cap next.
+If the process is alive, `/readyz` is green, but the queue keeps growing, look at the concurrency cap next.
 
 ## Events worth alerting on
 
@@ -74,8 +100,10 @@ See [Configuration](configuration.md) for the full list.
 
 On `SIGTERM` or `SIGINT`:
 
-1. RabbitMQ subscription closes immediately — no new messages are accepted.
-2. The process waits up to 60 seconds for in-flight migrations to finish naturally.
-3. If the deadline passes, the process exits anyway. Anything still running becomes a stale-running zombie and is reclaimed by the next consumer.
+1. `/readyz` flips to 503 immediately so load balancers stop routing to the pod.
+2. RabbitMQ subscription closes — no new messages are accepted.
+3. The process waits up to 60 seconds for in-flight migrations to finish naturally.
+4. The ops HTTP server closes.
+5. If the drain deadline passed, the process exits anyway. Anything still running becomes a stale-running zombie and is reclaimed by the next consumer.
 
 This matters for rolling deployments: if you expect migrations to routinely run longer than 60 seconds (most do), you'll see `service.stopped` with `drained: false` on every rollout. That's not a regression — it's the heartbeat recovery doing its job.

package-lock.json

@@ -14,7 +14,8 @@
   "dependencies": {
     "@linagora/rabbitmq-client": "^0.1.2",
     "cozy-stack-client": "^60.23.0",
-    "pino": "^9.6.0"
+    "pino": "^9.6.0",
+    "prom-client": "^15.1.3"
   },
   "devDependencies": {
     "@types/node": "^20.17.0",

package.json

@@ -1,5 +1,6 @@
 import type { Logger } from 'pino'
 import { MIGRATION_TOKEN_SCOPE } from '../domain/doctypes.js'
+import { clouderyTokenRequests } from '../runtime/metrics.js'
 
 export interface ClouderyClient {
   /** Returns a Stack token for the given instance, using the cache when fresh. */
@@ -127,7 +128,9 @@ export function createClouderyClient(
     let lastError: unknown
     for (let attempt = 0; attempt < MAX_ATTEMPTS; attempt++) {
       try {
-        return await fetchOnce(workplaceFqdn)
+        const token = await fetchOnce(workplaceFqdn)
+        clouderyTokenRequests.inc({ outcome: 'success' })
+        return token
       } catch (error) {
         lastError = error
         if (!isRetryable(error) || attempt === MAX_ATTEMPTS - 1) break
@@ -143,6 +146,7 @@ export function createClouderyClient(
         await new Promise<void>((resolve) => setTimeout(resolve, delay))
       }
     }
+    clouderyTokenRequests.inc({ outcome: 'failed' })
     logger.error({
       event: 'cloudery.token_failed',
       instance: workplaceFqdn,

src/clients/cloudery-client.ts

@@ -2,6 +2,12 @@ import type { Logger } from 'pino'
 import type { CozyDir, StackClient } from '../clients/stack-client.js'
 import type { MigrationCommand } from './types.js'
 import { getErrorMessage } from './errors.js'
+import {
+  fileTransferDuration,
+  filesProcessed,
+  migrationsFinished,
+  migrationsStarted,
+} from '../runtime/metrics.js'
 import {
   setRunning,
   flushProgress,
@@ -150,6 +156,9 @@ async function handleFileEntry(
   try {
     const fileStart = Date.now()
     const file = await ctx.stackClient.transferFile(accountId, entry.path, cozyDirId)
+    const durationMs = Date.now() - fileStart
+    fileTransferDuration.observe(durationMs / 1000)
+    filesProcessed.inc({ outcome: 'transferred' })
     ctx.transferred.bytes += file.size
     ctx.transferred.files += 1
     ctx.pending.bytesImported += file.size
@@ -160,7 +169,7 @@ async function handleFileEntry(
       event: 'migration.file_transferred',
       nc_path: entry.path,
       size: file.size,
-      duration_ms: Date.now() - fileStart,
+      duration_ms: durationMs,
       transferred_bytes: ctx.transferred.bytes,
       transferred_files: ctx.transferred.files,
       discovered_bytes: ctx.discovered.bytesTotal,
@@ -175,6 +184,7 @@ async function handleFileEntry(
     }
   } catch (error) {
     if (isConflictError(error)) {
+      filesProcessed.inc({ outcome: 'skipped' })
       ctx.totalSkipped += 1
       ctx.logger.info({
         event: 'migration.file_skipped',
@@ -187,6 +197,7 @@ async function handleFileEntry(
       ctx.pending.skipped.push({ path: entry.path, reason: 'already exists', size: entry.size })
       return
     }
+    filesProcessed.inc({ outcome: 'failed' })
     ctx.totalErrors += 1
     const message = getErrorMessage(error)
     ctx.logger.error({
@@ -253,6 +264,7 @@ export async function runMigration(
     startedAt: Date.now(),
   }
 
+  migrationsStarted.inc()
   try {
     migrationLogger.info({ event: 'migration.started' }, 'Migration started')
 
@@ -267,6 +279,7 @@ export async function runMigration(
       ctx.discovered.filesTotal,
     )
 
+    migrationsFinished.inc({ outcome: 'completed' })
     migrationLogger.info({
       event: 'migration.completed',
       duration_ms: Date.now() - ctx.startedAt,
@@ -279,6 +292,7 @@ export async function runMigration(
     }, 'Migration completed')
   } catch (error) {
     const message = getErrorMessage(error)
+    migrationsFinished.inc({ outcome: 'failed' })
     migrationLogger.error({
       event: 'migration.failed',
       duration_ms: Date.now() - ctx.startedAt,

src/domain/migration.ts

@@ -4,6 +4,12 @@ import { loadConfig } from './runtime/config.js'
 import { createClouderyClient } from './clients/cloudery-client.js'
 import { handleMigrationMessage } from './runtime/consumer.js'
 import { createMigrationRunner } from './runtime/migration-runner.js'
+import { createOpsServer } from './runtime/http-server.js'
+import {
+  bindActiveMigrationsSource,
+  enableDefaultMetrics,
+  rabbitmqConnected,
+} from './runtime/metrics.js'
 import { parseMigrationCommand } from './domain/types.js'
 
 const EXCHANGE = 'migration'
@@ -24,8 +30,10 @@ async function main(): Promise<void> {
 
   logger.info({ event: 'service.starting' }, 'Starting Nextcloud migration service')
 
+  enableDefaultMetrics()
   const clouderyClient = createClouderyClient(config.clouderyUrl, config.clouderyToken, logger)
   const migrationRunner = createMigrationRunner(config.maxConcurrentMigrations, logger)
+  bindActiveMigrationsSource(() => migrationRunner.active)
 
   const rabbitClient = new RabbitMQClient({
     url: config.rabbitmqUrl,
@@ -35,7 +43,22 @@ async function main(): Promise<void> {
     logger,
   })
 
+  let shuttingDown = false
+  const opsServer = createOpsServer(
+    config.httpPort,
+    {
+      isRabbitMQConnected: () => rabbitClient.isConnected(),
+      isShuttingDown: () => shuttingDown,
+    },
+    logger,
+  )
+  // Start the ops server BEFORE RabbitMQ so /healthz is answerable
+  // while init runs — Kubernetes sees a live pod even if broker
+  // connection takes a moment.
+  await opsServer.start()
+
   await rabbitClient.init()
+  rabbitmqConnected.set(1)
   logger.info({ event: 'rabbitmq.connected' }, 'Connected to RabbitMQ')
 
   await rabbitClient.subscribe(
@@ -66,7 +89,6 @@ async function main(): Promise<void> {
     routing_key: ROUTING_KEY,
   }, 'Subscribed to migration queue')
 
-  let shuttingDown = false
   const shutdown = async (signal: string) => {
     if (shuttingDown) return
     shuttingDown = true
@@ -76,7 +98,9 @@ async function main(): Promise<void> {
       active_migrations: migrationRunner.active,
     }, 'Shutting down')
     await rabbitClient.close()
+    rabbitmqConnected.set(0)
     const drained = await migrationRunner.drain(SHUTDOWN_DRAIN_MS)
+    await opsServer.stop()
     logger.info({
       event: 'service.stopped',
       drained,

src/index.ts

@@ -11,6 +11,8 @@ export interface Config {
   /** Maximum number of migrations allowed to run concurrently. Defaults
    * to 10 to match the RabbitMQ prefetch. */
   maxConcurrentMigrations: number
+  /** TCP port the ops HTTP server (probes + /metrics) binds on. */
+  httpPort: number
 }
 
 function requireEnv(name: string): string {
@@ -39,6 +41,12 @@ export function loadConfig(): Config {
       `MAX_CONCURRENT_MIGRATIONS must be a positive integer, got: ${process.env.MAX_CONCURRENT_MIGRATIONS}`,
     )
   }
+  const httpPort = parseInt(process.env.HTTP_PORT ?? '8080', 10)
+  if (!Number.isFinite(httpPort) || httpPort < 1 || httpPort > 65535) {
+    throw new Error(
+      `HTTP_PORT must be a TCP port (1-65535), got: ${process.env.HTTP_PORT}`,
+    )
+  }
   return {
     rabbitmqUrl: requireEnv('RABBITMQ_URL'),
     clouderyUrl: requireEnv('CLOUDERY_URL'),
@@ -47,5 +55,6 @@ export function loadConfig(): Config {
     flushInterval: parseInt(process.env.FLUSH_INTERVAL ?? '25', 10),
     stackUrlScheme: rawScheme,
     maxConcurrentMigrations: maxConcurrent,
+    httpPort,
   }
 }