tutorial

How to Monitor NimbleCSV with Vigilmon

Learn how to add health checks, CSV pipeline observability, and alerts to Elixir applications using the NimbleCSV library with Vigilmon.

How to Monitor NimbleCSV with Vigilmon

NimbleCSV is a fast and flexible CSV parsing and dumping library for Elixir. Built for streaming, it processes large CSV files without loading them entirely into memory — making it the right choice for ETL pipelines, data imports, report generation, and bulk data exports. You define a parser module once with NimbleCSV.define/2 and get a full streaming API backed by binary pattern matching.

But CSV pipelines fail in subtle ways. A malformed row in the middle of a 10 million-line import silently aborts or corrupts downstream data. A streaming parse that runs for hours without progress tracking looks healthy to uptime monitors. A scheduled export that never pings completion leaves operations blind. These failures are invisible without monitoring.

This tutorial adds production observability to an Elixir application using NimbleCSV:

  • An application health check with pipeline metrics
  • HTTP uptime monitoring with Vigilmon
  • Row throughput and error rate tracking
  • Heartbeat monitoring for scheduled import and export jobs
  • Alerts on pipeline failures and stalled processing

Step 1: Add NimbleCSV to your project

# mix.exs
defp deps do
  [
    {:nimble_csv, "~> 1.2"},
    {:plug_cowboy, "~> 2.0"},
    {:plug, "~> 1.14"},
    {:jason, "~> 1.4"},
    {:req, "~> 0.4"}
  ]
end

Define a parser module (typically in lib/my_app/csv.ex):

# lib/my_app/csv.ex
defmodule MyApp.CSV do
  NimbleCSV.define(MyApp.CSV.Parser, separator: ",", escape: "\"")
  NimbleCSV.define(MyApp.CSV.TsvParser, separator: "\t", escape: "\"")
end

Step 2: Parse and stream CSV in your application

# lib/my_app/import/product_importer.ex
defmodule MyApp.Import.ProductImporter do
  alias MyApp.CSV.Parser
  require Logger

  def import_from_file(path) do
    path
    |> File.stream!(read_ahead: 100_000)
    |> Parser.parse_stream(skip_headers: false)
    |> Stream.with_index()
    |> Stream.flat_map(fn
      {[_header | _], 0} ->
        # Skip header row
        []
      {[sku, name, price_str, stock_str], _index} ->
        case parse_row(sku, name, price_str, stock_str) do
          {:ok, product} -> [product]
          {:error, reason} ->
            Logger.warning("Skipping malformed row: #{inspect(reason)}")
            []
        end
      {row, index} ->
        Logger.warning("Unexpected column count at row #{index}: #{inspect(row)}")
        []
    end)
    |> Stream.chunk_every(500)
    |> Enum.reduce({0, 0}, fn batch, {imported, errors} ->
      case MyApp.Repo.insert_all(MyApp.Product, batch, on_conflict: :replace_all) do
        {count, _} -> {imported + count, errors}
      end
    rescue
      e ->
        Logger.error("Batch insert failed: #{inspect(e)}")
        {imported, errors + length(batch)}
    end)
  end

  defp parse_row(sku, name, price_str, stock_str) do
    with {price, ""} <- Float.parse(price_str),
         {stock, ""} <- Integer.parse(stock_str) do
      {:ok, %{sku: sku, name: name, price: price, stock: stock}}
    else
      _ -> {:error, "invalid price or stock: #{price_str}, #{stock_str}"}
    end
  end
end

Export to CSV:

# lib/my_app/export/order_exporter.ex
defmodule MyApp.Export.OrderExporter do
  alias MyApp.CSV.Parser

  def export_to_file(path) do
    header = [["order_id", "customer", "total", "status", "created_at"]]

    rows =
      MyApp.Repo.stream(MyApp.Order)
      |> Stream.map(fn order ->
        [to_string(order.id), order.customer_email,
         to_string(order.total), order.status,
         DateTime.to_iso8601(order.inserted_at)]
      end)

    (header ++ rows)
    |> Parser.dump_to_stream()
    |> Stream.into(File.stream!(path))
    |> Stream.run()
  end
end

Step 3: Add a health check with pipeline metrics

