tutorial

How to Monitor Msgpax with Vigilmon

Learn how to add health checks, MessagePack serialization observability, and alerts to Elixir applications using the Msgpax library with Vigilmon.

How to Monitor Msgpax with Vigilmon

Msgpax is a MessagePack implementation for Elixir that provides binary serialization more compact and faster than JSON. MessagePack encodes data in a binary format that can be 20–50% smaller than equivalent JSON, with significantly faster decode performance — making it ideal for high-throughput inter-service communication, cache storage, message queues, and WebSocket payloads where bandwidth and latency matter.

But binary serialization introduces failure modes that text-based formats don't have. A schema mismatch between a producer and consumer causes silent data corruption — a field that changed type decodes without error but produces wrong values. An oversized message that exceeds the unpacker's byte limit raises at decode time, not at send time. A MessagePack extension type that's registered in one service but not another causes opaque decode failures. These failures are invisible without monitoring.

This tutorial adds production observability to an Elixir application using Msgpax:

  • An application health check that verifies MessagePack pack/unpack
  • HTTP uptime monitoring with Vigilmon
  • Pack/unpack error rate and payload size tracking
  • Heartbeat monitoring for MessagePack-powered message workers
  • Alerts on serialization failures

Step 1: Add Msgpax to your project

# mix.exs
defp deps do
  [
    {:msgpax, "~> 2.4"},
    {:plug_cowboy, "~> 2.0"},
    {:plug, "~> 1.14"},
    {:jason, "~> 1.4"},   # still used for health check HTTP responses
    {:req, "~> 0.4"}
  ]
end

Basic usage:

# Pack a map to binary
{:ok, packed} = Msgpax.pack(%{"user_id" => 42, "action" => "purchase"})
# packed is a binary (iolist — call IO.iodata_to_binary/1 to get a plain binary)

# Unpack binary back to a map
{:ok, unpacked} = Msgpax.unpack(IO.iodata_to_binary(packed))
# unpacked => %{"user_id" => 42, "action" => "purchase"}

For structs, implement the Msgpax.Packer protocol:

# lib/my_app/schemas/event.ex
defmodule MyApp.Event do
  defstruct [:id, :type, :payload, :timestamp]

  defimpl Msgpax.Packer do
    def pack(%MyApp.Event{} = event, options) do
      Msgpax.Packer.Map.pack(%{
        "id" => event.id,
        "type" => event.type,
        "payload" => event.payload,
        "timestamp" => DateTime.to_unix(event.timestamp)
      }, options)
    end
  end
end

Step 2: Use Msgpax for inter-service messaging

# lib/my_app/messaging/event_publisher.ex
defmodule MyApp.Messaging.EventPublisher do
  require Logger

  @max_payload_bytes 1_048_576  # 1 MB limit

  def publish(event) do
    with {:ok, packed_iolist} <- Msgpax.pack(event),
         packed = IO.iodata_to_binary(packed_iolist),
         :ok <- check_payload_size(packed),
         :ok <- send_to_queue(packed) do
      MyApp.MsgpaxMetrics.record_pack(:ok, byte_size(packed))
      {:ok, byte_size(packed)}
    else
      {:error, :payload_too_large} = err ->
        MyApp.MsgpaxMetrics.record_pack(:error, 0)
        Logger.error("Event payload exceeds #{@max_payload_bytes} bytes")
        err
      {:error, reason} ->
        MyApp.MsgpaxMetrics.record_pack(:error, 0)
        Logger.error("Pack failed: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp check_payload_size(packed) when byte_size(packed) > @max_payload_bytes,
    do: {:error, :payload_too_large}
  defp check_payload_size(_packed), do: :ok

  defp send_to_queue(packed) do
    url = Application.get_env(:my_app, :queue)[:endpoint_url]
    case Req.post(url,
      body: packed,
      headers: [{"content-type", "application/msgpack"}]
    ) do
      {:ok, %{status: s}} when s in 200..299 -> :ok
      {:ok, %{status: status}} -> {:error, "queue rejected: #{status}"}
      {:error, reason} -> {:error, reason}
    end
  end
end

# lib/my_app/messaging/event_consumer.ex
defmodule MyApp.Messaging.EventConsumer do
  require Logger

  def consume(packed_binary) when is_binary(packed_binary) do
    case Msgpax.unpack(packed_binary) do
      {:ok, event_map} ->
        MyApp.MsgpaxMetrics.record_unpack(:ok, byte_size(packed_binary))
        process_event(event_map)
      {:error, reason} ->
        MyApp.MsgpaxMetrics.record_unpack(:error, byte_size(packed_binary))
        Logger.error("Unpack failed: #{inspect(reason)}")
        {:error, :unpack_failed}
    end
  end

  defp process_event(%{"type" => type} = event) do
    Logger.info("Processing event type=#{type}")
    # Dispatch to handlers...
    {:ok, event}
  end

  defp process_event(event) do
    Logger.warning("Unknown event shape: #{inspect(Map.keys(event))}")
    {:error, :unknown_schema}
  end
end

Step 3: Add a health check with serialization verification

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

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health"} = conn, _opts) do
    checks = %{
      database: check_database(),
      msgpax: check_msgpax(),
      memory: check_memory()
    }

    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,
      msgpax_stats: MyApp.MsgpaxMetrics.stats()
    }))
    |> 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, _} -> :error
    end
  rescue
    _ -> :error
  end

  defp check_msgpax do
    probe = %{"probe" => "health_check", "value" => 42}

    with {:ok, packed_iolist} <- Msgpax.pack(probe),
         packed = IO.iodata_to_binary(packed_iolist),
         {:ok, unpacked} <- Msgpax.unpack(packed),
         "health_check" <- Map.get(unpacked, "probe") do
      :ok
    else
      _ -> :error
    end
  rescue
    _ -> :error
  end

  defp check_memory do
    total = :erlang.memory(:total)
    if total < 2_147_483_648, do: :ok, else: :error
  end
