tutorial

How to Monitor Opus with Vigilmon

Opus is an Elixir pipeline pattern library implementing the service object pattern with automatic telemetry instrumentation. Learn how to monitor your Opus pipelines, detect stage failures, and alert on service object errors using Vigilmon.

Opus is an Elixir library that implements the service object pattern through composable pipeline stages. Each stage can validate, transform, or enrich data, and Opus automatically instruments every pipeline with :telemetry events — emitting [:opus, :pipeline, :step, :start], [:opus, :pipeline, :step, :stop], and [:opus, :pipeline, :step, :exception] events for every stage. When a pipeline stage fails — because an external API times out, a guard clause rejects unexpected input, or a database call raises — Opus halts the pipeline and returns an error tuple. Vigilmon adds external uptime and heartbeat monitoring so you know when your Opus pipelines stop completing successfully, not just when a stage raises an exception you happened to catch in logs.

What You'll Set Up

  • HTTP health endpoint that surfaces Opus pipeline success and failure rates
  • Heartbeat monitor for recurring Opus-driven service operations
  • Telemetry handler that feeds pipeline metrics to your health endpoint
  • Slack alerts when pipeline failure rates exceed thresholds

Prerequisites

  • Elixir 1.14+ with opus in mix.exs
  • A Phoenix or OTP application using Opus pipelines
  • A free Vigilmon account

Step 1: Attach a Telemetry Handler to Track Pipeline Results

Opus emits telemetry events automatically. Attach a handler to count successes and failures:

# lib/my_app/pipeline_tracker.ex
defmodule MyApp.PipelineTracker do
  use Agent

  def start_link(_opts) do
    Agent.start_link(
      fn -> %{} end,
      name: __MODULE__
    )
  end

  def record(pipeline, :success) do
    Agent.update(__MODULE__, fn state ->
      Map.update(state, pipeline, %{success: 1, failure: 0}, fn counts ->
        %{counts | success: counts.success + 1}
      end)
    end)
  end

  def record(pipeline, :failure) do
    Agent.update(__MODULE__, fn state ->
      Map.update(state, pipeline, %{success: 0, failure: 1}, fn counts ->
        %{counts | failure: counts.failure + 1}
      end)
    end)
  end

  def stats do
    Agent.get(__MODULE__, & &1)
  end
end

Add it to your supervision tree:

# lib/my_app/application.ex
children = [
  MyApp.Repo,
  MyAppWeb.Endpoint,
  MyApp.PipelineTracker,
  # ...
]

Attach the telemetry handler at startup:

# lib/my_app/application.ex
def start(_type, _args) do
  :telemetry.attach_many(
    "opus-pipeline-tracker",
    [
      [:opus, :pipeline, :stop],
      [:opus, :pipeline, :exception],
    ],
    &MyApp.PipelineTelemetryHandler.handle_event/4,
    nil
  )

  # ... supervisor start
end

# lib/my_app/pipeline_telemetry_handler.ex
defmodule MyApp.PipelineTelemetryHandler do
  require Logger

  def handle_event([:opus, :pipeline, :stop], measurements, metadata, _config) do
    pipeline = metadata[:pipeline] || :unknown
    result = metadata[:result]

    case result do
      {:ok, _} ->
        MyApp.PipelineTracker.record(pipeline, :success)

      {:error, _} ->
        MyApp.PipelineTracker.record(pipeline, :failure)
        Logger.warning("Opus pipeline #{inspect(pipeline)} failed: #{inspect(result)}")
    end
  end

  def handle_event([:opus, :pipeline, :exception], _measurements, metadata, _config) do
    pipeline = metadata[:pipeline] || :unknown
    MyApp.PipelineTracker.record(pipeline, :failure)
    Logger.error("Opus pipeline #{inspect(pipeline)} raised: #{inspect(metadata[:error])}")
  end
end

Step 2: Define an Opus Pipeline with Instrumented Stages

# lib/my_app/pipelines/create_order_pipeline.ex
defmodule MyApp.Pipelines.CreateOrderPipeline do
  use Opus.Pipeline

  step :validate_input
  step :check_inventory
  step :charge_payment
  step :create_order_record
  step :send_confirmation_email

  def validate_input(%{user_id: uid, items: items} = input)
      when is_integer(uid) and is_list(items) and items != [] do
    {:ok, input}
  end
  def validate_input(_), do: {:error, :invalid_input}

  def check_inventory(%{items: items} = input) do
    case MyApp.Inventory.check(items) do
      :ok -> {:ok, input}
      {:error, :out_of_stock} -> {:error, :items_unavailable}
    end
  end

  def charge_payment(%{user_id: uid, total: total} = input) do
    case MyApp.Payments.charge(uid, total) do
      {:ok, charge_id} -> {:ok, Map.put(input, :charge_id, charge_id)}
      {:error, reason} -> {:error, {:payment_failed, reason}}
    end
  end

  def create_order_record(input) do
    case MyApp.Orders.create(input) do
      {:ok, order} -> {:ok, Map.put(input, :order, order)}
      {:error, changeset} -> {:error, {:db_error, changeset}}
    end
  end

  def send_confirmation_email(%{order: order} = input) do
    MyApp.Mailer.send_order_confirmation(order)
    {:ok, input}
  end
