How to Monitor Commanded with Vigilmon

Commanded brings CQRS and Event Sourcing to Elixir — command dispatch, aggregate roots, event handlers, and process managers, all built on top of EventStore. These distributed, event-driven systems are robust by design, but they introduce failure modes that traditional monitoring can't see: a command that dispatches successfully but whose event handler crashes silently, or a projection that falls 10,000 events behind because a subscriber died.

This tutorial wires up observability for a Commanded application:

• Health checks that verify the command bus and event store
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitors for event handler liveness
• Projection lag alerting
• Slack alerts for command failures

Why Monitor Commanded?

| Failure mode | What breaks | |---|---| | EventStore connection lost | Commands dispatch but events are never persisted | | Event handler crash | Projections stop updating; read models go stale | | Projection lag | Read-side data is minutes/hours behind writes | | Aggregate version conflict | Commands rejected with optimistic concurrency errors | | Process manager timeout | Sagas stall mid-workflow |

None of these surface as HTTP 500s. Your API stays green while your domain logic silently fails.

Step 1: Add a Health Plug That Checks the Command Bus

Add an HTTP health endpoint that verifies Commanded's dependencies are reachable:

# lib/my_app_web/plugs/commanded_health.ex
defmodule MyAppWeb.Plugs.CommandedHealth do
  import Plug.Conn

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health/commanded"} = conn, _opts) do
    checks = %{
      event_store: check_event_store(),
      router: check_router()
    }

    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: status_label(status), checks: checks}))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp check_event_store do
    case EventStore.Config.parse(MyApp.EventStore.config()) do
      {:ok, _config} ->
        case Postgrex.query(MyApp.EventStore, "SELECT 1", []) do
          {:ok, _} -> :ok
          _ -> :error
        end
      _ -> :error
    end
  end

  defp check_router do
    # Dispatch a no-op command to verify router is running
    case MyApp.Router.dispatch(%MyApp.Commands.HealthCheck{}) do
      :ok -> :ok
      {:error, :no_handler} -> :ok  # Router is up, just no handler for health check
      _ -> :error
    end
  end

  defp status_label(200), do: "ok"
  defp status_label(_), do: "error"
end

Register in your endpoint:

# lib/my_app_web/endpoint.ex
plug MyAppWeb.Plugs.CommandedHealth
plug MyAppWeb.Router

Step 2: Add a No-op Health Check Command

Create a lightweight command that does nothing but lets you verify the command bus:

# lib/my_app/commands/health_check.ex
defmodule MyApp.Commands.HealthCheck do
  defstruct [:correlation_id]
end

# lib/my_app/aggregates/health.ex
defmodule MyApp.Aggregates.Health do
  defstruct [:id]

  def execute(%__MODULE__{}, %MyApp.Commands.HealthCheck{}) do
    [] # emit no events
  end

  def apply(health, _event), do: health
end

# In your router
defmodule MyApp.Router do
  use Commanded.Commands.Router

  dispatch MyApp.Commands.HealthCheck,
    to: MyApp.Aggregates.Health,
    identity: :correlation_id
end

Step 3: Add Vigilmon Uptime Monitoring

1. Log in to vigilmon.online and click Add Monitor.
2. Set Type to HTTP / HTTPS.
3. Enter https://yourapp.com/health/commanded.
4. Set Check interval to 1 minute.
5. Set Expected HTTP status to 200.
6. Add a Response body contains check: "status":"ok".
7. Click Save.

Step 4: Monitor Event Handler Liveness with Heartbeats

Event handlers are supervised processes that can crash and restart. A restart resets their subscription position, which may cause events to be replayed or skipped. Monitor handler liveness with a Vigilmon heartbeat:

# lib/my_app/event_handler_heartbeat.ex
defmodule MyApp.EventHandlerHeartbeat do
  use GenServer
  require Logger

  @interval :timer.minutes(1)
  @vigilmon_url "https://vigilmon.online/api/heartbeat/YOUR_HEARTBEAT_ID"

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

  def init(_) do
    schedule()
    {:ok, nil}
  end

  def handle_info(:ping, state) do
    handlers_alive = check_handlers()
    if handlers_alive do
      case HTTPoison.get(@vigilmon_url) do
        {:ok, %{status_code: 200}} -> :ok
        error -> Logger.warning("Vigilmon heartbeat failed: #{inspect(error)}")
      end
    else
      Logger.error("Event handlers unhealthy — skipping Vigilmon heartbeat")
    end
    schedule()
    {:noreply, state}
  end

  defp check_handlers do
    handlers = [
      MyApp.EventHandlers.OrderProjection,
      MyApp.EventHandlers.UserProjection,
      MyApp.EventHandlers.NotificationHandler
    ]
    Enum.all?(handlers, fn handler ->
      case Process.whereis(handler) do
        nil -> false
        pid -> Process.alive?(pid)
      end
    end)
  end

  defp schedule, do: Process.send_after(self(), :ping, @interval)
end

Add to your supervision tree in application.ex:

children = [
  # ... your other children
  MyApp.EventHandlerHeartbeat
]

Create the heartbeat monitor in Vigilmon:

1. Click Add Monitor → Cron / Heartbeat.
2. Set expected interval to 1 minute.
3. Set grace period to 2 minutes (allows one missed beat before alerting).
4. Copy the generated URL into @vigilmon_url.

Step 5: Alert on Projection Lag

Projection lag is a critical metric in CQRS systems — when your read models fall behind, users see stale data. Track lag using EventStore's stream position:

# lib/my_app/projection_lag_monitor.ex
defmodule MyApp.ProjectionLagMonitor do
  use GenServer
  require Logger

  @max_lag_events 100
  @check_interval :timer.minutes(2)

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

  def init(_), do: {:ok, nil, {:continue, :schedule}}

  def handle_continue(:schedule, state) do
    schedule()
    {:noreply, state}
  end

  def handle_info(:check, state) do
    check_projection_lag()
    schedule()
    {:noreply, state}
  end

  defp check_projection_lag do
    # Get the latest event number from EventStore
    {:ok, %{last_stream_version: head}} =
      EventStore.stream_info("$all")

    # Get each projection's last processed event
    projections = [
      {"orders", MyApp.Projections.OrderProjection},
      {"users", MyApp.Projections.UserProjection}
    ]

    Enum.each(projections, fn {name, projection} ->
      last_seen = projection.last_seen_event_number()
      lag = head - last_seen

      if lag > @max_lag_events do
        Logger.error("Projection #{name} is #{lag} events behind (head: #{head}, seen: #{last_seen})")
        # Optionally: skip the heartbeat for this projection's monitor
      end
    end)
  end

  defp schedule, do: Process.send_after(self(), :check, @check_interval)
end

Step 6: Track Command Dispatch Errors with Telemetry

Commanded emits telemetry events you can use for error tracking:

# lib/my_app/commanded_telemetry.ex
defmodule MyApp.CommandedTelemetry do
  def setup do
    :telemetry.attach_many("commanded-monitoring", [
      [:commanded, :command, :dispatch, :stop],
      [:commanded, :event, :handler, :handle, :stop],
    ], &handle_event/4, nil)
  end

  def handle_event([:commanded, :command, :dispatch, :stop], measurements, metadata, _) do
    if metadata[:error] do
      Logger.error("Command dispatch failed: #{inspect(metadata[:command])} — #{inspect(metadata[:error])}")
    end
    # Send to your metrics system: measurements[:duration] is in native time units
    :telemetry.execute([:my_app, :command, :duration], %{
      duration: System.convert_time_unit(measurements[:duration], :native, :millisecond)
    }, %{command: metadata[:command].__struct__})
  end

  def handle_event([:commanded, :event, :handler, :handle, :stop], _measurements, metadata, _) do
    if metadata[:error] do
      Logger.error("Event handler failed: #{inspect(metadata[:handler])} — #{inspect(metadata[:error])}")
    end
  end
end

Attach in application.ex:

def start(_type, _args) do
  MyApp.CommandedTelemetry.setup()
  # ...
end

Key Metrics to Watch

| Metric | Vigilmon feature | Alert threshold | |---|---|---| | EventStore availability | HTTP monitor on /health/commanded | Any failure | | Event handler liveness | Heartbeat monitor | Missing for > 2 min | | Projection lag | Custom GenServer + heartbeat | > 100 events behind | | Command dispatch latency | Response time chart | P95 > 500ms | | SSL certificate | Cert expiry monitor | < 14 days remaining |

Conclusion

Commanded's event-driven architecture gives you resilience and auditability, but also new failure modes invisible to standard monitors. With a command bus health endpoint, event handler heartbeats, and projection lag checks, Vigilmon can alert you to CQRS failures before they reach your users.

Start monitoring for free at vigilmon.online.