How to Monitor GenStage with Vigilmon

GenStage lets you build concurrent, back-pressure-aware data processing pipelines in Elixir. Producers emit events, consumers process them, and back-pressure ensures consumers never get overwhelmed. But pipelines have their own failure modes: a producer that stalls emitting events, a consumer whose demand accumulates without being met, or a stage that crashes and isn't restarted by its supervisor.

This tutorial covers end-to-end monitoring for GenStage pipelines:

• Process liveness checks for each pipeline stage
• HTTP health endpoints that report pipeline status
• Heartbeat monitors for pipeline throughput
• Back-pressure and queue depth alerting
• Slack alerts when the pipeline stalls

Why Monitor GenStage?

| Failure mode | What breaks | |---|---| | Producer stalls | No events emitted; consumers idle; work backlog grows | | Consumer crash | Events accumulate; pipeline falls behind | | Back-pressure saturation | System starves upstream producers | | Stage supervisor crash | Entire pipeline stops; process not restarted | | Slow consumer | Upstream buffers fill; memory pressure builds |

GenStage handles back-pressure automatically, but it won't restart a stalled external source or alert you when throughput drops to zero.

Step 1: Add Health Checks to Each Stage

Instrument each stage with a way to report its health. The simplest approach is a status/0 function:

# lib/my_app/pipeline/producer.ex
defmodule MyApp.Pipeline.Producer do
  use GenStage

  defstruct [:queue, :demand, :processed_count, :last_emit_at]

  def start_link(opts), do: GenStage.start_link(__MODULE__, opts, name: __MODULE__)

  def status do
    GenStage.call(__MODULE__, :status)
  end

  def init(_opts) do
    {:producer, %__MODULE__{queue: :queue.new(), demand: 0, processed_count: 0, last_emit_at: nil}}
  end

  def handle_call(:status, _from, state) do
    status = %{
      queue_size: :queue.len(state.queue),
      pending_demand: state.demand,
      total_processed: state.processed_count,
      last_emit_at: state.last_emit_at,
      alive: true
    }
    {:reply, status, [], state}
  end

  def handle_demand(demand, state) do
    {events, new_queue} = take_from_queue(state.queue, demand)
    new_state = %{state |
      queue: new_queue,
      demand: demand - length(events),
      processed_count: state.processed_count + length(events),
      last_emit_at: if(events != [], do: System.monotonic_time(:second), else: state.last_emit_at)
    }
    {:noreply, events, new_state}
  end

  defp take_from_queue(queue, n), do: take_from_queue(queue, n, [])
  defp take_from_queue(queue, 0, acc), do: {Enum.reverse(acc), queue}
  defp take_from_queue(queue, n, acc) do
    case :queue.out(queue) do
      {{:value, item}, new_queue} -> take_from_queue(new_queue, n - 1, [item | acc])
      {:empty, queue} -> {Enum.reverse(acc), queue}
    end
  end
end

# lib/my_app/pipeline/consumer.ex
defmodule MyApp.Pipeline.Consumer do
  use GenStage

  defstruct [:processed_count, :last_processed_at, :error_count]

  def start_link(opts), do: GenStage.start_link(__MODULE__, opts, name: __MODULE__)

  def status do
    GenStage.call(__MODULE__, :status)
  end

  def init(_opts) do
    {:consumer, %__MODULE__{processed_count: 0, last_processed_at: nil, error_count: 0},
     subscribe_to: [{MyApp.Pipeline.Producer, max_demand: 50, min_demand: 10}]}
  end

  def handle_call(:status, _from, state) do
    status = %{
      total_processed: state.processed_count,
      last_processed_at: state.last_processed_at,
      error_count: state.error_count,
      alive: true
    }
    {:reply, status, [], state}
  end

  def handle_events(events, _from, state) do
    {successes, errors} = process_events(events)
    new_state = %{state |
      processed_count: state.processed_count + successes,
      last_processed_at: System.monotonic_time(:second),
      error_count: state.error_count + errors
    }
    {:noreply, [], new_state}
  end

  defp process_events(events) do
    Enum.reduce(events, {0, 0}, fn event, {ok, err} ->
      case process(event) do
        :ok -> {ok + 1, err}
        _ -> {ok, err + 1}
      end
    end)
  end

  defp process(_event), do: :ok
end

Step 2: Add a Health Endpoint for the Pipeline

Create an HTTP endpoint that polls all stages and reports overall pipeline health:

# lib/my_app_web/plugs/pipeline_health.ex
defmodule MyAppWeb.Plugs.PipelineHealth do
  import Plug.Conn
  require Logger

  @stall_threshold_seconds 60

  def init(opts), do: opts

  def call(%Plug.Conn{request_path: "/health/pipeline"} = conn, _opts) do
    checks = collect_health()
    status = if pipeline_healthy?(checks), do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      pipeline: checks
    }))
    |> halt()
  end

  def call(conn, _opts), do: conn

  defp collect_health do
    stages = [
      {"producer", &MyApp.Pipeline.Producer.status/0},
      {"consumer", &MyApp.Pipeline.Consumer.status/0}
    ]

    Enum.map(stages, fn {name, status_fn} ->
      status = try do
        status_fn.()
      rescue
        _ -> %{alive: false}
      catch
        :exit, _ -> %{alive: false}
      end
      {name, Map.put(status, :stalled, stalled?(status))}
    end)
    |> Map.new()
  end

  defp stalled?(%{alive: false}), do: true
  defp stalled?(%{last_emit_at: nil}), do: false  # Never emitted — not stalled, just empty
  defp stalled?(%{last_emit_at: last}) do
    System.monotonic_time(:second) - last > @stall_threshold_seconds
  end
  defp stalled?(_), do: false

  defp pipeline_healthy?(checks) do
    Enum.all?(checks, fn {_name, stage} ->
      Map.get(stage, :alive, false) and not Map.get(stage, :stalled, false)
    end)
  end
