How to Monitor Membrane Framework with Vigilmon

title: How to Monitor Membrane Framework with Vigilmon published: true description: Monitor your Elixir multimedia pipelines built with Membrane Framework — track pipeline health, element throughput, buffer backpressure, and get alerted when streams stall or drop. tags: elixir, membrane, multimedia, monitoring

Membrane Framework is Elixir's composable multimedia processing toolkit. You assemble pipelines from modular elements — sources, filters, sinks — to transcode video, relay audio streams, record calls, or build WebRTC applications. Because each element is a supervised process, BEAM keeps individual pieces alive. But BEAM cannot tell you when a pipeline processes zero frames for thirty seconds, when a buffer overflows and your audio sink starts dropping packets, or when your WebRTC negotiation element stalls mid-handshake.

That requires external monitoring: a health check that interrogates your pipeline state, a heartbeat that fires every time a stream successfully completes, and Vigilmon to alert your team when something breaks.

Why Monitor Membrane Pipelines?

Membrane elements are OTP processes. A crashed element gets restarted by its supervisor — but it restarts empty, with no stream in flight. The pipeline keeps running, all processes stay up, and your monitoring dashboard shows green. The only signal something is wrong is that users stopped receiving audio or their video session silently froze.

Common failure modes that need external monitoring:

• Pipeline starts but produces no output — a source element is connected but not playing; no frames flow through the pipeline
• Buffer backpressure — a slow sink causes upstream buffers to fill; elements start dropping packets without logging errors
• Stream stall after reconnect — a WebRTC peer disconnects and reconnects; the pipeline restarts the source element but fails to re-link the downstream sink
• Transcoding error loop — a corrupted segment causes an element to crash and restart repeatedly; throughput drops while supervisor tree stays healthy
• Heartbeat stream stops unexpectedly — a continuous broadcast job that runs indefinitely stops emitting data without explicitly crashing

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Pipeline uptime | Whether the top-level pipeline process is running | | Element throughput (buffers/sec) | Whether data is actually flowing through each element | | Buffer queue depth | Whether a slow sink is causing upstream backpressure | | Stream restart count | How often elements crash and recover under supervision | | Heartbeat ping rate | Whether a continuous broadcast job is still alive | | Health endpoint response time | Overall pipeline process and Erlang VM health |

Step 1: Add a Health Check to Your Pipeline

Expose a health endpoint that queries your pipeline running state:

# lib/my_app/pipeline_health.ex
defmodule MyApp.PipelineHealth do
  def check do
    case check_pipeline_running() do
      :ok ->
        %{status: "ok", checks: %{pipeline: "ok"}}
      {:error, reason} ->
        %{status: "error", checks: %{pipeline: to_string(reason)}}
    end
  end

  defp check_pipeline_running do
    case Process.whereis(MyApp.Pipeline) do
      nil -> {:error, :not_running}
      pid -> if Process.alive?(pid), do: :ok, else: {:error, :not_alive}
    end
  end
end

Wire it into your Phoenix endpoint as a plug:

# 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
    result = MyApp.PipelineHealth.check()
    status = if result.status == "ok", do: 200, else: 503

    conn
    |> put_resp_content_type("application/json")
    |> send_resp(status, Jason.encode!(result))
    |> halt()
  end

  def call(conn, _opts), do: conn
end

Register the plug before your router in endpoint.ex:

plug MyAppWeb.Plugs.HealthCheck
plug MyAppWeb.Router

Test it:

mix phx.server
curl http://localhost:4000/health
# {"status":"ok","checks":{"pipeline":"ok"}}

Step 2: Track Element Throughput with Telemetry

Membrane elements emit telemetry events. Attach a handler to count buffer deliveries and detect stalls:

# lib/my_app/pipeline_telemetry.ex
defmodule MyApp.PipelineTelemetry do
  require Logger

  def attach do
    :telemetry.attach_many(
      "membrane-pipeline-metrics",
      [
        [:membrane, :element, :handle_buffer, :stop],
        [:membrane, :pipeline, :handle_crash_group_down, :stop]
      ],
      &handle_event/4,
      nil
    )
  end

  def handle_event([:membrane, :element, :handle_buffer, :stop], measurements, metadata, _) do
    :telemetry.execute(
      [:my_app, :membrane, :buffer_processed],
      %{duration: measurements[:duration]},
      %{element: metadata[:element_module]}
    )
  end

  def handle_event([:membrane, :pipeline, :handle_crash_group_down, :stop], _m, metadata, _) do
    Logger.error("Membrane crash group down",
      group: metadata[:crash_group_name],
      pipeline: metadata[:pipeline]
    )

    :telemetry.execute(
      [:my_app, :membrane, :crash_group_down],
      %{count: 1},
      %{group: metadata[:crash_group_name]}
    )
  end
end

Attach it in your application start:

# lib/my_app/application.ex
def start(_type, _args) do
  MyApp.PipelineTelemetry.attach()
  # ... rest of your supervision tree
