Monitoring #

When operating large-scale distributed systems in production, we often admire how RabbitMQ can run very quietly and stably in the background. This reliability sometimes lulls operations teams into complacency, ignoring one of the most important aspects of asynchronous system management: consistent monitoring. Problems in message delivery systems rarely appear suddenly and dramatically like synchronous REST APIs. Instead, RabbitMQ problems tend to creep slowly and invisibly at first: queues pile up little by little, producer confirm latency rises gradually, Erlang VM RAM memory approaches warning thresholds, or Dead Letter Queues (DLQs) silently hoard failed transaction messages.

Without a well-designed monitoring system, all those failure symptoms get buried in system asynchrony. And when problems are finally detected by developer teams because of user complaints, the system is usually already in a critical state—clusters frozen from memory exhaustion or all queues totally clogged. Therefore, monitoring isn’t merely optional or a post-deploy complement. Monitoring is an absolute foundation guaranteeing our message architecture’s long-term stability. We must make our asynchronous systems fully transparent, where every data movement can be measured, analyzed, and equipped with alerting systems ready to act before incidents occur.

Broker Health Pillars (Key SLI/SLO Indicators) #

To objectively measure RabbitMQ cluster health, we must map Service Level Indicators (SLIs) into several main dimensions. Here are the key metrics that must be monitored in real-time:

1. Queue Depth #

This is the most basic metric showing the number of messages settled inside queues. We must separate this metric into two parts:

• Messages Ready: The number of messages ready to be delivered to consumers. Increases in this number show consumer processing capacity (consume rate) is slower than message production speed (publish rate).
• Messages Unacked: The number of messages already delivered to consumers but not yet receiving ACK confirmations back. High unacked numbers indicate problems on consumer application sides (e.g., consumer thread crashes, deadlocks, or QoS prefetch limit values set too large).

2. Data Flow Ratios (Publish Rate vs Consume Rate) #

This ratio shows message traffic balance. In healthy system conditions, Publish Rate and Consume Rate curves must run side by side in parallel. If the Publish Rate consistently stays above the Consume Rate for a certain period, this is an early alarm that the system is heading toward backlog accumulation conditions.

3. Confirmation Latency (Publisher Confirm Latency) #

For applications demanding data reliability by enabling Quorum Queues and Publisher Confirms, monitoring confirmation latency is the best early detector. If the broker takes increasingly longer times (e.g., spiking from 5 milliseconds to 500 milliseconds) to send ACKs back to producers, this signals disk I/O congestion on broker servers or Raft consensus replication delays between cluster nodes.

4. Broker Resource Limits (Resource Watermark Alarms) #

The Erlang BEAM VM actively monitors RAM availability and disk storage space on servers. If RAM usage exceeds vm_memory_high_watermark or free disk space drops below critical thresholds:

• RabbitMQ immediately activates alarms and blocks all producer connections (connection blocked).
• We must install instant alarms for these RAM and disk usage metrics (e.g., if disk utilization > 80% or RAM usage > 90% of the watermark limit).

5. Connection and Channel Counts #

RabbitMQ TCP connection instantiation is an expensive process for broker CPUs. Good client applications must pool connections and use many channels within those connections. Sudden active connection count spikes usually indicate connection leaks on client application sides (applications creating new connections every time they send messages without closing old connections).

6. Dead Letter Queue Depth (DLQ Depth Buildup) #

DLQs must not be treated as ignored final dump sites. DLQ depth is a direct indicator of application code bugs, external API integration failures, or event schema contract damage. We must configure instant alerts if critical DLQ message counts are greater than zero (messages_ready > 0).

Modern Monitoring Architectures with Prometheus & Grafana #

The industry best practice for monitoring RabbitMQ is using a combination of Prometheus as a time-series metric collection engine and Grafana as a dashboard visualization portal.

Since version 3.8, RabbitMQ has included a native monitoring plugin named rabbitmq_prometheus in its distribution. This plugin eliminates the need for third-party exporters. When enabled, this plugin exposes a /metrics endpoint on HTTP port 15692 serving all broker internal metrics in an efficient standard Prometheus format.

