How to Monitor Saxy with Vigilmon

Saxy is a fast, spec-compliant SAX-style XML parser for Elixir. Unlike DOM parsers that load the entire document into memory, Saxy streams events (start element, text, end element) to a handler module as it reads — making it practical for processing gigabyte-scale SOAP payloads, RSS feeds, and enterprise data exports without exhausting heap.

Production XML pipelines built with Saxy have a specific failure profile: feeds change format without warning, upstream suppliers go down, and parse errors in the middle of a 500 MB file leave partial imports that corrupt your data model. These failures don't raise GenServer crashes — they surface as error tuples in your pipeline code, often swallowed by a retry loop that eventually gives up silently.

External monitoring is how you detect that your XML ingestion stopped hours before your on-call team notices an empty database table.

This tutorial covers:

• A health check endpoint that exposes XML pipeline status
• HTTP uptime monitoring with Vigilmon
• Heartbeat monitoring for scheduled feed ingestion jobs
• Alerts for parse failures, feed timeouts, and throughput drops

Why Monitor Saxy?

XML pipelines fail in ways that don't crash processes:

| Failure mode | Symptom without monitoring | |---|---| | Upstream feed server returns 503 | Pipeline silently skips; stale data accumulates | | Feed format change breaks SAX handler | {:error, reason} tuples; records stop importing | | Parse error mid-document | Partial import; data inconsistency | | Feed size grows beyond memory budget | Streaming buffer pressure; node slowdown | | SSL certificate expiry on feed endpoint | HTTP fetch fails; pipeline queues backlog | | Cron job not scheduled after deploy | Ingestion stops entirely; no alerts |

Step 1: SAX handler with error tracking

Implement a handler module that counts parse events and tracks errors. Saxy calls your handle_event/2 callback for each XML event:

# lib/my_app/xml/feed_handler.ex
defmodule MyApp.Xml.FeedHandler do
  @behaviour Saxy.Handler

  defstruct items: [], current_item: nil, depth: 0, errors: 0, events: 0

  @impl Saxy.Handler
  def handle_event(:start_document, _prolog, state) do
    {:ok, state}
  end

  def handle_event(:end_document, _data, state) do
    {:ok, state}
  end

  def handle_event(:start_element, {tag, attributes}, state) do
    state = %{state | events: state.events + 1, depth: state.depth + 1}

    state =
      case tag do
        "item" -> %{state | current_item: %{attributes: Map.new(attributes)}}
        _ -> state
      end

    {:ok, state}
  end

  def handle_event(:end_element, "item", %{current_item: item} = state) when not is_nil(item) do
    {:ok, %{state | items: [item | state.items], current_item: nil, depth: state.depth - 1}}
  end

  def handle_event(:end_element, _tag, state) do
    {:ok, %{state | depth: state.depth - 1}}
  end

  def handle_event(:characters, chars, %{current_item: item} = state) when not is_nil(item) do
    # Accumulate text content into current item
    {:ok, %{state | current_item: Map.put(item, :text, String.trim(chars))}}
  end

  def handle_event(:characters, _chars, state) do
    {:ok, state}
  end
end

Parse a feed with streaming:

defmodule MyApp.Xml.FeedParser do
  alias MyApp.Xml.FeedHandler

  def parse_stream(stream) do
    initial_state = %FeedHandler{}

    case Saxy.parse_stream(stream, FeedHandler, initial_state) do
      {:ok, final_state} ->
        {:ok, %{
          items: Enum.reverse(final_state.items),
          events_processed: final_state.events,
          errors: final_state.errors
        }}

      {:error, exception} ->
        {:error, Exception.message(exception)}
    end
  end
end

Step 2: Pipeline monitor GenServer

Track feed ingestion stats so the health endpoint can expose them:

# lib/my_app/xml/pipeline_monitor.ex
defmodule MyApp.Xml.PipelineMonitor do
  use GenServer

  def start_link(_), do: GenServer.start_link(__MODULE__, initial_state(), name: __MODULE__)

  def record_success(feed_name, item_count, duration_ms) do
    GenServer.cast(__MODULE__, {:success, feed_name, item_count, duration_ms})
  end

  def record_failure(feed_name, reason) do
    GenServer.cast(__MODULE__, {:failure, feed_name, reason})
  end

  def stats, do: GenServer.call(__MODULE__, :stats)

  defp initial_state do
    %{
      total_runs: 0,
      total_failures: 0,
      total_items_processed: 0,
      last_failure_reason: nil,
      last_run_at: nil
    }
  end

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

  @impl true
  def handle_cast({:success, _feed, items, _duration}, state) do
    {:noreply, %{state |
      total_runs: state.total_runs + 1,
      total_items_processed: state.total_items_processed + items,
      last_run_at: DateTime.utc_now()
    }}
  end

  def handle_cast({:failure, _feed, reason}, state) do
    {:noreply, %{state |
      total_runs: state.total_runs + 1,
      total_failures: state.total_failures + 1,
      last_failure_reason: inspect(reason),
      last_run_at: DateTime.utc_now()
    }}
  end

  @impl true
  def handle_call(:stats, _from, state) do
    failure_rate =
      if state.total_runs > 0,
        do: Float.round(state.total_failures / state.total_runs * 100, 1),
        else: 0.0

    {:reply, Map.put(state, :failure_rate_pct, failure_rate), state}
  end
end

Add to supervision tree:

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

Step 3: Health check endpoint

