How to Monitor Flow with Vigilmon

Flow is a computational parallel processing library for Elixir that builds on GenStage to provide a MapReduce-style API with automatic back-pressure. You define a pipeline of map, filter, reduce, and partition stages, and Flow distributes the work across multiple CPU cores with configurable concurrency and window semantics. Flow is designed for processing large datasets — event logs, user activity streams, batch imports, analytics aggregations — where GenStage's producer/consumer model gives you safety from memory overload but GenStage alone is too low-level to express multi-stage transformations concisely. When a Flow pipeline stops processing, fails mid-stream, or backs up indefinitely, data goes unprocessed without visible errors. Vigilmon gives you heartbeat monitoring, HTTP health checks, and alerting to catch those silent failures.

What You'll Set Up

• HTTP health endpoint that validates your Flow pipeline processes a test batch
• Heartbeat monitor for scheduled Flow processing jobs
• Slack alerts on pipeline failure or stall
• SSL monitoring for data ingestion endpoints that feed Flow pipelines

Prerequisites

• Elixir 1.14+ with flow in mix.exs
• A GenStage/Flow-based data processing application
• A free Vigilmon account

Step 1: Add a Health Endpoint That Exercises Your Flow Pipeline

Rather than checking only if the BEAM is alive, exercise a miniature version of your Flow pipeline in the health check to catch logic regressions:

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

  def index(conn, _params) do
    checks = %{
      flow_pipeline: check_flow_pipeline(),
      database: check_database(),
      pipeline_supervisor: check_pipeline_supervisor()
    }

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

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

  defp check_flow_pipeline do
    # Run a tiny test batch through the pipeline to verify end-to-end
    result =
      1..10
      |> Flow.from_enumerable(max_demand: 5)
      |> Flow.map(fn n -> n * 2 end)
      |> Flow.filter(fn n -> rem(n, 4) == 0 end)
      |> Enum.to_list()

    if length(result) > 0, do: :ok, else: :error
  rescue
    _ -> :error
  end

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

  defp check_pipeline_supervisor do
    # Verify the Flow supervisor tree is alive
    case Process.whereis(MyApp.PipelineSupervisor) do
      nil -> :error
      _pid -> :ok
    end
  end
end

Wire it to a route:

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

The check_flow_pipeline/0 function runs a tiny enumerable through a sample pipeline and verifies the output is non-empty. If Flow is misconfigured or a dependency is missing, this check fails before Prometheus or your logs reveal anything.

Step 2: Add Heartbeat Monitoring for Batch Processing Jobs

Flow pipelines commonly run as Oban workers, GenServers, or Mix tasks that process batches of records on a schedule. Add a heartbeat ping at the end of each successful batch:

# lib/my_app/workers/event_processing_worker.ex
defmodule MyApp.Workers.EventProcessingWorker do
  use Oban.Worker, queue: :processing, max_attempts: 3

  require Logger

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"batch_id" => batch_id}}) do
    with {:ok, events} <- fetch_pending_events(batch_id),
         {:ok, results} <- process_events_with_flow(events),
         :ok <- persist_results(results) do
      Logger.info("Processed #{length(results)} events for batch #{batch_id}")
      ping_heartbeat()
      :ok
    else
      {:error, reason} ->
        Logger.error("Event processing failed for batch #{batch_id}: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp fetch_pending_events(batch_id) do
    events = MyApp.Repo.all(
      from e in MyApp.Event,
      where: e.batch_id == ^batch_id and e.status == "pending",
      limit: 10_000
    )
    {:ok, events}
  end

  defp process_events_with_flow(events) do
    results =
      events
      |> Flow.from_enumerable(max_demand: 500, stages: System.schedulers_online())
      |> Flow.map(&enrich_event/1)
      |> Flow.filter(&valid_event?/1)
      |> Flow.partition(key: fn event -> event.user_id end)
      |> Flow.reduce(fn -> %{} end, fn event, acc ->
        Map.update(acc, event.user_id, [event], &[event | &1])
      end)
      |> Flow.on_trigger(fn acc ->
        {Map.to_list(acc), %{}}
      end)
      |> Enum.to_list()

    {:ok, List.flatten(results)}
  rescue
    error ->
      Logger.error("Flow processing crashed: #{inspect(error)}")
      {:error, error}
  end

  defp enrich_event(event) do
    # Add computed fields, resolve references, etc.
    Map.put(event, :processed_at, DateTime.utc_now())
  end

  defp valid_event?(event) do
    not is_nil(event.user_id) and not is_nil(event.type)
  end

  defp persist_results(results) do
    {_count, _} = MyApp.Repo.insert_all(MyApp.ProcessedEvent, results,
      on_conflict: :nothing,
      conflict_target: [:event_id]
    )
    :ok
  end

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:processing_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,
  processing_heartbeat_url: System.get_env("VIGILMON_PROCESSING_HEARTBEAT_URL")

Step 3: Monitor Long-Running Flow GenServer Pipelines

For persistent Flow pipelines that run continuously (consuming from a message queue or event stream), add a periodic heartbeat from the GenServer:

# lib/my_app/pipeline_server.ex
defmodule MyApp.PipelineServer do
  use GenServer

  require Logger

  @heartbeat_interval :timer.minutes(5)

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  @impl true
  def init(opts) do
    state = %{
      source: Keyword.fetch!(opts, :source),
      processed_count: 0,
      last_heartbeat: nil
    }

    schedule_heartbeat()
    {:ok, state, {:continue, :start_pipeline}}
  end

  @impl true
  def handle_continue(:start_pipeline, state) do
    # Start the Flow pipeline as a supervised process
    pid = start_flow_pipeline(state.source)
    {:noreply, Map.put(state, :pipeline_pid, pid)}
  end

  @impl true
  def handle_info(:send_heartbeat, state) do
    new_state =
      if pipeline_healthy?(state) do
        ping_heartbeat()
        %{state | last_heartbeat: DateTime.utc_now()}
      else
        Logger.error("Pipeline health check failed — skipping heartbeat")
        state
      end

    schedule_heartbeat()
    {:noreply, new_state}
  end

  @impl true
  def handle_cast({:event_processed, count}, state) do
    {:noreply, %{state | processed_count: state.processed_count + count}}
  end

  defp start_flow_pipeline(source) do
    # Start a supervised Flow consumer on your GenStage source
    {:ok, pid} =
      source
      |> Flow.from_stages(max_demand: 1000, stages: System.schedulers_online())
      |> Flow.map(&process_event/1)
      |> Flow.partition(key: &event_partition_key/1)
      |> Flow.each(fn event ->
        GenServer.cast(__MODULE__, {:event_processed, 1})
        persist_event(event)
      end)
      |> Flow.start_link()

    pid
  end

  defp pipeline_healthy?(%{pipeline_pid: pid}) when is_pid(pid) do
    Process.alive?(pid)
  end
  defp pipeline_healthy?(_), do: false

  defp process_event(event) do
    Map.put(event, :processed_at, System.system_time(:millisecond))
  end

  defp event_partition_key(event), do: event.user_id

  defp persist_event(_event) do
    # Write to database, emit to downstream queue, etc.
    :ok
  end

  defp schedule_heartbeat do
    Process.send_after(self(), :send_heartbeat, @heartbeat_interval)
  end

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

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

Register the GenServer in your supervision tree:

# lib/my_app/application.ex
children = [
  MyApp.Repo,
  MyAppWeb.Endpoint,
  {MyApp.PipelineServer, source: MyApp.EventSource},
]

Step 4: Set Up Monitoring in Vigilmon

HTTP health monitor:

1. Sign in at vigilmon.online
2. New Monitor → HTTP
3. URL: https://yourdomain.com/health
4. Expected status: 200
5. Interval: 1 minute

Heartbeat monitor (for batch Oban workers):

1. New Monitor → Heartbeat
2. Name: event-processing-pipeline
3. Expected interval: match your Oban schedule (e.g. 10 minutes)
4. Tolerance: 3 minutes
5. Copy the ping URL and set it as VIGILMON_PROCESSING_HEARTBEAT_URL

Heartbeat monitor (for the persistent GenServer pipeline):

1. New Monitor → Heartbeat
2. Name: flow-pipeline-server
3. Expected interval: 5 minutes (matches @heartbeat_interval)
4. Tolerance: 2 minutes
5. Copy the ping URL and set it as VIGILMON_PIPELINE_HEARTBEAT_URL

If the GenServer crashes and its supervisor stops retrying, Vigilmon alerts you after the next missed 5-minute window — before the back-pressure causes cascading failures upstream.

Step 5: Configure Slack Alerts

1. In Vigilmon: Notifications → New Channel → Slack
2. Paste your Slack incoming webhook URL
3. Enable on all monitors

Alert on pipeline stall:

🔴 MISSED: flow-pipeline-server
Last ping: 12 minutes ago (expected: 5 minutes)

Alert on health failure:

🔴 DOWN: yourdomain.com/health
Status: 503 (flow_pipeline: error)
Detected: EU-West, US-East

Step 6: SSL Monitoring for Data Ingestion Endpoints

Flow pipelines often consume from HTTP-based sources (webhook receivers, REST API polling endpoints). Add SSL monitoring:

1. New Monitor → SSL Certificate
2. URL: https://yourdomain.com (your ingestion endpoint)
3. Alert: 14 days before expiry

A lapsed certificate on your ingestion endpoint stops data from reaching the pipeline. Your Flow GenServer stays alive and healthy — but processes nothing because the upstream source is blocked.

What You've Built

| What | How | |------|-----| | Active health check | /health running a test batch through Flow end-to-end | | Batch job monitoring | Heartbeat with 10-minute window per Oban worker | | Persistent pipeline monitoring | 5-minute heartbeat from GenServer on each healthy tick | | Slack alerts | Downtime and missed-heartbeat notifications | | SSL monitoring | Certificate expiry alert for ingestion endpoints |

Flow handles back-pressure automatically — it won't overload downstream stages. Vigilmon handles a different class of failure: when the entire pipeline stops silently.

Next Steps

• Add a heartbeat per partition key family if you run Flow with multiple partitions processing different event types
• Use Vigilmon's response time history to track how long your health check takes — growth in the /health response time indicates the test batch is getting slower, which predicts real pipeline slowdowns
• Add a monitor for your GenStage producer source (message queue, database polling query) to catch upstream failures before they stall Flow
• If you run multiple Flow workers in parallel, add a heartbeat per worker to isolate which partition is failing

Get started free at vigilmon.online.