How to Monitor Tortoise with Vigilmon

Tortoise is an MQTT 3.1.1 client library for Elixir. It powers IoT telemetry ingestion in Nerves-based systems, connecting devices to brokers like EMQX, Mosquitto, and HiveMQ with support for QoS 0/1/2, persistent sessions, last will messages, and TLS. But MQTT connectivity failures are famously silent: a Tortoise.Connection process can report :connected in your supervisor tree while its underlying TCP socket has timed out, QoS 1 messages sit unacknowledged in a retry queue, and your IoT pipeline has quietly stopped delivering data.

This tutorial adds external observability to a Tortoise-powered application:

• A Phoenix health endpoint that probes live MQTT connectivity
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitoring to catch silent message-flow interruptions
• Alerting when the broker connection degrades
• Status page for your IoT infrastructure team

Why Monitor Tortoise?

| Signal | What it catches | |---|---| | Broker reachability | Network partitions, broker restarts, firewall changes | | Connection state | Tortoise.Connection reporting connected but TCP dead | | QoS 1/2 delivery | Messages stuck in unacknowledged retry queues | | Subscription health | Topics silently unsubscribed after reconnect | | Pipeline throughput | Message flow halted without supervisor-visible crash | | TLS certificate expiry | Broker cert renewal breaking client TLS handshake |

A standard HTTP uptime monitor tells you your Phoenix API is alive. It says nothing about whether the MQTT pipeline feeding it is still moving data. You need both layers.

Step 1: Add a Health Endpoint That Probes MQTT Connectivity

Build a health plug that checks Tortoise connection state and exercises the broker with a test publish.

# lib/my_app_web/plugs/mqtt_health.ex
defmodule MyAppWeb.Plugs.MqttHealth do
  import Plug.Conn

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health/mqtt"} = conn, _opts) do
    checks = %{
      connection: check_connection(),
      broker_ping: check_broker_ping()
    }

    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_connection do
    case Tortoise.Connection.status(MyApp.MqttClient) do
      {:ok, :connected} -> :ok
      _ -> :error
    end
  end

  defp check_broker_ping do
    # Publish to a dedicated health topic and verify no error
    topic = "health/ping/#{node()}"
    case Tortoise.publish(MyApp.MqttClient, topic, "ping", qos: 0) do
      :ok -> :ok
      {:ok, _ref} -> :ok
      _ -> :error
    end
  end
end

Register the plug in your endpoint before authentication middleware:

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

  plug MyAppWeb.Plugs.MqttHealth   # ← before authentication
  plug MyAppWeb.Router
end

Test it while Tortoise is connected:

curl http://localhost:4000/health/mqtt
# {"status":"ok","checks":{"connection":"ok","broker_ping":"ok"}}

Shut down EMQX or Mosquitto and recheck — you'll see the 503 Vigilmon needs to trigger an alert.

Step 2: Configure Tortoise with Reconnect Telemetry

Wrap your Tortoise connection in a GenServer that emits telemetry events on state changes:

# lib/my_app/mqtt_client.ex
defmodule MyApp.MqttClient do
  use GenServer
  require Logger

  @connection_name __MODULE__

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def init(_opts) do
    connect()
    {:ok, %{connected_at: nil, reconnect_count: 0}}
  end

  def connection_name, do: @connection_name

  defp connect do
    {:ok, _pid} = Tortoise.Supervisor.start_child(
      client_id: @connection_name,
      handler: {MyApp.MqttHandler, []},
      server: {Tortoise.Transport.Tcp, host: broker_host(), port: 1883},
      subscriptions: [
        {"telemetry/#", 1},
        {"devices/+/status", 1}
      ]
    )
  end

  defp broker_host do
    Application.get_env(:my_app, :mqtt_broker_host, "localhost")
  end
end

Define a handler module that emits telemetry on connect/disconnect:

# lib/my_app/mqtt_handler.ex
defmodule MyApp.MqttHandler do
  use Tortoise.Handler
  require Logger

  def init(_opts), do: {:ok, %{}}

  def connection(:up, state) do
    :telemetry.execute([:my_app, :mqtt, :connection], %{status: 1}, %{})
    Logger.info("MQTT connected")
    {:ok, state}
  end

  def connection(:down, state) do
    :telemetry.execute([:my_app, :mqtt, :connection], %{status: 0}, %{})
    Logger.warning("MQTT disconnected")
    {:ok, state}
  end

  def handle_message(topic, payload, state) do
    :telemetry.execute([:my_app, :mqtt, :message_received], %{count: 1}, %{topic: topic})
    # forward to your pipeline
    MyApp.TelemetryPipeline.ingest(topic, payload)
    {:ok, state}
  end

  def terminate(_reason, state), do: {:ok, state}
  def subscription(:up, _topic, state), do: {:ok, state}
  def subscription(:down, _topic, state), do: {:ok, state}