end

Mount it before your router:

# 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:

curl http://localhost:4000/health
# {"status":"ok","checks":{"database":"ok","msgpax":"ok","memory":"ok"},"msgpax_stats":{"packs":840,"unpacks":840,"pack_errors":0,"unpack_errors":0,"avg_payload_bytes":128}}

Step 4: Track pack/unpack metrics and payload sizes

# lib/my_app/msgpax_metrics.ex
defmodule MyApp.MsgpaxMetrics do
  use Agent

  def start_link(_), do: Agent.start_link(fn ->
    %{
      packs: 0,
      unpacks: 0,
      pack_errors: 0,
      unpack_errors: 0,
      total_packed_bytes: 0,
      total_unpacked_bytes: 0
    }
  end, name: __MODULE__)

  def record_pack(:ok, bytes) do
    Agent.update(__MODULE__, fn s ->
      s
      |> Map.update!(:packs, &(&1 + 1))
      |> Map.update!(:total_packed_bytes, &(&1 + bytes))
    end)
  end

  def record_pack(:error, _bytes) do
    Agent.update(__MODULE__, fn s -> Map.update!(s, :pack_errors, &(&1 + 1)) end)
  end

  def record_unpack(:ok, bytes) do
    Agent.update(__MODULE__, fn s ->
      s
      |> Map.update!(:unpacks, &(&1 + 1))
      |> Map.update!(:total_unpacked_bytes, &(&1 + bytes))
    end)
  end

  def record_unpack(:error, _bytes) do
    Agent.update(__MODULE__, fn s -> Map.update!(s, :unpack_errors, &(&1 + 1)) end)
  end

  def stats do
    Agent.get(__MODULE__, fn s ->
      avg_pack_bytes =
        if s.packs > 0, do: div(s.total_packed_bytes, s.packs), else: 0

      avg_unpack_bytes =
        if s.unpacks > 0, do: div(s.total_unpacked_bytes, s.unpacks), else: 0

      s
      |> Map.put(:avg_payload_bytes, avg_pack_bytes)
      |> Map.put(:avg_received_bytes, avg_unpack_bytes)
    end)
  end
end

Add MyApp.MsgpaxMetrics to your supervision tree:

children = [
  MyApp.Repo,
  MyApp.MsgpaxMetrics,
  MyAppWeb.Endpoint
]

Step 5: Set up HTTP monitoring in Vigilmon

Point Vigilmon at your health endpoint:

  1. Sign up at vigilmon.online
  2. Click New Monitor → HTTP
  3. Enter https://yourdomain.com/health
  4. Set check interval: 1 minute (paid) or 5 minutes (free)
  5. Under Advanced, add a keyword check for "msgpax":"ok" to verify serialization health
  6. Save

Why external monitoring matters for MessagePack applications:

