tutorial

How to Monitor Broadway with Vigilmon

Set up heartbeat monitoring, pipeline health checks, and instant alerts for your Broadway data ingestion and processing pipelines with Vigilmon.

Broadway is Elixir's concurrent, multi-stage data ingestion and processing pipeline built on GenStage. It handles back-pressure, batching, concurrency, and fault tolerance automatically — letting you focus on your producer and processor logic instead of the plumbing. Broadway pipelines power everything from message queue consumers (SQS, RabbitMQ, Kafka) to event-driven ETL workflows. But because Broadway runs continuously as a supervised process tree, failures are often silent: the pipeline tree stays alive, the supervisor is happy, but messages stop flowing and you only notice when queues back up or downstream systems go stale. Vigilmon closes that gap with heartbeat monitors that alert you the moment a pipeline stops processing, plus HTTP monitors for any web-facing surfaces your Broadway app exposes.

What You'll Build

  • A Broadway processor that pings a Vigilmon heartbeat on each successful batch
  • A /health endpoint for web-facing Broadway apps
  • Vigilmon heartbeat monitors per pipeline stage
  • Alert channels (email and Slack)

Prerequisites

  • An Elixir application with Broadway (Broadway 1.x)
  • A free Vigilmon account

Step 1: Add a Heartbeat Ping to Your Broadway Pipeline

The most important thing to monitor in a Broadway pipeline is whether batches are actually completing successfully. Add a heartbeat ping at the end of your handle_batch/4 callback — it fires only when a batch processes without error.

Basic pipeline with heartbeat

# lib/my_app/pipelines/event_pipeline.ex
defmodule MyApp.Pipelines.EventPipeline do
  use Broadway

  require Logger

  alias Broadway.Message

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {BroadwaySQS.Producer, queue_url: System.get_env("SQS_QUEUE_URL")},
        concurrency: 1
      ],
      processors: [
        default: [concurrency: 10]
      ],
      batchers: [
        default: [
          batch_size: 100,
          batch_timeout: 5_000,
          concurrency: 5
        ]
      ]
    )
  end

  @impl true
  def handle_message(_, message, _) do
    message
    |> Message.update_data(&process_event/1)
  rescue
    e ->
      Logger.error("EventPipeline message processing failed: #{inspect(e)}")
      Message.failed(message, inspect(e))
  end

  @impl true
  def handle_batch(:default, messages, _batch_info, _context) do
    successful = Enum.count(messages, &(&1.status == :ok))

    if successful > 0 do
      ping_heartbeat()
    end

    messages
  end

  defp process_event(data) do
    # Your actual event processing logic
    Jason.decode!(data)
  end

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

    if url do
      case Req.get(url, receive_timeout: 5_000) do
        {:ok, _} ->
          :ok

        {:error, reason} ->
          Logger.warning("Broadway heartbeat ping failed: #{inspect(reason)}")
      end
    end
  end
end

Add the pipeline to your supervision tree:

# lib/my_app/application.ex
def start(_type, _args) do
  children = [
    MyApp.Repo,
    MyApp.Pipelines.EventPipeline
  ]

  Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)
end

Add config:

# config/runtime.exs
config :my_app, :vigilmon,
  event_pipeline_heartbeat_url: System.get_env("VIGILMON_EVENT_PIPELINE_HEARTBEAT_URL")

Step 2: Configure Vigilmon Heartbeat Monitors

In Vigilmon → Monitors → New Heartbeat Monitor:

Heartbeat 1: Event Pipeline

Set the expected interval based on your actual throughput. If you process at least one batch every 5 minutes, set a 10-minute window:

| Field | Value | |---|---| | Name | Event Pipeline — Broadway | | Expected interval | 10 minutes | | Grace period | 2 minutes |

Copy the ping URL:

export VIGILMON_EVENT_PIPELINE_HEARTBEAT_URL=https://vigilmon.online/api/heartbeat/xxxxxxxx

Heartbeat 2: Per-stage monitoring for multi-pipeline apps

If you run multiple Broadway pipelines, create a separate heartbeat for each:

# lib/my_app/pipelines/notification_pipeline.ex
defmodule MyApp.Pipelines.NotificationPipeline do
  use Broadway

  # ... pipeline config

  @impl true
  def handle_batch(:default, messages, _batch_info, _context) do
    if Enum.any?(messages, &(&1.status == :ok)) do
      url = Application.get_env(:my_app, :vigilmon)[:notification_pipeline_heartbeat_url]
      if url, do: Req.get(url, receive_timeout: 5_000)
    end

    messages
  end
end

# config/runtime.exs
config :my_app, :vigilmon,
  event_pipeline_heartbeat_url: System.get_env("VIGILMON_EVENT_PIPELINE_HEARTBEAT_URL"),
  notification_pipeline_heartbeat_url: System.get_env("VIGILMON_NOTIFICATION_PIPELINE_HEARTBEAT_URL")