end

Step 3: Set Up HTTP Monitoring in Vigilmon

Point Vigilmon at the health endpoint:

1. Sign up at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health/mqtt
4. Set check interval: 1 minute (paid) or 5 minutes (free)
5. Add a keyword check: body must contain "status":"ok"
6. Save

The keyword check matters here because a broker restart can return HTTP 200 with "status":"degraded" — without it, Vigilmon would declare you healthy when MQTT is down.

Step 4: Heartbeat Monitoring for Message Flow

Your health endpoint confirms the connection is alive. It doesn't confirm messages are actually flowing. Use a Vigilmon heartbeat monitor to verify end-to-end pipeline throughput.

Add a heartbeat ping at the end of each successful pipeline ingestion batch:

# lib/my_app/telemetry_pipeline.ex
defmodule MyApp.TelemetryPipeline do
  require Logger

  @heartbeat_interval_ms 60_000

  def child_spec(_opts) do
    %{
      id: __MODULE__,
      start: {__MODULE__, :start_link, []},
      type: :worker
    }
  end

  def start_link do
    GenServer.start_link(__MODULE__, %{last_ping: nil, count: 0}, name: __MODULE__)
  end

  def ingest(topic, payload) do
    GenServer.cast(__MODULE__, {:ingest, topic, payload})
  end

  def handle_cast({:ingest, _topic, _payload}, state) do
    new_count = state.count + 1
    new_state =
      if should_ping?(state) do
        ping_heartbeat()
        %{state | count: new_count, last_ping: System.monotonic_time(:millisecond)}
      else
        %{state | count: new_count}
      end
    {:noreply, new_state}
  end

  defp should_ping?(%{last_ping: nil}), do: true
  defp should_ping?(%{last_ping: last}) do
    System.monotonic_time(:millisecond) - last >= @heartbeat_interval_ms
  end

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

Create the heartbeat monitor in Vigilmon:

1. Click New Monitor → Heartbeat
2. Set expected interval: 2 minutes (double your pipeline interval as buffer)
3. Copy the ping URL
4. Set as env variable: VIGILMON_MQTT_HEARTBEAT_URL

# config/runtime.exs
config :my_app, :vigilmon,
  mqtt_heartbeat_url: System.get_env("VIGILMON_MQTT_HEARTBEAT_URL")

If your pipeline stops receiving messages — broker partition, topic subscription drop, backpressure deadlock — Vigilmon alerts within the heartbeat window. This is the failure mode that kills IoT pipelines silently while supervisors report all-green.

Step 5: Alerts via Slack

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

Enable the channel on both monitors (HTTP + heartbeat). When your MQTT connection drops:

🔴 DOWN: yourdomain.com/health/mqtt
Body keyword "status":"ok" not found
Detected from: EU-West, US-East
2 minutes ago

When the heartbeat misses:

🔴 MISSED HEARTBEAT: MQTT Pipeline
Expected ping every 2 minutes
Last received: 7 minutes ago

Set up a separate notification channel for your on-call rotation and assign it to the heartbeat monitor — a missed heartbeat on an IoT pipeline is higher urgency than a web-tier blip.

Step 6: Status Page

Create a status page for your IoT platform team:

1. Go to Status Pages → New Status Page in Vigilmon
2. Add your MQTT HTTP monitor and heartbeat monitor
3. Label them descriptively: "MQTT Broker Connection" and "Telemetry Pipeline"
4. Copy the public URL and share it in your ops Slack channel

When a device fleet loses connectivity, your operations team can immediately distinguish broker issues (HTTP monitor down) from pipeline processing issues (heartbeat missed) without digging through logs.

What You've Built

| What | How | |------|-----| | MQTT health check | MqttHealth plug probing Tortoise.Connection.status/1 | | Broker connectivity | Live publish to health topic on every check | | HTTP uptime monitoring | Vigilmon HTTP monitor + keyword check | | Pipeline throughput monitoring | Vigilmon heartbeat on message ingestion batches | | Disconnect telemetry | Tortoise.Handler callbacks with :telemetry events | | Slack alerts | Vigilmon Slack notification channel | | Status page | Vigilmon public status page for ops team |

BEAM keeps Tortoise processes alive after broker restarts. Vigilmon tells you when that invisible reconnect loop has lasted six minutes and your IoT data has a gap.

Next Steps

• Add QoS 1 acknowledgement latency to your health response using a test publish/subscribe cycle
• Monitor per-topic message rates to detect device fleet failures (devices going silent) vs broker failures
• Use Vigilmon's response time history to correlate broker latency spikes with device firmware updates
• Add a separate heartbeat per Oban queue that processes MQTT-ingested telemetry

Get started free at vigilmon.online.