# 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
    pipeline_stats = MyApp.Xml.PipelineMonitor.stats()
    db_ok = check_database()

    pipeline_ok =
      pipeline_stats.failure_rate_pct < 25.0 and
      pipeline_stats.total_failures < 10

    status = if db_ok and pipeline_ok, do: 200, else: 503

    body = Jason.encode!(%{
      status: if(status == 200, do: "ok", else: "degraded"),
      checks: %{
        database: if(db_ok, do: "ok", else: "error"),
        xml_pipeline: if(pipeline_ok, do: "ok", else: "degraded"),
        pipeline_stats: pipeline_stats
      }
    })

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, body)
    |> halt()
  end

  def call(conn, _opts), do: conn

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

Plug it before the router:

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

Step 4: Set up HTTP monitoring in Vigilmon

1. Log in at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health
4. Set interval: 1 minute
5. Add a keyword check: alert when "degraded" or "error" appears in the response
6. Set threshold: 2 consecutive failures

Step 5: Heartbeat monitoring for feed ingestion jobs

Feed ingestion is the most important process to heartbeat-monitor — it runs on a schedule and fails silently when skipped:

# lib/my_app/workers/feed_ingestion_worker.ex
defmodule MyApp.Workers.FeedIngestionWorker do
  use Oban.Worker, queue: :feeds, max_attempts: 3

  @heartbeat_url System.get_env("VIGILMON_FEED_INGESTION_HEARTBEAT_URL")

  @impl Oban.Worker
  def perform(%Oban.Job{args: %{"feed_name" => feed_name, "url" => url}}) do
    start_time = System.monotonic_time(:millisecond)

    with {:ok, response} <- fetch_feed(url),
         {:ok, result} <- parse_feed(response.body),
         :ok <- import_items(result.items) do
      duration_ms = System.monotonic_time(:millisecond) - start_time
      MyApp.Xml.PipelineMonitor.record_success(feed_name, length(result.items), duration_ms)
      ping_heartbeat()
      :ok
    else
      {:error, reason} ->
        MyApp.Xml.PipelineMonitor.record_failure(feed_name, reason)
        {:error, reason}
    end
  end

  defp fetch_feed(url) do
    case Req.get(url, receive_timeout: 30_000, max_retries: 0) do
      {:ok, %{status: 200} = response} -> {:ok, response}
      {:ok, response} -> {:error, {:http_error, response.status}}
      {:error, reason} -> {:error, {:fetch_failed, reason}}
    end
  end

  defp parse_feed(body) do
    stream = Stream.resource(
      fn -> {:ok, body} end,
      fn
        {:ok, data} -> {[data], :done}
        :done -> {:halt, :done}
      end,
      fn _ -> :ok end
    )
    MyApp.Xml.FeedParser.parse_stream(stream)
  end

  defp import_items(items) do
    Enum.each(items, fn item ->
      # your import logic
      _ = item
    end)
    :ok
  end

  defp ping_heartbeat do
    if @heartbeat_url do
      Task.start(fn ->
        :httpc.request(:get, {String.to_charlist(@heartbeat_url), []}, [{:timeout, 5000}], [])
      end)
    end
  end
end

Schedule it with Oban:

# config/config.exs
config :my_app, Oban,
  queues: [feeds: 5],
  plugins: [
    {Oban.Plugins.Cron,
     crontab: [
       {"0 * * * *", MyApp.Workers.FeedIngestionWorker,
        args: %{"feed_name" => "product_catalog", "url" => System.get_env("PRODUCT_FEED_URL")}}
     ]}
  ]

Create the heartbeat in Vigilmon:

1. Go to Monitors → New → Heartbeat
2. Set grace period: 75 minutes (hourly job + 15 minute buffer for slow feeds)
3. Copy the ping URL into VIGILMON_FEED_INGESTION_HEARTBEAT_URL

Step 6: Monitor upstream feed endpoints directly

Add a separate Vigilmon HTTP monitor for each upstream feed URL:

• URL: https://supplier.example.com/feed.xml
• Expected status: 200
• Keyword check: look for a root XML tag like <catalog or <rss

This tells you whether a failure is on your side (parser) or the supplier's side (upstream down), without having to dig through logs.

Step 7: Key metrics to alert on

| Metric | Alert threshold | Why | |---|---|---| | /health HTTP status | Non-200 for 2 checks | Full application down | | xml_pipeline keyword "degraded" | Any match | Parse failures accumulating | | failure_rate_pct | > 25% | Majority of ingestion runs failing | | Feed ingestion heartbeat silence | Grace period exceeded | Cron job not running | | Upstream feed endpoint status | Non-200 | Supplier outage affecting your data | | Response time on /health | > 5 s | Database or pipeline coordination slow |

Step 8: Status page

1. In Vigilmon, go to Status Pages → New
2. Add: health endpoint, feed ingestion heartbeat, upstream feed monitors
3. Label them: "Application (Web)", "Feed Ingestion", "Product Catalog Feed", "Pricing Feed"
4. Share the URL with your data team so they can self-serve when imports look stale

Conclusion

Saxy makes streaming large XML documents practical in Elixir, but the pipeline that drives it — HTTP fetches, cron scheduling, SAX handler logic, database imports — can fail at any layer without raising an exception you'll see. With Vigilmon you get:

• HTTP uptime checks that surface health endpoint degradation as parse failure rates rise
• Upstream feed monitors that separate supplier outages from your own parse bugs
• Heartbeat monitors that detect silent cron failures before your data goes stale

Start with a heartbeat for your most critical feed ingestion job and a direct monitor on the upstream URL — those two together catch the majority of XML pipeline failures in production.

Get started free at vigilmon.online.