end

Each stage emits telemetry automatically — your handler above captures the final result.

Step 3: Add a Health Endpoint

# lib/my_app_web/controllers/health_controller.ex
defmodule MyAppWeb.HealthController do
  use MyAppWeb, :controller

  alias MyApp.PipelineTracker

  def index(conn, _params) do
    stats = PipelineTracker.stats()

    # Compute overall failure rate across all pipelines
    total_success = stats |> Map.values() |> Enum.sum_by(& &1.success)
    total_failure = stats |> Map.values() |> Enum.sum_by(& &1.failure)
    total = total_success + total_failure

    failure_rate = if total > 0, do: total_failure / total, else: 0.0
    pipelines_ok = failure_rate < 0.1  # alert if >10% failure rate

    checks = %{
      pipelines: stats,
      failure_rate: Float.round(failure_rate * 100, 2),
      database: check_database()
    }

    status = if pipelines_ok and checks.database == :ok, do: 200, else: 503

    conn
    |> put_status(status)
    |> json(%{status: if(status == 200, do: "ok", else: "degraded"), checks: checks})
  end

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

Register the route:

# lib/my_app_web/router.ex
scope "/", MyAppWeb do
  get "/health", HealthController, :index
end

Step 4: Set Up HTTP Monitoring in Vigilmon

  1. Sign in at vigilmon.online
  2. Click New Monitor → HTTP
  3. URL: https://yourdomain.com/health
  4. Expected status: 200
  5. Add a keyword check: "status":"ok" to confirm the pipeline tracker reports healthy
  6. Interval: 1 minute
  7. Save

The keyword check catches cases where your application is up but pipelines are failing at a high rate — the HTTP response will be 503 with "status":"degraded", and Vigilmon will alert immediately.

Step 5: Heartbeat Monitoring for Recurring Opus Pipelines

If you run an Opus pipeline on a schedule (via Oban, Quantum, or a GenServer timer), add a heartbeat so Vigilmon detects when that pipeline stops completing:

# lib/my_app/workers/daily_report_worker.ex
defmodule MyApp.Workers.DailyReportWorker do
  use Oban.Worker, queue: :reports, max_attempts: 3

  require Logger

  @impl Oban.Worker
  def perform(%Oban.Job{}) do
    case MyApp.Pipelines.GenerateReportPipeline.call(%{date: Date.utc_today()}) do
      {:ok, _result} ->
        ping_heartbeat()
        :ok

      {:error, reason} ->
        Logger.error("Daily report pipeline failed: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:report_heartbeat_url]

    if url do
      case Req.get(url, receive_timeout: 5_000) do
        {:ok, _} -> :ok
        {:error, reason} ->
          Logger.warning("Vigilmon heartbeat ping failed: #{inspect(reason)}")
      end
    end
  end
end

Configure the heartbeat URL:

# config/runtime.exs
config :my_app, :vigilmon,
  report_heartbeat_url: System.get_env("VIGILMON_REPORT_HEARTBEAT_URL")

In Vigilmon:

  1. Click New Monitor → Heartbeat
  2. Name: daily-report-pipeline
  3. Expected interval: 24 hours
  4. Copy the ping URL and set it as VIGILMON_REPORT_HEARTBEAT_URL

Step 6: Alert on Pipeline Failures via Slack

  1. In Vigilmon: Notifications → New Channel → Slack
  2. Paste your Slack incoming webhook URL
  3. Enable it on your /health monitor and all heartbeat monitors

A pipeline failure alert looks like:

🔴 DOWN: yourdomain.com/health
Status: 503 — "status":"degraded" (failure_rate: 14.3%)
Detected: EU-West, US-East

Recovery:

✅ RECOVERED: yourdomain.com/health
Downtime: 8 minutes

What You've Built

| What | How | |------|-----| | Pipeline telemetry capture | :telemetry handler counting Opus stop and exception events | | Active health check | /health with failure-rate threshold and keyword monitor | | Scheduled pipeline monitoring | Heartbeat on Oban workers running recurring Opus pipelines | | Failure rate alerts | Slack notification when health degrades | | Per-pipeline visibility | Stats grouped by pipeline module in health response |

Opus makes your service operations composable and instrumented. Vigilmon makes the health of those service operations visible from outside your application.

Next Steps

  • Add a per-pipeline Vigilmon heartbeat for any Opus pipeline critical enough to page on individually
  • Use Vigilmon's response-time history to track /health endpoint latency as a proxy for pipeline throughput
  • Extend PipelineTelemetryHandler to record stage-level failures, not just pipeline-level, for more granular debugging
  • Reset pipeline counters on a schedule so failure rates reflect recent activity

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