end

Step 3: Add a Heartbeat for Continuous Broadcast Jobs

For a pipeline that runs continuously — a live stream, a recording job, a transcoding worker — add a heartbeat that pings Vigilmon on each successful output segment:

# lib/my_app/pipeline.ex
defmodule MyApp.Pipeline do
  use Membrane.Pipeline

  require Logger

  @impl true
  def handle_init(_ctx, options) do
    spec = [
      child(:source, %Membrane.File.Source{location: options.input_path}),
      child(:decoder, Membrane.H264.FFmpeg.Decoder),
      child(:encoder, %Membrane.H264.FFmpeg.Encoder{preset: :fast}),
      child(:sink, %Membrane.File.Sink{location: options.output_path})
    ]

    {[spec: spec], %{output_path: options.output_path, segment_count: 0}}
  end

  @impl true
  def handle_element_end_of_stream(:sink, _pad, _ctx, state) do
    new_count = state.segment_count + 1
    ping_heartbeat()
    Logger.info("Segment #{new_count} complete")
    {[], %{state | segment_count: new_count}}
  end

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

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

Configure the heartbeat URL:

# config/runtime.exs
config :my_app, :vigilmon,
  pipeline_heartbeat_url: System.get_env("VIGILMON_PIPELINE_HEARTBEAT_URL")

Step 4: Set Up HTTP Monitoring in Vigilmon

1. Sign up at vigilmon.online
2. Click New Monitor → HTTP
3. Enter https://yourdomain.com/health
4. Set check interval: 1 minute
5. Save

This catches cases where the pipeline process crashes entirely and the health endpoint returns 503.

Step 5: Create a Heartbeat Monitor for Your Pipeline

1. In Vigilmon, click New Monitor → Heartbeat
2. Name it Membrane Pipeline — Segment Heartbeat
3. Set expected interval to 2 minutes (or match your pipeline segment cadence)
4. Copy the URL and set VIGILMON_PIPELINE_HEARTBEAT_URL in your environment
5. Enable Slack notifications

If your pipeline stalls — source element runs out of data unexpectedly, an encoder crashes mid-segment, or a sink fills up and blocks — no heartbeat is sent and Vigilmon alerts you within the missed-interval window.

Step 6: Monitor Buffer Backpressure

Buffer backpressure is the subtlest Membrane failure mode. A slow sink causes upstream elements to accumulate buffers, eventually dropping packets without raising a supervisor alarm.

Add a periodic watchdog:

# lib/my_app/pipeline_watchdog.ex
defmodule MyApp.PipelineWatchdog do
  use GenServer

  require Logger

  @check_interval_ms 10_000
  @buffer_warn_threshold 1_000

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

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

  @impl true
  def handle_info(:check, state) do
    check_pipeline_buffers()
    schedule_check()
    {:noreply, state}
  end

  defp check_pipeline_buffers do
    case Process.whereis(MyApp.Pipeline) do
      nil ->
        Logger.warning("PipelineWatchdog: pipeline not running")

      pid ->
        {:message_queue_len, queue_len} = Process.info(pid, :message_queue_len)

        if queue_len > @buffer_warn_threshold do
          Logger.warning("Pipeline message queue high",
            queue_length: queue_len,
            threshold: @buffer_warn_threshold
          )
        end
    end
  end

  defp schedule_check, do: Process.send_after(self(), :check, @check_interval_ms)
end

Add the watchdog to your supervision tree alongside the pipeline.

Step 7: Alerting

In Vigilmon, go to Notifications → New Channel → Slack and connect your alerts channel.

When the health endpoint goes down:

🔴 DOWN: yourdomain.com/health
Status: 503 Service Unavailable
Detected from: EU-West, US-East

When the heartbeat misses:

🔴 MISSED: Membrane Pipeline — Segment Heartbeat
Last ping: 6 minutes ago (expected every 2 minutes)
Action: Check pipeline — possible stall or encoder crash

What You Built

| What | How | |------|-----| | Pipeline health endpoint | Custom plug querying pipeline PID | | Element throughput tracking | Telemetry attachment on buffer events | | Crash group alerting | Telemetry on handle_crash_group_down | | Continuous stream heartbeat | Ping on each segment completion | | Buffer backpressure watchdog | GenServer checking message queue depth | | HTTP uptime monitoring | Vigilmon HTTP monitor → /health | | Heartbeat monitoring | Vigilmon heartbeat — alerts when pipeline stalls | | Slack alerting | Vigilmon Slack notification channel |

BEAM keeps your Membrane elements supervised. Vigilmon tells you when the data stops flowing through them.

Next Steps

• Add per-element telemetry to track throughput at each pipeline stage separately
• Use Vigilmon response time history to catch slow transcoding before it causes downstream buffering
• Create separate heartbeat monitors for each pipeline variant if you run parallel transcoding jobs
• Expose buffer queue depth in your /health response for richer monitoring context

Get started free at vigilmon.online.