Architecture Overview
rabbitmq-backup is a Rust-based CLI tool that non-destructively backs up and restores RabbitMQ messages and definitions to pluggable cloud storage. This page describes the high-level architecture, component relationships, and data flow.
Design Principles
- Non-destructive -- Messages remain in queues after backup. No broker shutdown, no filesystem access, no message loss.
- Deployment-agnostic -- Connects as an external AMQP client. Works with any RabbitMQ deployment (bare metal, Docker, Kubernetes, managed services).
- Multi-cloud storage -- Pluggable storage backends via the
object_storecrate. S3, Azure Blob, GCS, and local filesystem are supported identically. - Resumable -- SQLite checkpoints enable interrupted backups to resume from the last known offset.
- Point-in-Time Recovery -- Every message is timestamped at capture time (
backed_up_at), enabling time-window filtering during restore.
Design Heritage
This project follows the architecture established by osodevops/kafka-backup, adapting the segment-based storage format and metrics patterns from Kafka to the AMQP protocol.
Component Diagram
┌─────────────────────────────────────────────────────────────────────┐
│ rabbitmq-backup CLI │
│ (clap 4.x parser) │
├──────────┬──────────┬──────────┬──────────┬─────────┬───────────────┤
│ backup │ restore │ list │ describe │validate │ definitions- │
│ │ │ │ │ │ export/import │
└─────┬────┴─────┬────┴──────────┴──────────┴─────────┴───────┬───────┘
│ │ │
▼ ▼ ▼
┌───────────┐ ┌───────────┐ ┌────────────────┐
│ Backup │ │ Restore │ │ Definitions │
│ Engine │ │ Engine │ │ Exporter / │
│ │ │ │ │ Importer │
└─┬───┬───┬─┘ └──┬───┬───┘ └───────┬────────┘
│ │ │ │ │ │
│ │ │ │ │ │
▼ │ ▼ ▼ │ ▼
┌─────┤ ┌──────┐┌────┤ ┌───────────────┐
│Queue│ │Stream││Pub-│ │ Management │
│Read-│ │Read- ││lish│ │ Client │
│er │ │er ││er │ │ (reqwest) │
└──┬──┘ └──┬───┘└──┬─┘ └───────┬───────┘
│ │ │ │
▼ ▼ ▼ ▼
┌────────────┐ ┌───────────────────┐ ┌─────────────────┐
│ AMQP 0-9-1 │ │ Stream Protocol │ │ Management HTTP │
│ Client │ │ Client │ │ API │
│(amq-proto- │ │(rabbitmq-stream- │ │ (port 15672) │
│ col 10.x) │ │ client 0.5.x) │ └─────────────────┘
└──────┬─────┘ └────────┬──────────┘
│ │
▼ ▼
┌──────────────────────────────────┐
│ RabbitMQ Broker │
│ AMQP 0-9-1 (5672) │
│ Stream Protocol (5552) │
│ Management API (15672) │
└──────────────────────────────────┘
┌──────────────┐
│ Segment │
│ Writer / │
│ Reader │
└──────┬───────┘
│
▼
┌─────────────────────────┐
│ Storage Abstraction │
│ (object_store 0.11) │
├────┬────┬────┬──────────┤
│ S3 │Azr │GCS │Filesystem│
└────┴────┴────┴──────────┘
Crate Structure
The project is organized as a Cargo workspace with two crates:
rabbitmq-backup-core
The core library containing all business logic. It has no CLI dependencies and can be embedded in other Rust applications.
| Module | Responsibility |
|---|---|
backup/engine.rs | Backup orchestrator: queue discovery, parallel backup, manifest generation. |
backup/queue_reader.rs | AMQP 0-9-1 message reader with frame-level message assembly (Deliver + Header + Body). |
backup/stream_reader.rs | RabbitMQ Stream Protocol reader for x-queue-type: stream queues. |
restore/engine.rs | Restore orchestrator: manifest loading, PITR filtering, parallel publish. |
restore/publisher.rs | AMQP message publisher with publisher confirms. |
amqp/client.rs | Low-level AMQP 0-9-1 client: connection, channel, consume, cancel, QoS. |
amqp/channel_pool.rs | Channel pool for managing multiple AMQP channels. |
amqp/tls.rs | TLS configuration for AMQP connections (rustls). |
stream/client.rs | Stream Protocol client wrapper around rabbitmq-stream-client. |
definitions/management.rs | RabbitMQ Management API HTTP client. |
definitions/exporter.rs | Definitions export logic. |
definitions/importer.rs | Definitions import logic. |
definitions/types.rs | Type definitions for Management API responses. |
segment/writer.rs | RBAK segment writer: batching, compression, header/footer generation. |
segment/reader.rs | RBAK segment reader: decompression, CRC32 verification, record parsing. |
storage/config.rs | Storage backend configuration (tagged enum for S3/Azure/GCS/filesystem/memory). |
storage/backend.rs | Storage backend trait and implementations. |
manifest.rs | Backup manifest, queue metadata, segment metadata, and BackupRecord types. |
config.rs | Top-level configuration structures (serde YAML). |
compression.rs | zstd and LZ4 compression/decompression utilities. |
offset_store/ | SQLite checkpoint persistence for resumable backups. |
metrics/ | Prometheus metrics registry, labels, and HTTP server. |
circuit_breaker.rs | Transient failure handling with exponential backoff. |
health.rs | Health check types. |
error.rs | Structured error type with variants for each subsystem. |
rabbitmq-backup-cli
The binary crate providing the CLI interface. It depends on rabbitmq-backup-core and clap for argument parsing.
| Module | Responsibility |
|---|---|
main.rs | CLI entry point, subcommand dispatch, logging setup. |
commands/backup.rs | backup subcommand handler. |
commands/restore.rs | restore subcommand handler. |
commands/list.rs | list subcommand handler. |
commands/describe.rs | describe subcommand handler. |
commands/validate.rs | validate subcommand handler. |
commands/definitions_export.rs | definitions-export subcommand handler. |
commands/definitions_import.rs | definitions-import subcommand handler. |
Data Flow: Backup
RabbitMQ Broker rabbitmq-backup Object Storage
│ │ │
│ 1. GET /api/overview │ │
│◄──────────────────────────────│ │
│ cluster info │ │
│──────────────────────────────►│ │
│ │ │
│ 2. GET /api/definitions │ │
│◄──────────────────────────────│ │
│ definitions JSON │ │
│───────────────────── ─────────►│ 3. PUT definitions.json.zst │
│ │───────────────────────────────────►│
│ │ │
│ 4. GET /api/queues │ │
│◄──────────────────────────────│ │
│ queue list │ │
│──────────────────────────────►│ │
│ │ │
│ 5. basic.consume (per queue) │ │
│◄──────────────────────────────│ │
│ │ │
│ 6. Deliver + Header + Body │ │
│──────────────────────────────►│ 7. Add record to SegmentWriter │
│ ... (repeat for N msgs) │ ... (batch until threshold) │
│ │ │
│ │ 8. Finalize segment │
│ │ Compress + CRC32 + SHA-256 │
│ │───────────────────────────────────►│
│ │ PUT segment-NNNN.zst │
│ │ │
│ 9. basic.cancel │ │
│◄──────────────────────── ──────│ │
│ (all unacked msgs requeue) │ │
│ │ │
│ │ 10. PUT manifest.json │
│ │───────────────────────────────────►│
Step-by-Step
- Cluster info: Fetch the cluster name and RabbitMQ version from
GET /api/overview. - Definitions export: Fetch the full definitions from
GET /api/definitions, compress with zstd, and store. - Queue discovery: Fetch queue list from
GET /api/queues, apply include/exclude filters, partition into AMQP and stream queues. - Parallel queue backup: For each queue (up to
max_concurrent_queuesin parallel):- Open an AMQP connection and channel.
- Issue
basic.consumewithno_ack=false. - Receive Deliver + ContentHeader + ContentBody frames.
- Assemble complete messages in the
MessageAssemblerstate machine. - Add records to the
SegmentWriterbuffer. - When the buffer reaches
segment_max_bytesorsegment_max_interval_ms, finalize the segment (compress, add header/footer), upload to storage, and record metadata in the manifest.
- Cancel consumer: Issue
basic.cancel. The broker requeues all unacknowledged messages. - Finalize manifest: Write
manifest.jsonto storage with totals and checksums.
Data Flow: Restore
Object Storage rabbitmq-backup RabbitMQ Broker
│ │ │
│ 1. GET manifest.json │ │
│◄──────────────────────────────│ │
│ │ │
│ 2. GET definitions.json.zst │ │
│◄──────────────────────────────│ │
│ │ 3. POST /api/definitions │
│ │───────────────────────────────────►│
│ │ │
│ 4. GET segment-NNNN.zst │ │
│◄──────────────────────────────│ │
│ │ 5. Verify CRC32 + decompress │
│ │ 6. Apply PITR filter │
│ │ 7. basic.publish (per message) │
│ │───────────────────────────────────►│
│ │ 8. publisher confirm │
│ │◄───────────────────────────────────│
│ ... (repeat for all segs) │ │
Step-by-Step
- Load manifest: Read
manifest.jsonfrom storage. - Restore definitions: Download and decompress the definitions file, POST to the target Management API.
- Parallel queue restore: For each queue in the manifest (up to
max_concurrent_queues):- Download each segment file.
- Verify CRC32 integrity.
- Decompress and parse the length-prefixed JSON records.
- Apply PITR filtering: include only records where
backed_up_atfalls within[time_window_start, time_window_end]. - Apply queue/vhost/exchange remapping.
- Publish messages to the target broker in batches, with optional rate limiting.
- Wait for publisher confirms.
Technology Choices
| Technology | Crate | Version | Purpose |
|---|---|---|---|
| AMQP 0-9-1 codec | amq-protocol | 10.x | Wire-level protocol encoding/decoding. Provides frame types, method types, and property accessors without imposing a specific connection model. |
| Stream Protocol | rabbitmq-stream-client | 0.5.x | Native RabbitMQ Stream Protocol client for offset-based non-destructive reads from stream queues. |
| HTTP client | reqwest | 0.12 | Management API HTTP calls (definitions export/import, queue discovery, cluster overview). Uses rustls for TLS. |
| Object storage | object_store | 0.11 | Unified abstraction over AWS S3, Azure Blob Storage, Google Cloud Storage, and local filesystem. Handles authentication, retry, and streaming. |
| Checkpoint | sqlx | 0.8 | SQLite persistence for offset/checkpoint data, enabling resumable backups. |
| CLI | clap | 4.x | Command-line argument parsing with derive macros. |
| Async runtime | tokio | 1.x | Multi-threaded async runtime. Used for concurrent queue backup/restore, I/O, and timers. |
| Serialization | serde + serde_yaml + serde_json | 1.x / 0.9 / 1.x | Configuration (YAML), manifest and record (JSON), and storage format serialization. |
| Compression | zstd + lz4_flex | 0.13 / 0.11 | Segment payload compression. zstd is the default (good ratio at moderate CPU). LZ4 for maximum speed. |
| Integrity | crc32fast + sha2 | 1.4 / 0.10 | CRC32 for segment footer checksums. SHA-256 for manifest-level checksums. |
| Metrics | prometheus-client | 0.24 | Prometheus metrics collection with the prometheus-client crate (not the deprecated prometheus crate). Uses Family<Labels, Counter/Gauge> pattern. |
| HTTP server | hyper | 1.x | Lightweight HTTP server for the metrics /metrics and /health endpoints. |
| Logging | tracing + tracing-subscriber | 0.1 / 0.3 | Structured logging with environment filter support. |
| Error handling | thiserror | 2.x | Derive macros for the structured Error enum. |
| TLS | tokio-rustls + rustls | 0.26 / 0.23 | TLS support for AMQP connections. Supports CA certificates and mutual TLS (mTLS). |
Concurrency Model
rabbitmq-backup uses Tokio for async concurrency. The key concurrency patterns are:
-
Semaphore-based parallelism: A
tokio::sync::Semaphorelimits concurrent queue backups/restores tomax_concurrent_queues. Each queue runs as an independent Tokio task. -
Shared manifest: The
BackupManifestis wrapped inArc<Mutex<BackupManifest>>and shared across all queue backup tasks. Each task locks the manifest briefly to add segment metadata. -
Shutdown coordination: A
tokio::sync::broadcastchannel propagates shutdown signals (Ctrl+C) to all concurrent tasks, enabling graceful shutdown with segment flush. -
Per-queue connections: Each queue backup/restore task opens its own AMQP connection and channel, avoiding contention on a shared connection.
Error Handling Strategy
The crate uses a central Error enum with variants for each subsystem (AMQP, Stream, Storage, Config, etc.). The thiserror derive macro generates Display and Error implementations.
Key design decisions:
- Non-fatal queue errors: If a single queue fails during backup, the error is logged but other queues continue. The final manifest reports partial results.
- Non-fatal definitions errors: If definitions export/import fails, the backup/restore continues with a warning.
- Fatal config errors: Missing required configuration causes an immediate exit.
- Circuit breaker: The
circuit_breaker.rsmodule provides transient failure handling with exponential backoff for storage and connection operations.