How to Monitor AMQP with Vigilmon

The AMQP library is the idiomatic Elixir client for RabbitMQ, providing connection management, channel multiplexing, queue/exchange declaration, message publishing, and consumer setup — all integrated cleanly with OTP supervision trees.

RabbitMQ is reliable, but the connection between your Elixir app and the broker is not always. Network partitions drop channels, consumer acknowledgements can stall under memory pressure, queues can grow unbounded if consumers fall behind. BEAM lets it crash and restarts, but during the restart window messages pile up or get dropped.

This tutorial adds production observability to an Elixir + AMQP application:

• A health check endpoint that verifies the broker connection is alive
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitoring for consumer workers and publishers
• Slack alerts when the broker is unreachable
• A status page for your messaging infrastructure

Why Monitor AMQP Applications?

The AMQP library integrates with supervision trees, so crashed channels and connections restart automatically. But automatic restarts create monitoring blind spots:

• Consumer stall: a consumer crashes and its supervisor restarts it, but during restart messages accumulate in the queue. If restarts loop, the queue grows indefinitely — your supervisor is green but throughput is zero.
• Publisher failure: a publisher GenServer is alive but can't open a channel because the broker network interface is degraded. It logs errors and drops messages silently.
• Broker unreachable: the RabbitMQ node is running but your Elixir node lost network access to it. Supervisors restart workers endlessly, each restart consuming CPU while no actual work happens.

External monitoring — especially heartbeats on successful message consumption cycles — catches all three.

Step 1: Add a Health Check Endpoint

The health check should verify the AMQP connection is alive, not just that the GenServer is running.

# lib/my_app_web/plugs/health_check.ex
defmodule MyAppWeb.Plugs.HealthCheck do
  import Plug.Conn
  require Logger

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health"} = conn, _opts) do
    checks = %{
      database: check_database(),
      amqp: check_amqp_connection(),
      consumers: check_consumers()
    }

    status = if Enum.all?(checks, fn {_, v} -> v == :ok end), do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: checks
    }))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp check_database do
    case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
      {:ok, _} -> :ok
      _ -> :error
    end
  end

  defp check_amqp_connection do
    # Use a dedicated lightweight channel just for the health check
    case MyApp.AMQP.ConnectionPool.checkout() do
      {:ok, channel} ->
        result = case AMQP.Basic.qos(channel, prefetch_count: 1) do
          :ok -> :ok
          _ -> :error
        end
        MyApp.AMQP.ConnectionPool.checkin(channel)
        result
      _ ->
        :error
    end
  rescue
    _ -> :error
  end

  defp check_consumers do
    consumers = [
      MyApp.Consumers.OrderConsumer,
      MyApp.Consumers.NotificationConsumer
    ]

    all_alive = Enum.all?(consumers, fn mod ->
      case Process.whereis(mod) do
        nil -> false
        pid -> Process.alive?(pid)
      end
    end)

    if all_alive, do: :ok, else: :error
  end
end

Add it to your Phoenix endpoint:

# lib/my_app_web/endpoint.ex
defmodule MyAppWeb.Endpoint do
  use Phoenix.Endpoint, otp_app: :my_app

  plug MyAppWeb.Plugs.HealthCheck

  plug MyAppWeb.Router
end

Test it:

mix phx.server
curl http://localhost:4000/health
# {"status":"ok","checks":{"database":"ok","amqp":"ok","consumers":"ok"}}

Step 2: Set Up HTTP Monitoring in Vigilmon

1. Sign up at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health
4. Set the check interval to 1 minute
5. Save

Vigilmon probes from multiple regions — if your broker is regional (e.g. hosted in EU), multi-region monitoring can also catch BGP anomalies that affect only cross-region traffic.

Step 3: Heartbeat Monitoring for Consumers

The most important thing to monitor in an AMQP application is not whether the consumer process is alive — it's whether it's actually consuming messages successfully.

Consumer with Heartbeat Ping