flowchart LR
    subgraph Cluster["Production RabbitMQ Cluster"]
        Node1["Node 1 (Prometheus Plugin)"]
        Node2["Node 2 (Prometheus Plugin)"]
        Node3["Node 3 (Prometheus Plugin)"]
    end
    
    PrometheusServer["Prometheus Server"] -->|"Periodically Scrape /metrics"| Cluster
    PrometheusServer -->|"Store Time-Series Data"| PrometheusDB[("Prometheus DB")]
    
    Grafana["Grafana Dashboard"] -->|"Query Metrics"| PrometheusServer
    Grafana -->|"Show Visualizations & Trend Graphs"| AlertingSystem["Alerting Rules (PagerDuty / Slack)"]

    style Cluster stroke:#7b1fa2,stroke-width:2px
    style PrometheusServer stroke:#e65100,stroke-width:2px
    style Grafana stroke:#388e3c,stroke-width:2px

Some of the most crucial Prometheus queries (PromQL) we must put into production Grafana dashboards include:

• Monitoring Ready Messages:

  rabbitmq_queue_messages_ready{queue="queue.payment.main"}
  

• Monitoring Total Cluster Connections:

  sum(rabbitmq_connections)
  

• Detecting Broker Memory Alarms:

  rabbitmq_process_has_memory_alarm == 1
  

• Detecting Broker Disk Alarms:

  rabbitmq_node_has_disk_alarm == 1
  

Actionable Alerting Settings #

Beautiful Grafana visualization graphs are useless if operations teams are asleep when production clusters collapse. We must compose smart alerting rules that are actionable (demanding real actions, not just spam notifications triggering alert fatigue):

1. Critical Alert: DLQ Not Empty: If the queue.payment.dlq queue contains messages (messages > 0), immediately send high-priority notifications to the developer team’s Slack. This signals transactions totally failed and requiring immediate manual investigation.
2. Warning Alert: Queue Backlog Accumulation: If ready message counts on main queues increase linearly and constantly for more than 15 minutes, this signals our consumers are losing processing speed. Action: Perform scale out on consumer instances.
3. PagerDuty Alert: Broker Resource Alarm: If memory alarm or disk alarm metrics are 1, immediately contact the SRE team via PagerDuty to inspect Erlang memory allocations or clean broker server disk storage space.

Circuit Breaker Implementations in Go Applications #

Monitoring isn’t only done on the outside (infrastructure), but must also be actively integrated into our client application code. When RabbitMQ experiences resource exhaustion (RAM or disk), the broker publishes connection.blocked statuses to clients before unilaterally cutting network transmissions.

Good Go applications must listen to these blocking signals. When detecting the broker is blocked, Go producers must activate the Circuit Breaker pattern—temporarily stopping new message deliveries, storing new events in local memory buffers or elegantly returning failure responses to users, instead of letting goroutine threads pile up waiting for locked socket connections.

Here is complete Go code demonstrating how to listen for blocking signals from brokers and automatically perform circuit breaker handling:

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

const (
	amqpURI      = "amqp://admin:***@rabbitmq-cluster:5672/"
	exchangeName = "exchange.telemetry.direct"
	routingKey   = "telemetry.data"
)

type SafePublisher struct {
	conn          *amqp.Connection
	channel       *amqp.Channel
	mu            sync.RWMutex
	isBlocked     bool // Circuit breaker status flag
	blockedSignal chan amqp.ConnectionBlocked
	unblockSignal chan string
	closeSignal   chan *amqp.Error
}

func (sp *SafePublisher) Connect() error {
	var err error
	log.Println("Menghubungkan ke RabbitMQ...")
	sp.conn, err = amqp.Dial(amqpURI)
	if err != nil {
		return err
	}

	sp.channel, err = sp.conn.Channel()
	if err != nil {
		sp.conn.Close()
		return err
	}

	// 1. Register a listener to detect connection blocking by the broker (Watermark Alarms)
	sp.blockedSignal = make(chan amqp.ConnectionBlocked, 1)
	sp.unblockSignal = make(chan string, 1)
	sp.conn.NotifyBlocked(sp.blockedSignal, sp.unblockSignal)

	// 2. Register a listener to detect unexpected connection closures
	sp.closeSignal = make(chan *amqp.Error, 1)
	sp.channel.NotifyClose(sp.closeSignal)

	// Start the connection health status monitoring goroutine
	go sp.monitorConnection()

	return nil
}