Binary serialization schema mismatches are especially hard to detect because the unpack call may succeed but silently produce wrong data — the error surfaces as application-level data corruption rather than a serialization exception. Vigilmon's external check detects the infrastructure symptoms: if a schema mismatch causes consumer workers to crash and miss heartbeats, or if the health endpoint degrades, Vigilmon alerts before the corruption reaches downstream systems.

Response time monitoring also catches payload size growth: as MessagePack payloads grow over time (more fields added to events, larger schemas), pack/unpack latency increases and contributes to slower endpoint response times. Vigilmon's response time history shows the trend before it becomes an outage.

Step 6: Heartbeat monitoring for MessagePack message workers

# lib/my_app/workers/message_processing_worker.ex
defmodule MyApp.Workers.MessageProcessingWorker do
  use GenServer

  require Logger

  @interval :timer.minutes(1)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  @impl true
  def init(state) do
    schedule_tick()
    {:ok, state}
  end

  @impl true
  def handle_info(:process, state) do
    case fetch_and_process_messages() do
      {:ok, count} when count > 0 ->
        Logger.debug("Processed #{count} MessagePack messages")
        ping_heartbeat()
      {:ok, 0} ->
        # No messages — still ping to confirm worker is alive
        ping_heartbeat()
      {:error, reason} ->
        Logger.error("Message processing failed: #{inspect(reason)}")
        # Don't ping — let Vigilmon detect the missed heartbeat
    end

    schedule_tick()
    {:noreply, state}
  end

  defp schedule_tick, do: Process.send_after(self(), :process, @interval)

  defp fetch_and_process_messages do
    messages = MyApp.Repo.all(MyApp.PendingMessage)

    results =
      Enum.map(messages, fn msg ->
        case MyApp.Messaging.EventConsumer.consume(msg.payload) do
          {:ok, _} ->
            MyApp.Repo.delete!(msg)
            :ok
          {:error, reason} ->
            Logger.error("Failed to process message #{msg.id}: #{inspect(reason)}")
            :error
        end
      end)

    success_count = Enum.count(results, &(&1 == :ok))
    {:ok, success_count}
  end

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

In Vigilmon:

  1. New Monitor → Heartbeat
  2. Set interval to 3 minutes (1-minute job with 2-minute grace)
  3. Copy the ping URL → set as VIGILMON_MESSAGE_WORKER_HEARTBEAT_URL env var
  4. Start the worker in your supervision tree

Add to supervision:

children = [
  MyApp.Repo,
  MyApp.MsgpaxMetrics,
  MyApp.Workers.MessageProcessingWorker,
  MyAppWeb.Endpoint
]

Step 7: Alerts via Slack

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

Enable the channel on all monitors. Msgpax-related incidents often look like:

  • HTTP health check degraded (pack/unpack probe failed — often caused by a library version mismatch or corrupt message in the queue)
  • Response time spike (average payload size grew as event schemas expanded)
  • Message worker heartbeat missed (unpack errors caused the worker to stop processing)

Configure Vigilmon to send all three alert types to your on-call channel so binary serialization failures are caught before they corrupt downstream state.

Step 8: Status page

  1. Status Pages → New Status Page in Vigilmon
  2. Add your HTTP health monitor and all heartbeat monitors
  3. Share the URL in your README and runbook

README badge:

![Uptime](https://vigilmon.online/badge/your-monitor-slug)

What you've built

| What | How | |------|-----| | Health check plug | HealthCheck plug with Msgpax round-trip probe | | Serialization verification | Pack → unpack a probe map in health check | | Payload size tracking | MsgpaxMetrics Agent (bytes, averages, error rates) | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health with keyword check | | Slack failure alerts | Vigilmon Slack notification channel | | Message worker heartbeat | Heartbeat ping after each 1-minute processing cycle | | Status page | Vigilmon public status page |

Msgpax keeps your inter-service payloads compact. Vigilmon keeps your message pipeline healthy.

Next steps

  • Add schema version fields to your MessagePack payloads ("schema_version" => 2) so consumers can detect mismatches before attempting to decode, and alert via Vigilmon when version drift is detected
  • Use Vigilmon's response time history to track payload size trends — a growing average payload size is an early warning that your MessagePack messages are accumulating unneeded fields
  • Set a Vigilmon alert on the message worker heartbeat grace period to tighten it as your processing SLA improves — start with 3 minutes and reduce as the pipeline stabilizes
  • Add a dead-letter queue monitor: track messages that fail to unpack and alert via Vigilmon heartbeat when the dead-letter count exceeds a threshold, before manual intervention is needed

Get started free at vigilmon.online.

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles