How to Monitor Quark with Vigilmon

Quark is a functional programming library for Elixir that brings combinators (compose, flip, id, constant), currying, partial application, and point-free utilities to the language. It lets you build composable data pipelines and transformation chains with the same expressive power you'd find in Haskell or Scala. When those pipelines run in production — transforming events, enriching API responses, or powering ETL jobs — failures in the composed pipeline may surface as silent nil returns or unexpected data shapes rather than loud exceptions.

Vigilmon adds external observability to applications built with Quark: HTTP uptime checks for your API layer, heartbeat monitors for data pipeline workers, and Slack alerts when composed transformations break the flow.

This tutorial covers:

• HTTP monitoring for Elixir APIs built with Quark-powered transformation pipelines
• Heartbeat monitoring for GenServer workers using functional composition
• Alerting on pipeline throughput gaps
• Health check patterns idiomatic to functional Elixir

What to Monitor in a Quark-Powered Application

| Layer | What it is | How to monitor | |---|---|---| | HTTP API | Phoenix or Plug API serving composed transformations | HTTP uptime check | | Data pipeline worker | GenServer applying Quark pipelines to event streams | Heartbeat per batch | | ETL job | Scheduled mix task using compose chains | Heartbeat after each run | | Queue consumer | Broadway or GenStage producer using Quark mappers | Heartbeat on consumer tick |

Step 1: Add a health check to your Quark-powered API

If your Phoenix or Plug API applies Quark transformation pipelines to requests, expose a health endpoint that exercises the composition chain:

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

  import Quark

  def index(conn, _params) do
    checks = %{
      database: check_database(),
      pipeline: check_pipeline()
    }

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

    conn
    |> put_status(status)
    |> json(%{status: status_text(status), checks: checks})
  end

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

  defp check_pipeline do
    # Exercise the core Quark composition chain used in hot paths
    transformer =
      compose(
        &normalize_value/1,
        &validate_format/1,
        &enrich_metadata/1
      )

    case transformer.(%{value: "health-check", source: :internal}) do
      {:ok, _} -> :ok
      _ -> :error
    end
  rescue
    _ -> :error
  end

  defp normalize_value(%{value: v} = input), do: {:ok, %{input | value: String.downcase(v)}}
  defp validate_format({:ok, %{value: v}} = result) when is_binary(v), do: result
  defp validate_format(_), do: :error
  defp enrich_metadata({:ok, input}), do: {:ok, Map.put(input, :checked_at, DateTime.utc_now())}
  defp enrich_metadata(err), do: err

  defp status_text(200), do: "ok"
  defp status_text(_), do: "degraded"
end

Add the route in your router:

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

In Vigilmon:

1. Sign in at vigilmon.online.
2. Click New Monitor → HTTP.
3. Enter https://yourdomain.com/health.
4. Set check interval to 60 seconds.
5. Expected status: 200.

Step 2: Add heartbeat monitoring for a pipeline worker

GenServer workers that apply Quark's compose and curry chains to event batches can fail silently — the process stays alive but the composition produces unexpected output or stalls awaiting external data. Ping Vigilmon on every successful batch:

# lib/my_app/event_pipeline.ex
defmodule MyApp.EventPipeline do
  use GenServer
  require Logger

  import Quark

  @batch_interval_ms 30_000

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

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

  @impl true
  def handle_info(:process_batch, state) do
    case run_pipeline() do
      {:ok, count} ->
        Logger.debug("Pipeline processed #{count} events")
        ping_vigilmon()

      {:error, reason} ->
        Logger.error("Pipeline failed: #{inspect(reason)}")
        # No ping → Vigilmon alerts after grace period
    end

    schedule_batch()
    {:noreply, state}
  end

  defp run_pipeline do
    transform =
      compose(
        &filter_valid/1,
        &enrich_events/1,
        &emit_to_sink/1
      )

    events = MyApp.EventQueue.dequeue(100)
    results = Enum.map(events, &apply(transform, [&1]))
    successful = Enum.count(results, &match?({:ok, _}, &1))

    {:ok, successful}
  rescue
    e -> {:error, e}
  end

  defp ping_vigilmon do
    url = Application.get_env(:my_app, :vigilmon_heartbeat_url)

    if url do
      case :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5000}], []) do
        {:ok, _} -> :ok
        {:error, reason} -> Logger.warning("Heartbeat failed: #{inspect(reason)}")
      end
    end
  end

  defp schedule_batch, do: Process.send_after(self(), :process_batch, @batch_interval_ms)

  defp filter_valid(event), do: if(event[:valid], do: {:ok, event}, else: {:skip, event})
  defp enrich_events({:ok, event}), do: {:ok, Map.put(event, :enriched, true)}
  defp enrich_events(other), do: other
  defp emit_to_sink({:ok, event}), do: MyApp.Sink.emit(event)
  defp emit_to_sink(other), do: other
