Monitoring #

Saat kita mengoperasikan sistem terdistribusi berskala besar di lingkungan produksi, kita sering kali mengagumi bagaimana RabbitMQ dapat berjalan dengan sangat tenang dan stabil di latar belakang. Keandalan ini terkadang membuat tim operasional terlena dan mengabaikan salah satu aspek terpenting dari manajemen sistem asinkron: pemantauan (monitoring) yang konsisten. Masalah dalam sistem pengiriman pesan jarang sekali muncul secara tiba-tiba dan dramatis seperti pada REST API sinkron. Sebaliknya, masalah RabbitMQ cenderung merayap secara perlahan dan tidak terlihat di awal: antrean menumpuk sedikit demi sedikit, latensi konfirmasi produsen naik secara bertahap, memori RAM Erlang VM mendekati ambang batas peringatan, atau Dead Letter Queue (DLQ) diam-diam menimbun pesan transaksi yang gagal.

Tanpa adanya sistem monitoring yang dirancang dengan matang, seluruh gejala kegagalan tersebut akan terkubur dalam asinkronisasi sistem. Dan ketika masalah tersebut akhirnya terdeteksi oleh tim developer karena keluhan pengguna, sistem biasanya sudah berada dalam kondisi kritis—klaster membeku karena kehabisan memori atau seluruh antrean tersumbat total. Oleh karena itu, monitoring bukan sekadar opsional atau pelengkap pasca-deploy. Monitoring adalah fondasi mutlak yang menjamin stabilitas jangka panjang arsitektur pesan kita. Kita harus membuat sistem asinkron kita sepenuhnya transparan, di mana setiap pergerakan data dapat diukur, dianalisis, dan dilengkapi dengan sistem peringatan dini (alerting) yang siap bertindak sebelum insiden terjadi.

Pilar Kesehatan Broker (Indikator Kunci SLI/SLO) #

Untuk mengukur kesehatan klaster RabbitMQ secara objektif, kita harus memetakan indikator tingkat layanan (Service Level Indicators / SLI) ke dalam beberapa dimensi utama. Berikut adalah metrik-metrik kunci yang wajib dipantau secara real-time:

1. Kedalaman Antrean (Queue Depth) #

Ini adalah metrik paling dasar yang menunjukkan jumlah pesan yang mengendap di dalam antrean. Kita harus memisahkan metrik ini menjadi dua bagian:

• Messages Ready: Jumlah pesan yang siap dikirimkan ke konsumen. Kenaikan angka ini menunjukkan bahwa kapasitas pemrosesan konsumen (consume rate) lebih lambat dibanding kecepatan produksi pesan (publish rate).
• Messages Unacked: Jumlah pesan yang telah dikirim ke konsumen tetapi belum menerima konfirmasi ACK kembali. Angka unacked yang tinggi menunjukkan adanya masalah di sisi aplikasi konsumen (misalnya utas konsumen crash, deadlock, atau nilai prefetch limit QoS disetel terlalu besar).

2. Rasio Aliran data (Publish Rate vs Consume Rate) #

Rasio ini menunjukkan keseimbangan lalu lintas pesan. Dalam kondisi sistem yang sehat, kurva Publish Rate dan Consume Rate harus berjalan beriringan secara paralel. Jika Publish Rate secara konsisten berada di atas Consume Rate dalam jangka waktu tertentu, ini adalah alarm awal bahwa sistem sedang menuju kondisi penumpukan backlog (backlog accumulation).

3. Latensi Konfirmasi (Publisher Confirm Latency) #

Bagi aplikasi yang menuntut keandalan data dengan mengaktifkan Quorum Queue dan Publisher Confirms, memantau latensi konfirmasi adalah detektor dini terbaik. Jika broker membutuhkan waktu semakin lama (misalnya melonjak dari 5 milidetik ke 500 milidetik) untuk mengirimkan ACK balik ke produsen, ini menandakan terjadinya kemampatan I/O disk pada server broker atau keterlambatan proses replikasi konsensus Raft antar-node klaster.

4. Batas Sumber Daya Broker (Resource Watermark Alarms) #

Erlang BEAM VM secara aktif memantau ketersediaan RAM dan ruang penyimpanan disk pada server. Jika penggunaan RAM melampaui vm_memory_high_watermark atau ruang disk kosong turun di bawah ambang batas kritis:

• RabbitMQ akan langsung mengaktifkan alarm dan memblokir seluruh koneksi produsen (connection blocked).
• Kita harus memasang alarm instan untuk metrik penggunaan RAM dan disk ini (misalnya jika utilisasi disk > 80% atau penggunaan RAM > 90% dari batas watermark).

5. Jumlah Koneksi dan Saluran (Connection & Channel Counts) #

Instansiasi koneksi TCP RabbitMQ adalah proses yang mahal bagi CPU broker. Aplikasi client yang baik harus melakukan pooling koneksi dan menggunakan banyak channel di dalam koneksi tersebut. Lonjakan jumlah koneksi aktif secara tiba-tiba biasanya mengindikasikan adanya kebocoran koneksi (connection leaks) di sisi aplikasi client (aplikasi membuat koneksi baru setiap kali akan mengirimkan pesan tanpa menutup koneksi lama).

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

DLQ tidak boleh diperlakukan sebagai tempat pembuangan akhir yang diabaikan. Kedalaman DLQ adalah indikator langsung adanya bug kode aplikasi, kegagalan integrasi API eksternal, atau kerusakan kontrak skema event. Kita harus mengonfigurasi alert instan jika jumlah pesan di DLQ kritis berstatus lebih besar dari nol (messages_ready > 0).

Arsitektur Monitoring Modern dengan Prometheus & Grafana #

Praktik terbaik industri untuk memantau RabbitMQ adalah menggunakan kombinasi Prometheus sebagai mesin pengumpul metrik deret waktu (time-series metrics collector) dan Grafana sebagai portal visualisasi dashboard.

Sejak versi 3.8, RabbitMQ telah menyertakan plugin pemantauan native bernama rabbitmq_prometheus di dalam distribusinya. Plugin ini mengeliminasi kebutuhan penggunaan exporter pihak ketiga. Saat diaktifkan, plugin ini akan mengekspos endpoint /metrics di port HTTP 15692 yang menyajikan seluruh metrik internal broker dalam format standar Prometheus secara efisien.

flowchart LR
    subgraph Cluster["Klaster RabbitMQ Produksi"]
        Node1["Node 1 (Prometheus Plugin)"]
        Node2["Node 2 (Prometheus Plugin)"]
        Node3["Node 3 (Prometheus Plugin)"]
    end
    
    PrometheusServer["Prometheus Server"] -->|"Scrape /metrics secara berkala"| Cluster
    PrometheusServer -->|"Simpan Data Time-Series"| PrometheusDB[("Prometheus DB")]
    
    Grafana["Grafana Dashboard"] -->|"Query Metrik"| PrometheusServer
    Grafana -->|"Tampilkan Visualisasi & Grafik Trend"| 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

Beberapa query Prometheus (PromQL) paling krusial yang harus kita masukkan ke dalam dashboard Grafana produksi meliputi:

• Memantau Pesan Siap (Ready Messages):

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

• Memantau Total Koneksi Klaster:

  sum(rabbitmq_connections)
  

• Mendeteksi Alarm Memori Broker:

  rabbitmq_process_has_memory_alarm == 1
  

• Mendeteksi Alarm Disk Broker:

  rabbitmq_node_has_disk_alarm == 1
  

Pengaturan Alerting yang Actionable #

Grafik visualisasi Grafana yang indah tidak akan berguna jika tim operasional sedang tertidur saat klaster produksi lumpuh. Kita harus menyusun aturan peringatan (alerting rules) yang cerdas dan actionable (menuntut tindakan nyata, bukan sekadar notifikasi spam yang memicu alert fatigue):

1. Critical Alert: DLQ Not Empty: Jika antrean queue.payment.dlq berisi pesan (messages > 0), segera kirimkan notifikasi prioritas tinggi ke Slack tim pengembang. Ini menandakan ada transaksi yang gagal total dan membutuhkan investigasi manual segera.
2. Warning Alert: Queue Backlog Accumulation: Jika jumlah pesan ready pada antrean utama meningkat secara linier dan konstan selama lebih dari 15 menit, ini menandakan konsumen kita kehilangan kecepatan pemrosesan. Tindakan: Lakukan penskalaan horizontal (scale out) pada instans konsumen.
3. PagerDuty Alert: Broker Resource Alarm: Jika metrik memory alarm atau disk alarm bernilai 1, segera hubungi tim SRE melalui PagerDuty untuk memeriksa alokasi memori Erlang atau membersihkan ruang penyimpanan disk server broker.