# lib/my_app/consumers/order_consumer.ex
defmodule MyApp.Consumers.OrderConsumer do
  use GenServer
  use AMQP

  require Logger

  @exchange "orders"
  @queue "orders.process"
  @reconnect_interval 5_000

  def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)

  @impl true
  def init(_opts) do
    send(self(), :connect)
    {:ok, %{channel: nil, consumer_tag: nil, processed_count: 0}}
  end

  @impl true
  def handle_info(:connect, state) do
    case connect() do
      {:ok, channel} ->
        Process.monitor(channel.conn.pid)
        {:noreply, %{state | channel: channel}}
      {:error, reason} ->
        Logger.error("AMQP connect failed: #{inspect(reason)}")
        Process.send_after(self(), :connect, @reconnect_interval)
        {:noreply, state}
    end
  end

  def handle_info({:basic_deliver, payload, meta}, %{channel: channel} = state) do
    case process_message(payload) do
      :ok ->
        AMQP.Basic.ack(channel, meta.delivery_tag)
        new_count = state.processed_count + 1

        # Ping heartbeat every 100 messages
        if rem(new_count, 100) == 0, do: ping_heartbeat()

        {:noreply, %{state | processed_count: new_count}}

      {:error, reason} ->
        Logger.error("Message processing failed: #{inspect(reason)}")
        AMQP.Basic.nack(channel, meta.delivery_tag, requeue: false)
        {:noreply, state}
    end
  end

  def handle_info({:DOWN, _, :process, _pid, reason}, state) do
    Logger.warning("AMQP connection lost: #{inspect(reason)}, reconnecting...")
    Process.send_after(self(), :connect, @reconnect_interval)
    {:noreply, %{state | channel: nil}}
  end

  def handle_info({:basic_consume_ok, _meta}, state), do: {:noreply, state}
  def handle_info({:basic_cancel, _meta}, state), do: {:noreply, state}
  def handle_info({:basic_cancel_ok, _meta}, state), do: {:noreply, state}

  defp connect do
    amqp_url = Application.fetch_env!(:my_app, :amqp_url)
    with {:ok, conn} <- AMQP.Connection.open(amqp_url),
         {:ok, channel} <- AMQP.Channel.open(conn),
         :ok <- AMQP.Exchange.declare(channel, @exchange, :direct, durable: true),
         {:ok, _queue} <- AMQP.Queue.declare(channel, @queue, durable: true),
         :ok <- AMQP.Queue.bind(channel, @queue, @exchange),
         :ok <- AMQP.Basic.qos(channel, prefetch_count: 10),
         {:ok, _tag} <- AMQP.Basic.consume(channel, @queue) do
      {:ok, channel}
    end
  end

  defp process_message(payload) do
    payload
    |> Jason.decode()
    |> case do
      {:ok, data} -> MyApp.Orders.process(data)
      {:error, _} -> {:error, :invalid_json}
    end
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:order_consumer_heartbeat_url]
    if url do
      case Req.get(url, receive_timeout: 5_000) do
        {:ok, _} -> :ok
        {:error, reason} -> Logger.warning("Heartbeat failed: #{inspect(reason)}")
      end
    end
  end
end

For high-throughput consumers, pinging every N messages works well. For low-throughput consumers, use a periodic timer instead:

# Periodic heartbeat for low-throughput consumers
@heartbeat_interval :timer.minutes(5)

def handle_info(:heartbeat_tick, state) do
  if state.processed_since_last_tick > 0 do
    ping_heartbeat()
  end
  Process.send_after(self(), :heartbeat_tick, @heartbeat_interval)
  {:noreply, %{state | processed_since_last_tick: 0}}
end

Set the heartbeat URL in config:

# config/runtime.exs
config :my_app, :vigilmon,
  order_consumer_heartbeat_url: System.get_env("VIGILMON_ORDER_CONSUMER_HEARTBEAT_URL")

Create the Heartbeat Monitor

1. In Vigilmon, click New Monitor → Heartbeat
2. Set expected interval to match your consumer's ping cadence (e.g. 10 minutes if you process 100 messages every 10 minutes)
3. Copy the ping URL and set it as VIGILMON_ORDER_CONSUMER_HEARTBEAT_URL