# 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(),
      csv_parser: check_csv_parser(),
      memory: check_memory()
    }

    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: (if status == 200, do: "ok", else: "degraded"),
      checks: checks,
      pipeline_stats: MyApp.CsvMetrics.stats()
    }))
    |> 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, _} -> :error
    end
  rescue
    _ -> :error
  end

  defp check_csv_parser do
    # Verify NimbleCSV can parse a minimal CSV without error
    result =
      "a,b,c\n1,2,3\n"
      |> MyApp.CSV.Parser.parse_string(skip_headers: true)

    case result do
      [["1", "2", "3"]] -> :ok
      _ -> :error
    end
  rescue
    _ -> :error
  end

  defp check_memory do
    case :erlang.memory(:total) do
      total when total < 2_147_483_648 -> :ok  # Under 2 GB
      _ -> :error
    end
  end
end

Mount it before your router:

# lib/my_app_web/endpoint.ex
defmodule MyAppWeb.Endpoint do
  use Phoenix.Endpoint, otp_app: :my_app

  plug MyAppWeb.Plugs.HealthCheck
  plug MyAppWeb.Router
end

Test it:

curl http://localhost:4000/health
# {"status":"ok","checks":{"database":"ok","csv_parser":"ok","memory":"ok"},"pipeline_stats":{"imports_run":12,"rows_imported":48320,"import_errors":3}}

Step 4: Track pipeline metrics

# lib/my_app/csv_metrics.ex
defmodule MyApp.CsvMetrics do
  use Agent

  def start_link(_), do: Agent.start_link(fn ->
    %{
      imports_run: 0,
      rows_imported: 0,
      import_errors: 0,
      exports_run: 0,
      rows_exported: 0,
      last_import_at: nil,
      last_export_at: nil
    }
  end, name: __MODULE__)

  def record_import(rows_imported, errors) do
    Agent.update(__MODULE__, fn s ->
      s
      |> Map.update!(:imports_run, &(&1 + 1))
      |> Map.update!(:rows_imported, &(&1 + rows_imported))
      |> Map.update!(:import_errors, &(&1 + errors))
      |> Map.put(:last_import_at, DateTime.utc_now() |> DateTime.to_iso8601())
    end)
  end

  def record_export(rows_exported) do
    Agent.update(__MODULE__, fn s ->
      s
      |> Map.update!(:exports_run, &(&1 + 1))
      |> Map.update!(:rows_exported, &(&1 + rows_exported))
      |> Map.put(:last_export_at, DateTime.utc_now() |> DateTime.to_iso8601())
    end)
  end

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

Wire it into your importer and exporter:

# In ProductImporter.import_from_file/1, after the reduce:
{imported, errors} = ... # existing reduce result
MyApp.CsvMetrics.record_import(imported, errors)
{imported, errors}

# In OrderExporter.export_to_file/1, after stream run:
row_count = MyApp.Repo.aggregate(MyApp.Order, :count)
MyApp.CsvMetrics.record_export(row_count)

Add MyApp.CsvMetrics to your supervision tree:

children = [
  MyApp.Repo,
  MyApp.CsvMetrics,
  MyAppWeb.Endpoint
]

Step 5: Set up HTTP monitoring in Vigilmon

Point Vigilmon at your health endpoint:

  1. Sign up at vigilmon.online
  2. Click New Monitor → HTTP
  3. Enter https://yourdomain.com/health
  4. Set check interval: 1 minute (paid) or 5 minutes (free)
  5. Under Advanced, add a keyword check for "csv_parser":"ok" to verify parser availability
  6. Save

Why external monitoring matters for CSV pipeline applications:

A streaming CSV parse that silently fails mid-file leaves your database in an inconsistent state. Vigilmon's external check verifies the application is healthy enough to process new pipelines. Response time spikes on the health endpoint often coincide with large CSV operations consuming memory or blocking the database.

Step 6: Heartbeat monitoring for scheduled CSV jobs