Step 3: Health Endpoint for Web-Facing Broadway Apps

If your Broadway app also exposes a web interface or API (common with Phoenix-backed Broadway deployments), add an HTTP health check that validates Broadway pipeline status alongside the standard checks.

# 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(),
      event_pipeline: check_broadway_pipeline(MyApp.Pipelines.EventPipeline)
    }

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

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(%{
      status: if(degraded, do: "degraded", else: "ok"),
      checks: Map.new(checks, fn {k, v} ->
        {k, if(v == :ok, do: "ok", else: "error")}
      end)
    }))
    |> 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_broadway_pipeline(pipeline_module) do
    # Broadway.topology/1 returns producer/processor/batcher pids
    # If the pipeline is running, the call succeeds; if it crashed, it raises
    case Broadway.topology(pipeline_module) do
      topology when is_list(topology) -> :ok
      _ -> :error
    end
  rescue
    _ -> :error
  end
end

Register the health check in your Phoenix endpoint:

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

Add a Vigilmon HTTP monitor pointing at this endpoint:

| Field | Value | |---|---| | URL | https://yourapp.com/health | | Method | GET | | Expected status | 200 | | Expected body | "status":"ok" | | Check interval | 1 minute |

Step 4: Handle Failed Messages Without Losing Heartbeat Signal

Broadway messages that fail and get retried should not suppress the heartbeat — the pipeline is still processing. Only distinguish between "all messages in a batch failed" (pipeline broken) vs. "some messages failed" (expected error rate).

@impl true
def handle_batch(:default, messages, batch_info, _context) do
  successful = Enum.count(messages, &(&1.status == :ok))
  failed = Enum.count(messages, &(&1.status != :ok))

  Logger.info("Broadway batch complete: #{successful} ok, #{failed} failed",
    pipeline: __MODULE__,
    batch_key: batch_info.batch_key
  )

  # Ping heartbeat if ANY messages succeeded — the pipeline is running
  if successful > 0 do
    ping_heartbeat()
  end

  # If ALL messages failed, we still processed the batch — ping anyway
  # but log at error level so you can investigate separately
  if failed == length(messages) do
    Logger.error("Broadway batch: all #{failed} messages failed — check producer/processor logs",
      pipeline: __MODULE__
    )
    ping_heartbeat()   # pipeline is alive, but messages are broken
  end

  messages
end

This approach separates two distinct failure modes:

  • Pipeline down (no heartbeat ping): caught by Vigilmon heartbeat alert
  • High error rate (heartbeat still pings): caught by your application logs and error tracking

Step 5: Alert Channels

In Vigilmon → Alert Channels:

Email

  1. Alert Channels → Email → add your on-call address
  2. Attach to all heartbeat monitors

Slack

  1. Create a Slack Incoming Webhook for #alerts
  2. Alert Channels → Webhook → paste the URL
  3. Payload template for pipeline failures:
{
  "text": "🛑 *{{ monitor.name }}* missed its heartbeat\nLast ping: {{ monitor.last_checked_at }}\nExpected every: {{ monitor.interval }}\n<https://vigilmon.online|Open Vigilmon>"
}

Step 6: Verify Your Setup

# 1. Start your Broadway pipeline and confirm it pings the heartbeat
# Check Vigilmon → your heartbeat monitor → "Last ping" timestamp updates

# 2. Simulate a pipeline crash by stopping the application
# Wait for the heartbeat window to expire
# Expected: Vigilmon heartbeat alert fires

# 3. For HTTP health monitors:
curl -s https://yourapp.com/health | jq .

# 4. Stop Broadway manually and check health endpoint
# Expected: "event_pipeline": "error" and 503 status

# 5. Vigilmon → "Test Alert" → verify email and Slack delivery

Summary

| Monitor | What It Catches | |---|---| | Heartbeat per pipeline | Silent pipeline crashes, supervisor giving up retries | | HTTP health endpoint | Database failures, Broadway topology not running | | Separate heartbeats per stage | Identifies which pipeline in a multi-pipeline app failed |

Next Steps

  • Add a heartbeat for each Broadway batcher stage if you use multiple batcher keys
  • Use Vigilmon's response time graphs to correlate heartbeat ping latency with SQS/RabbitMQ queue depth changes
  • Set up a separate alert channel for heartbeat failures vs. HTTP failures — pipeline downtime is usually higher priority than web downtime for data ingestion apps
  • Monitor your message producer (SQS queue depth, RabbitMQ queue length) separately so you can distinguish "pipeline down" from "no messages to process"

Ready to get started? Sign up for Vigilmon free — have your first monitor running in under 5 minutes.

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