Step 4: Publisher Heartbeat

For critical publishers, ping a heartbeat after each successful publish batch:

# lib/my_app/publishers/notification_publisher.ex
defmodule MyApp.Publishers.NotificationPublisher do
  use GenServer
  use AMQP

  require Logger

  @exchange "notifications"
  @flush_interval :timer.seconds(10)

  def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__)
  def publish(message), do: GenServer.cast(__MODULE__, {:publish, message})

  @impl true
  def init(_) do
    send(self(), :connect)
    Process.send_after(self(), :flush, @flush_interval)
    {:ok, %{channel: nil, buffer: []}}
  end

  @impl true
  def handle_cast({:publish, message}, state) do
    {:noreply, %{state | buffer: [message | state.buffer]}}
  end

  def handle_info(:flush, %{channel: channel, buffer: buffer} = state) when channel != nil and buffer != [] do
    Enum.each(buffer, fn message ->
      AMQP.Basic.publish(channel, @exchange, "", Jason.encode!(message), persistent: true)
    end)

    ping_heartbeat()
    Process.send_after(self(), :flush, @flush_interval)
    {:noreply, %{state | buffer: []}}
  end

  def handle_info(:flush, state) do
    Process.send_after(self(), :flush, @flush_interval)
    {:noreply, state}
  end

  def handle_info(:connect, state) do
    case connect() do
      {:ok, channel} -> {:noreply, %{state | channel: channel}}
      _ ->
        Process.send_after(self(), :connect, 5_000)
        {:noreply, state}
    end
  end

  defp connect do
    amqp_url = Application.fetch_env!(:my_app, :amqp_url)
    with {:ok, conn} <- AMQP.Connection.open(amqp_url),
         {:ok, channel} <- AMQP.Channel.open(conn),
         :ok <- AMQP.Exchange.declare(channel, @exchange, :fanout, durable: true) do
      {:ok, channel}
    end
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:notification_publisher_heartbeat_url]
    if url, do: Req.get(url, receive_timeout: 5_000)
  end
end

Step 5: Alerts via Slack

In Vigilmon, go to Notifications → New Channel → Slack and paste your webhook URL.

Assign the Slack channel to all monitors. You'll get targeted alerts:

🔴 DOWN: yourdomain.com/health
Checks degraded: amqp
Detected from: EU-West
1 minute ago

🔴 HEARTBEAT MISSED: order-consumer
Last ping: 15 minutes ago (expected: every 10 minutes)

Step 6: Status Page

1. Go to Status Pages → New Status Page in Vigilmon
2. Add your health endpoint monitor and all heartbeat monitors
3. Name each monitor descriptively: "API Health", "Order Consumer", "Notification Publisher"

Share the status page with your ops team so they have a single view of your messaging infrastructure.

Key Metrics for AMQP Applications

| What to Monitor | Why | |----------------|-----| | Health endpoint AMQP check | Broker connection alive | | Consumer heartbeat (every N messages or T minutes) | Messages are actually being processed | | Publisher heartbeat | Outbound messages are being sent | | Response time trend on health check | Slow channel checkout = broker pressure | | Heartbeat miss frequency | Consumer restart loops |

What You've Built

| What | How | |------|-----| | AMQP health check | Verifies live channel, not just process alive | | Consumer process check | All critical consumers running | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health | | Consumer heartbeat | Ping after every N messages or on timer | | Publisher heartbeat | Ping after each flush cycle | | Slack alerts | Per-monitor notifications | | Status page | Unified messaging infrastructure health view |

BEAM restarts your consumers when they crash. Vigilmon tells you when the crashes are looping.

Next Steps

• Add one heartbeat monitor per queue for fine-grained visibility into which queue is stalled
• Use Vigilmon's response time history to detect broker slowdowns before they cause consumer timeouts
• Alert on heartbeat miss rate — occasional misses are normal, a sustained pattern indicates a systemic issue

Get started free at vigilmon.online.