# lib/my_app/workers/csv_import_worker.ex
defmodule MyApp.Workers.CsvImportWorker do
  use GenServer

  require Logger

  # Run daily at 2 AM — for hourly, adjust @interval to :timer.hours(1)
  @interval :timer.hours(24)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  @impl true
  def init(state) do
    schedule_tick()
    {:ok, state}
  end

  @impl true
  def handle_info(:import, state) do
    Logger.info("Starting scheduled CSV import...")

    case run_import() do
      {:ok, {imported, errors}} ->
        Logger.info("CSV import complete: #{imported} rows, #{errors} errors")
        ping_heartbeat(:import)
      {:error, reason} ->
        Logger.error("CSV import failed: #{inspect(reason)}")
    end

    schedule_tick()
    {:noreply, state}
  end

  defp schedule_tick, do: Process.send_after(self(), :import, @interval)

  defp run_import do
    path = Application.get_env(:my_app, :csv)[:import_path] || "/data/products.csv"

    if File.exists?(path) do
      result = MyApp.Import.ProductImporter.import_from_file(path)
      {:ok, result}
    else
      {:error, "Import file not found: #{path}"}
    end
  end

  defp ping_heartbeat(:import) do
    url = Application.get_env(:my_app, :vigilmon)[:csv_import_heartbeat_url]
    if url do
      case Req.get(url, receive_timeout: 10_000) do
        {:ok, _} -> :ok
        {:error, reason} -> Logger.warning("Import heartbeat failed: #{inspect(reason)}")
      end
    end
  end
end

# lib/my_app/workers/csv_export_worker.ex
defmodule MyApp.Workers.CsvExportWorker do
  use GenServer

  require Logger

  @interval :timer.hours(6)

  def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)

  @impl true
  def init(state) do
    schedule_tick()
    {:ok, state}
  end

  @impl true
  def handle_info(:export, state) do
    path = "/data/exports/orders_#{Date.utc_today()}.csv"

    case MyApp.Export.OrderExporter.export_to_file(path) do
      :ok ->
        Logger.info("CSV export written to #{path}")
        ping_heartbeat()
      {:error, reason} ->
        Logger.error("CSV export failed: #{inspect(reason)}")
    end

    schedule_tick()
    {:noreply, state}
  end

  defp schedule_tick, do: Process.send_after(self(), :export, @interval)

  defp ping_heartbeat do
    url = Application.get_env(:my_app, :vigilmon)[:csv_export_heartbeat_url]
    if url, do: Req.get(url, receive_timeout: 5_000)
  end
end

Add both workers to your supervision tree:

children = [
  MyApp.Repo,
  MyApp.CsvMetrics,
  MyApp.Workers.CsvImportWorker,
  MyApp.Workers.CsvExportWorker,
  MyAppWeb.Endpoint
]

In Vigilmon, create heartbeat monitors for each worker:

  1. New Monitor → Heartbeat
  2. For daily import: set interval to 25 hours (24h job with 1h grace)
  3. For 6-hour export: set interval to 7 hours (6h job with 1h grace)
  4. Copy ping URLs → set as environment variables

Step 7: Alerts via Slack

In Vigilmon, go to Notifications → New Channel → Slack and paste your incoming webhook URL.

Enable the channel on all monitors. CSV pipeline incidents often look like:

  • HTTP health check degraded (parser probe failed or memory pressure)
  • Import heartbeat missed (file missing, database down, or malformed CSV aborted the stream)
  • Export heartbeat missed (export worker crashed or disk full)

Configure Vigilmon to send all three alert types to your on-call channel so pipeline failures are caught before downstream consumers notice missing data.

Step 8: Status page

  1. Status Pages → New Status Page in Vigilmon
  2. Add your HTTP health monitor and both heartbeat monitors
  3. Share the URL in your README and runbook

README badge:

![Uptime](https://vigilmon.online/badge/your-monitor-slug)

What you've built

| What | How | |------|-----| | Health check plug | HealthCheck plug with live CSV parse probe | | Parser availability check | parse_string/2 round-trip in health check | | Pipeline throughput tracking | CsvMetrics Agent (rows, errors, timestamps) | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health with keyword check | | Slack pipeline failure alerts | Vigilmon Slack notification channel | | Import heartbeat | Heartbeat ping after each daily import run | | Export heartbeat | Heartbeat ping after each 6-hour export run | | Status page | Vigilmon public status page |

NimbleCSV moves your data. Vigilmon makes sure it keeps moving.

Next steps

  • Add per-file row-count validation after each import: if the parsed row count differs significantly from the expected count, skip the heartbeat ping and trigger an alert
  • Use Vigilmon's response time history to correlate large import runs with application latency spikes and size your batch processing windows accordingly
  • Set up a Vigilmon port monitor on your SFTP server if CSV files arrive via file transfer, and alert before the import worker even runs if the source is unavailable
  • Emit per-batch metrics to your observability stack to build a real-time import progress dashboard alongside Vigilmon's uptime monitoring

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