didi-lot2-backend/backend/services/data-layer/didiQueue

didiQueue - RabbitMQ Message Queue Service 🐰

Overview

RabbitMQ message broker for asynchronous communication between the Orchestrator and Analysis Service in the DIDI Backend platform.

🎯 Purpose

Provides reliable message queuing for pipeline execution jobs, decoupling the API layer from the processing layer.

🚀 Quick Start

Start the Service

# From this directory
docker compose up -d

# Or from data-layer directory
make up-queue

Access Points

• AMQP Protocol: localhost:5672
• Management UI: http://localhost:15672
• Default Credentials: admin / rabbitmq123

📊 Queue Architecture

Since we're merging all analysis services into one unified service, we use a single queue:

Orchestrator → publishes → analysis_queue → consumed by → Analysis Service

Queue Configuration

• Queue Name: analysis_queue
• Type: Durable (survives restarts)
• Dead Letter Queue: analysis_dlq (for failed messages)
• Message TTL: 24 hours
• Auto-delete: No

🔧 Configuration

Environment Variables

Edit .env file to customize:

RABBITMQ_USER=admin
RABBITMQ_PASSWORD=rabbitmq123  # CHANGE IN PRODUCTION!
RABBITMQ_VHOST=/
RABBITMQ_PORT=5672
RABBITMQ_MGMT_PORT=15672

Resource Limits

Memory: 1GB (max) / 512MB (reserved)
CPU: 0.5 cores (max) / 0.25 cores (reserved)

📝 Message Format

Messages published to the queue follow this structure:

{
  "run_id": "analysis_abc123_20250901_120000",
  "pipeline_id": "uuid-here",
  "pipeline_version": 1,
  "input_data": {
    "text": "Content to analyze",
    "image": "base64_or_url",
    "audio": "url_to_audio",
    "video": "url_to_video"
  },
  "media_type": "text|image|audio|video",
  "created_at": "2025-09-01T12:00:00Z"
}

🔍 Management

View Queue Status

# Using Management UI
http://localhost:15672

# Using CLI
docker exec didi-queue rabbitmqctl list_queues

# Check queue depth
docker exec didi-queue rabbitmqctl list_queues name messages_ready messages_unacknowledged

Purge Queue (Development Only)

# Remove all messages from queue
docker exec didi-queue rabbitmqctl purge_queue analysis_queue

Health Check

# Check if RabbitMQ is responsive
docker exec didi-queue rabbitmq-diagnostics -q ping

# Detailed health check
docker exec didi-queue rabbitmq-diagnostics check_running

🏗️ Integration Points

Publishers (Orchestrator)

import aio_pika

# Connect
connection = await aio_pika.connect_robust(
    "amqp://admin:rabbitmq123@localhost:5672/"
)
channel = await connection.channel()

# Publish message
await channel.default_exchange.publish(
    aio_pika.Message(body=json.dumps(message).encode()),
    routing_key="analysis_queue"
)

Consumers (Analysis Service)

# Declare queue
queue = await channel.declare_queue("analysis_queue", durable=True)

# Consume messages
async for message in queue:
    async with message.process():
        body = json.loads(message.body.decode())
        # Process the message

🛠️ Troubleshooting

Queue is not created

The init-queues.sh script runs automatically on container start. Check logs:

docker logs didi-queue

Messages not being consumed

1. Check if Analysis Service is running
2. Verify queue has messages: docker exec didi-queue rabbitmqctl list_queues
3. Check for dead letter queue: docker exec didi-queue rabbitmqctl list_queues | grep dlq

High memory usage

# Check memory usage
docker exec didi-queue rabbitmq-diagnostics memory_breakdown

# Set memory limit
docker exec didi-queue rabbitmqctl set_vm_memory_high_watermark 0.4

🔐 Security

Production Checklist

• Change default password in .env
• Enable SSL/TLS for connections
• Restrict management UI access
• Set up user permissions
• Configure firewall rules
• Enable audit logging

Create Production User

# Create new user
docker exec didi-queue rabbitmqctl add_user analysis_service SECURE_PASSWORD

# Set permissions
docker exec didi-queue rabbitmqctl set_permissions -p / analysis_service ".*" ".*" ".*"

# Set user tags
docker exec didi-queue rabbitmqctl set_user_tags analysis_service monitoring

📊 Monitoring

Key Metrics

• Queue depth (messages waiting)
• Message rates (publish/consume)
• Connection count
• Memory usage
• Disk usage

Prometheus Metrics

RabbitMQ exposes metrics at: http://localhost:15692/metrics

🔄 Backup & Recovery

Backup

# Export definitions
docker exec didi-queue rabbitmqctl export_definitions /var/lib/rabbitmq/backup.json
docker cp didi-queue:/var/lib/rabbitmq/backup.json ./backup.json

Restore

# Import definitions
docker cp ./backup.json didi-queue:/var/lib/rabbitmq/backup.json
docker exec didi-queue rabbitmqctl import_definitions /var/lib/rabbitmq/backup.json

📚 Related Documentation

• Data Layer README
• RabbitMQ Documentation
• AMQP Protocol

---

Part of the DIDI Backend Data Layer