end

Register in endpoint.ex:

plug MyAppWeb.Plugs.PipelineHealth
plug MyAppWeb.Router

Step 3: Add Vigilmon HTTP Monitoring

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

Vigilmon will alert you the moment your pipeline stalls or a stage crashes.

Step 4: Monitor Pipeline Throughput with a Heartbeat

Uptime monitoring tells you if the pipeline process is alive. A heartbeat tells you if it's actually doing work. Add a throughput heartbeat that only pings Vigilmon when events are flowing:

# lib/my_app/pipeline_heartbeat.ex
defmodule MyApp.PipelineHeartbeat do
  use GenServer
  require Logger

  @interval :timer.minutes(1)
  @min_throughput_per_minute 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, %{last_count: 0}}
  end

  def handle_info(:check, %{last_count: last_count} = state) do
    current_count = get_current_count()
    delta = current_count - last_count
    throughput_ok = delta >= @min_throughput_per_minute

    if throughput_ok do
      case HTTPoison.get(@vigilmon_url) do
        {:ok, _} -> :ok
        error -> Logger.warning("Vigilmon heartbeat failed: #{inspect(error)}")
      end
    else
      Logger.warning("Pipeline throughput too low: #{delta} events/min (min: #{@min_throughput_per_minute})")
    end

    schedule()
    {:noreply, %{state | last_count: current_count}}
  end

  defp get_current_count do
    case MyApp.Pipeline.Consumer.status() do
      %{total_processed: count} -> count
      _ -> 0
    end
  end

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

Add to your supervision tree:

children = [
  MyApp.Pipeline.Producer,
  MyApp.Pipeline.Consumer,
  MyApp.PipelineHeartbeat
]

Create the heartbeat in Vigilmon:

1. Click Add Monitor → Cron / Heartbeat.
2. Set expected interval to 1 minute.
3. Set grace period to 2 minutes.
4. Copy the heartbeat URL into @vigilmon_url.

Step 5: Set Up Telemetry for Back-Pressure Metrics

Track back-pressure and queue depth with Elixir's telemetry:

# lib/my_app/pipeline_telemetry.ex
defmodule MyApp.PipelineTelemetry do
  def setup do
    :telemetry.attach_many("genstage-pipeline", [
      [:my_app, :pipeline, :queue_depth],
      [:my_app, :pipeline, :demand_pending],
      [:my_app, :pipeline, :events_processed],
    ], &handle_event/4, nil)
  end

  def handle_event([:my_app, :pipeline, :queue_depth], %{depth: depth}, _meta, _cfg) do
    if depth > 1000 do
      Logger.warning("Pipeline queue depth critical: #{depth} events buffered")
    end
  end

  def handle_event([:my_app, :pipeline, :demand_pending], %{demand: demand}, _meta, _cfg) do
    if demand > 500 do
      Logger.warning("Unsatisfied consumer demand: #{demand} events requested but not delivered")
    end
  end

  def handle_event([:my_app, :pipeline, :events_processed], %{count: _count}, _meta, _cfg) do
    :ok
  end
end

Emit telemetry from your producer's handle_demand/2 and periodically from a GenServer:

defmodule MyApp.PipelineMetricsReporter do
  use GenServer

  def handle_info(:report, state) do
    case MyApp.Pipeline.Producer.status() do
      %{queue_size: depth, pending_demand: demand} ->
        :telemetry.execute([:my_app, :pipeline, :queue_depth], %{depth: depth}, %{})
        :telemetry.execute([:my_app, :pipeline, :demand_pending], %{demand: demand}, %{})
      _ -> :ok
    end
    Process.send_after(self(), :report, 10_000)
    {:noreply, state}
  end
end

Key Metrics to Watch

| Metric | Vigilmon feature | Alert threshold | |---|---|---| | Pipeline stage liveness | HTTP monitor on /health/pipeline | Any 5xx or "stalled":true | | Pipeline throughput | Heartbeat monitor | Missing for > 2 min | | Queue depth | Custom GenServer logging | > 1,000 events buffered | | Pending demand | Telemetry event | > 500 unmet demand units | | SSL certificate | Cert expiry monitor | < 14 days remaining |

Conclusion

GenStage's back-pressure guarantees protect your system from overload, but they don't protect you from pipeline stalls, consumer crashes, or zero-throughput scenarios. With a pipeline health endpoint, a throughput heartbeat, and telemetry-based depth alerts, Vigilmon gives you full visibility into your data pipeline's health.

Start monitoring your pipelines for free at vigilmon.online.