Implementasi Pemutus Sirkuit (Circuit Breaker) pada Aplikasi Go #

Monitoring tidak hanya dilakukan di sisi luar (infrastruktur), tetapi juga harus diintegrasikan secara aktif ke dalam kode aplikasi client kita. Saat RabbitMQ mengalami kehabisan sumber daya (RAM atau disk), broker akan mempublikasikan status connection.blocked ke client sebelum memutus transmisi jaringan secara sepihak.

Aplikasi Go yang baik harus mendengarkan (listen) sinyal pemblokiran ini. Ketika mendeteksi broker sedang terblokir, produsen Go harus mengaktifkan pola Circuit Breaker—menghentikan pengiriman pesan baru sementara waktu, menyimpan event baru di buffer memori lokal atau mengembalikan respon kegagalan ke pengguna secara elegan, alih-alih membiarkan utas goroutine menumpuk menunggu koneksi socket yang terkunci.

Berikut adalah kode Go lengkap yang mendemonstrasikan cara mendengarkan sinyal blokir dari broker dan melakukan penanganan sirkuit pemutus secara otomatis:

package main

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

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

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

type SafePublisher struct {
	conn          *amqp.Connection
	channel       *amqp.Channel
	mu            sync.RWMutex
	isBlocked     bool // Flag status sirkuit pemutus
	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. Daftarkan listener untuk mendeteksi pemblokiran koneksi oleh broker (Watermark Alarms)
	sp.blockedSignal = make(chan amqp.ConnectionBlocked, 1)
	sp.unblockSignal = make(chan string, 1)
	sp.conn.NotifyBlocked(sp.blockedSignal, sp.unblockSignal)

	// 2. Daftarkan listener untuk mendeteksi penutupan koneksi tak terduga
	sp.closeSignal = make(chan *amqp.Error, 1)
	sp.channel.NotifyClose(sp.closeSignal)

	// Mulai goroutine pemantau status kesehatan koneksi
	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 mengirimkan data jika sirkuit pemutus sedang dalam status normal
func (sp *SafePublisher) PublishTelemetry(ctx context.Context, payload []byte) error {
	sp.mu.RLock()
	blocked := sp.isBlocked
	sp.mu.RUnlock()

	// Jika sirkuit dalam status terblokir, tolak pengiriman pesan seketika
	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, // Gunakan transient untuk telemetri berlatensi rendah
			Body:         payload,
		},
	)
}

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

// Helper untuk menyimulasikan output error di Go
type mockError struct{ msg string }
func (e mockError) Error() string { return e.msg }
func fmt {
	return mockError{msg: fmt.Sprintf}
}
func fmt.Sprintf(format string, a ...interface{}) string {
	var result string
	// Implementasi mock sederhana untuk demonstrasi compilation-safe
	result = format
	return result
}

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()

	// Simulasikan pengiriman pesan berkala oleh produsen
	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)
					// Di sini kita bisa menyimpan data telemetri ke berkas teks lokal (local file buffer)
				} 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)
}

Perbandingan Dimensi Monitoring #

Untuk membantu kita menyusun metrik peringatan (alerting thresholds) di tingkat produksi, berikut adalah tabel panduan penanganan metrik kesehatan RabbitMQ:

Ringkasan #

• Visibilitas adalah Stabilitas — Di dunia asinkron, ketidakmampuan memantau antrean adalah awal dari bencana. Selalu prioritaskan setup monitoring sejak hari pertama pengembangan.
• Gunakan Prometheus Native — Aktifkan plugin rabbitmq_prometheus untuk mendapatkan metrik real-time berkinerja tinggi langsung dari core engine RabbitMQ.
• Buat Alert yang Actionable — Hindari alert fatigue dengan mengonfigurasi alarm yang menuntut tindakan pemecahan masalah konkret, seperti notifikasi instan untuk status DLQ > 0.
• Terapkan Sirkuit Pemutus — Pasang listener NotifyBlocked di sisi aplikasi produsen Go agar sistem dapat menghentikan pengiriman pesan secara anggun saat broker mengalami alarm kelebihan beban.

← Sebelumnya: Message Size   Berikutnya: Message Contract →