end

Configure the heartbeat URL:

# config/runtime.exs
config :my_app,
  vigilmon_heartbeat_url: System.get_env("VIGILMON_HEARTBEAT_URL")

In Vigilmon, create a Heartbeat Monitor:

1. Click New Monitor → Heartbeat.
2. Name it Event pipeline worker.
3. Set interval to 2 minutes (4× the 30-second batch cycle, giving two missed batches before alert).
4. Copy the heartbeat URL into your deployment config.

Step 3: Monitor scheduled ETL jobs using Quark pipelines

If you run scheduled mix tasks or Oban jobs that apply Quark transformation chains to data:

# lib/my_app/workers/etl_worker.ex
defmodule MyApp.Workers.EtlWorker do
  use Oban.Worker, queue: :etl

  import Quark

  require Logger

  @impl Oban.Worker
  def perform(%Oban.Job{}) do
    transform_pipeline =
      compose(
        &extract_records/1,
        &transform_records/1,
        &load_records/1
      )

    case transform_pipeline.(:source) do
      {:ok, count} ->
        Logger.info("ETL completed: #{count} records")
        ping_heartbeat()
        :ok

      {:error, reason} ->
        Logger.error("ETL failed: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp ping_heartbeat do
    url = System.get_env("VIGILMON_ETL_HEARTBEAT_URL")
    if url, do: :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5000}], [])
  end

  defp extract_records(_source), do: {:ok, []}   # your extraction logic
  defp transform_records({:ok, records}), do: {:ok, Enum.map(records, & &1)}
  defp load_records({:ok, records}), do: {:ok, length(records)}
end

Set the Oban job schedule and create a Vigilmon heartbeat monitor with a grace period matching your job frequency.

Step 4: Configure alerts

In Vigilmon go to Alerts and configure notification channels:

• Slack (#data-engineering) — pipeline and ETL failures
• Email — backend engineering team
• PagerDuty — for pipelines feeding customer-facing features

Recommended alert policies for Quark-powered applications:

| Monitor | Condition | Action | |---|---|---| | HTTP health check | Non-200 or timeout | Alert on-call immediately | | Event pipeline heartbeat | Missed ≥ 2× | Alert data engineering | | ETL job heartbeat | Missed ≥ 1× | Alert backend team | | Response time | > 2 s | Warning — composition chain slow |

Step 5: Status page

Create a Vigilmon status page for your data platform:

1. Go to Status Pages → New Page.
2. Add all pipeline and HTTP monitors.
3. Set visibility to Team for internal data engineering use.
4. Share the URL in your ops runbook.

Key Metrics to Watch

| Metric | Threshold | Meaning | |---|---|---| | HTTP health | Must return 200 | API and composition chain healthy | | Pipeline heartbeat | < 2 missed | Worker processing events | | ETL job heartbeat | 0 missed per schedule | Job running on time | | HTTP response time | < 500 ms | Composition not causing latency | | Error rate | < 0.1% | Transformation pipeline stable |

Why Monitor Quark-Powered Applications?

Functional composition pipelines fail in subtle ways:

• Silent nil propagation — a compose chain that receives unexpected input returns nil through every stage rather than raising
• Partial application mismatch — curried functions called with wrong arity crash at runtime, not compile time
• Worker process health vs. work health — the GenServer stays alive and healthy even when the composed transformation produces zero output
• Scheduled job drift — ETL jobs that should run hourly may silently stop if their queue isn't being polled

Vigilmon's external heartbeat pattern catches all of these: the pipeline only pings when it successfully completes useful work.

Conclusion

Quark makes Elixir pipelines composable and expressive. Vigilmon makes those pipelines observable. With HTTP monitoring on your API health endpoint, heartbeat monitors on each pipeline worker, and threshold-aware alerts for ETL jobs, you have full visibility into whether your functional composition chains are doing real work — not just staying alive.

Get started free at vigilmon.online.