func (sp *SafePublisher) monitorConnection() {
	for {
		select {
		case blockInfo := <-sp.blockedSignal:
			sp.mu.Lock()
			sp.isBlocked = true
			sp.mu.Unlock()
			log.Printf("[⚠️ ALARM] KONEKSI DIBLOKIR OLEH BROKER! Alasan: %s. Mengaktifkan Circuit Breaker...\n", blockInfo.Reason)
			
		case <-sp.unblockSignal:
			sp.mu.Lock()
			sp.isBlocked = false
			sp.mu.Unlock()
			log.Println("[✓ NORMAL] Koneksi telah dibebaskan oleh broker. Menonaktifkan Circuit Breaker.")
			
		case errClosed := <-sp.closeSignal:
			if errClosed != nil {
				log.Printf("[✗ ERROR] Koneksi channel ditutup secara paksa: %v\n", errClosed)
				return
			}
		}
	}
}

// PublishTelemetry sends data if the circuit breaker is in a normal status
func (sp *SafePublisher) PublishTelemetry(ctx context.Context, payload []byte) error {
	sp.mu.RLock()
	blocked := sp.isBlocked
	sp.mu.RUnlock()

	// If the circuit is in a blocked status, reject message deliveries immediately
	if blocked {
		return fmt.Errorf("pengiriman dibatalkan: broker sedang mengalami overload (Connection Blocked)")
	}

	return sp.channel.PublishWithContext(ctx,
		exchangeName,
		routingKey,
		false,
		false,
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Transient, // Use transient for low-latency telemetry
			Body:         payload,
		},
	)
}

func (sp *SafePublisher) Close() {
	if sp.channel != nil {
		sp.channel.Close()
	}
	if sp.conn != nil {
		sp.conn.Close()
	}
}

func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	publisher := &SafePublisher{}
	err := publisher.Connect()
	if err != nil {
		log.Fatalf("Koneksi awal ke broker gagal: %v\n", err)
	}
	defer publisher.Close()

	// Simulate periodic message delivery by producers
	go func() {
		ticker := time.NewTicker(1 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				payload := []byte(`{"sensor_id": "SN-098", "temperature": 27.5}`)
				
				publishCtx, pubCancel := context.WithTimeout(context.Background(), 2*time.Second)
				err := publisher.PublishTelemetry(publishCtx, payload)
				pubCancel()

				if err != nil {
					log.Printf("[KODE RESILIENCE] Gagal mengirim data sensor: %v. Mengalihkan ke penyimpanan lokal temporer...\n", err)
					// Here we could store telemetry data to local text files (local file buffers)
				} else {
					log.Println("[KIRIM] Berhasil mempublikasikan data sensor ke broker.")
				}
			}
		}
	}()

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, os.Interrupt, syscall.SIGTERM)

	<-stop
	log.Println("Menghentikan aplikasi produsen...")
	cancel()
	time.Sleep(1 * time.Second)
}

Monitoring Dimension Comparison #

To help us compose alerting thresholds at production levels, here is a RabbitMQ health metric handling guide table:

Summary #

• Visibility Is Stability — In the asynchronous world, the inability to monitor queues is the beginning of disaster. Always prioritize monitoring setup from the first development day.
• Use Native Prometheus — Enable the rabbitmq_prometheus plugin to get high-performance real-time metrics directly from the RabbitMQ core engine.
• Create Actionable Alerts — Avoid alert fatigue by configuring alarms demanding concrete troubleshooting actions, like instant notifications for DLQ status > 0.
• Apply Circuit Breakers — Install NotifyBlocked listeners on Go producer application sides so systems can gracefully stop message deliveries when brokers experience overload alarms.

← Previous: Message Size   Next: Message Contract →