rmq-healer 0.2.0

Abstract, extensible RabbitMQ error-queue remediation tool

Abstract, extensible RabbitMQ error-queue remediation.
Classify → remediate → audit — fully configurable, no code required for common cases.

---

What Is It?

RMQ Healer watches RabbitMQ error queues and automatically remediates known failure patterns without touching your application code. You write YAML rules; the healer classifies every message and runs the matching workflow.

Core Features

	Feature
Zero-loss processing	Messages are acked only after a terminal action succeeds. If anything fails the message stays in the queue — untouched.
Declarative rules	YAML rule files with nested boolean logic: body_contains, json_path, header_equals, property_equals, and combinators (all/any/not).
Rich workflow engine	log, mutate_json, set_headers, python_script, republish, quarantine, route_unknown, drop.
Jinja2 templating	Every action param is a Jinja2 template — reference headers, body fields, script outputs, queue config.
Python script plugins	Drop any Python script in; it receives full message context on stdin and returns a JSON decision on stdout.
Idempotency guard	Bump-and-quarantine: after N failed runs the message is quarantined automatically.
Structured audit log	JSON-lines output — pipe straight to Fluentd, Loki, or CloudWatch.
Non-destructive peek	Inspect queue contents without consuming — broker requeues automatically after peek.
Graceful shutdown	SIGTERM/SIGINT safe — finishes the current message before stopping.
Developer GUI	PyQt6 desktop app for editing configs, viewing chains, peeking live queues, and browsing audit logs.
Plugin architecture	Register custom actions via Python entry_points or a local plugins/ directory.
Prometheus metrics	Optional prometheus_client integration — zero-overhead when not installed.
Dry-run with diffs	See exactly what mutate_json would change before committing.

---

Architecture

flowchart TD
    EQ[("Error Queue")]
    CL["Classifier\n(first match wins)"]
    WF["WorkflowRunner"]
    AU["AuditLogger\n(JSON-lines)"]

    EQ -->|basic_get| CL
    CL -->|matched rule| WF
    CL -->|no match| UQ[("Unknown Queue")]
    WF --> A1["log / set_headers\nmutate_json / python_script"]
    A1 --> T{"Terminal\naction"}
    T -->|republish| RQ[("Repaired Queue")]
    T -->|quarantine| Q[("Quarantine Queue")]
    T -->|route_unknown| UQ
    T -->|drop| DONE["Ack + discard"]
    WF --> AU
    CL --> AU

---

Quick Start

# 1. Install
pip install -e ".[dev]"

# 2. Set broker URL
export RMQ_URL="amqp://guest:guest@localhost/"

# 3. Generate config interactively
rmq-healer configure --output config/queues.yml

# 4. Validate
rmq-healer validate --config config/queues.yml

# 5. Test a rule against a sample message
rmq-healer test-rule --config config/queues.yml \
  --queue order-service.error --message examples/retry-timeout/sample.json

# 6. Dry-run with mutation diffs — nothing acked or published
rmq-healer dry-run --config config/queues.yml --all --pretty

# 7. Live run
rmq-healer run --config config/queues.yml --all --pretty

# 8. Peek without consuming
rmq-healer peek --config config/queues.yml --queue order-service.error --count 10

# 9. Visualise a workflow as a Mermaid diagram
rmq-healer workflow render --config config/queues.yml --queue order-service.error

# 10. Search audit logs
rmq-healer audit-search --log audit.jsonl --outcome quarantine

---

Example Configuration

rabbitmq:
  url: "${RMQ_URL}"
  prefetch: 10

idempotency:
  enabled: true
  max_handler_attempts: 3

queues:
  - name: order-service.error
    enabled: true
    rules_file: rules/order-errors.yml
    repaired_exchange: order.events
    repaired_routing_key: order.retry
    quarantine_exchange: ops.errors
    quarantine_routing_key: order.quarantine

# rules/order-errors.yml
rules:
  - id: transient_timeout
    match:
      any:
        - body_contains: "ETIMEDOUT"
        - json_path: $.error.code
          equals: "TIMEOUT"
    workflow:
      - action: set_headers
        params:
          headers:
            x-rmq-healer-rule: transient_timeout
      - action: republish
        params:
          exchange: "{{ queue.repaired_exchange }}"
          routing_key: "{{ queue.repaired_routing_key }}"

  - id: missing_customer_id
    match:
      all:
        - json_path: $.customer_id
          equals: null
    workflow:
      - action: python_script
        params:
          script: scripts/lookup_customer.py
          timeout: 30
          output:
            save_as: customer
      - action: mutate_json
        params:
          set:
            "$.customer_id": "{{ customer.id }}"
      - action: republish
        params:
          exchange: "{{ queue.repaired_exchange }}"
          routing_key: "{{ queue.repaired_routing_key }}"

See the examples/ directory for complete, runnable scenarios.

---

Documentation

Topic	Guide
Installation & setup	docs/installation.md
Configuration reference	docs/configuration.md
Writing rules	docs/rules.md
Python script actions	docs/python-scripts.md
CLI reference (all commands)	docs/cli-reference.md
Operations guide	docs/operations.md
Docker deployment	docs/docker.md
Developer GUI	docs/developer-gui.md
Plugin architecture	docs/plugins.md
Metrics & observability	docs/metrics.md
Contributing & tooling	docs/contributing.md
Troubleshooting	docs/troubleshooting.md

Browse all docs →

---

Developer Environment

# Windows
.\developer\scripts\setup.ps1      # start RabbitMQ + seed fixtures
.\developer\scripts\run-test.ps1   # dry-run
.\developer\scripts\launch-gui.ps1 # open GUI
.\developer\scripts\teardown.ps1   # stop

# Linux / macOS
bash developer/scripts/setup.sh
bash developer/scripts/run-test.sh
bash developer/scripts/launch-gui.sh
bash developer/scripts/teardown.sh

# Or cross-platform Python
python developer/scripts/dev_setup.py
python developer/scripts/run_test.py
python developer/scripts/launch_gui.py

---

License

MIT — see LICENSE.