tutorial

How to Monitor :pg (Erlang Process Groups) with Vigilmon

Track :pg distributed process group membership, publish-subscribe event flow, and cross-node group consistency in your Erlang/Elixir application, and use Vigilmon to alert when process group distribution degrades.

How to Monitor :pg (Erlang Process Groups) with Vigilmon

:pg is a built-in Erlang/OTP module introduced in OTP 23 that manages distributed process groups across cluster nodes. Unlike external libraries, :pg ships with every Erlang installation and provides a lightweight publish-subscribe mechanism: any number of processes can join a named group on any node, and senders can broadcast to all members of a group across the entire cluster — without knowing where those processes live.

:pg is used in Phoenix PubSub, distributed event broadcasting, fanout messaging systems, and anywhere you need pub-sub semantics across an Elixir or Erlang cluster without external brokers. Because :pg is deeply integrated with OTP's distribution layer, its failure modes are subtle: group membership may diverge between nodes during network partitions, messages may not reach all group members during a rebalancing period, and the :pg scope process itself can crash without an obvious external signal.

Vigilmon lets you expose :pg group health through a health endpoint and alert when process group distribution degrades.

Why Monitor :pg?

:pg failures affect pub-sub delivery across the cluster:

  • Group membership divergence — during a network partition, nodes build separate group membership lists; messages sent after partition only reach members on the sender's partition; on heal, membership re-syncs but in-flight messages are already lost
  • Scope process crash — :pg groups belong to a scope (defaulting to :pg); the scope is supervised, but a crash triggers full membership re-synchronization across all nodes, which takes time proportional to cluster size
  • Process not in group after join — a race between group join and the first message dispatch can miss a newly joined process; senders see a smaller group than expected
  • Silent delivery failure — :pg.get_members/2 returns PIDs, but :erlang.send/2 to a dead PID silently drops the message; without delivery confirmation, missed messages are invisible
  • Empty group on new node — a new node starts with empty group membership; until existing nodes sync their members to the new node, broadcasts from the new node miss all remote subscribers
  • High group churn — a large number of join/leave events per second (e.g., from many short-lived processes) overwhelms the sync protocol and creates persistent membership lag

Key Metrics to Track

| Metric | What it tells you | |--------|-------------------| | Group member count per node | Whether membership view is consistent across nodes | | Expected vs actual members | Whether all expected subscribers are in the group | | Scope process uptime | Whether the local :pg scope process is running | | Message delivery confirmation rate | What fraction of broadcast messages reach intended recipients | | Join/leave event rate | Whether group churn is within acceptable bounds | | Cross-node member visibility | Whether a member joined on node A is visible from node B | | Heartbeat from group member | Whether a specific critical subscriber is alive and in the group |

Step 1: Understand :pg Basics

:pg is available in Erlang/OTP 23+ and Elixir 1.12+. No dependencies required.

# Start the default scope (already started by OTP in most apps)
:pg.start_link()

# Or start a custom named scope for your application
:pg.start_link(:my_app_pubsub)

# Join a group
:pg.join(:my_app_pubsub, :order_events, self())

# Get all members of a group (across all nodes)
members = :pg.get_members(:my_app_pubsub, :order_events)

# Get local members only (on this node)
local_members = :pg.get_local_members(:my_app_pubsub, :order_events)

# Leave a group
:pg.leave(:my_app_pubsub, :order_events, self())

# Broadcast to all members
def broadcast(scope, group, message) do
  scope
  |> :pg.get_members(group)
  |> Enum.each(&send(&1, message))
end

Add :pg scope to your supervision tree for lifecycle control:

# lib/my_app/application.ex
def start(_type, _args) do
  children = [
    {MyApp.PubSubScope, []},
    MyApp.Repo,
    MyAppWeb.Endpoint,
    MyApp.PgMonitor
  ]

  Supervisor.start_link(children, strategy: :one_for_one, name: MyApp.Supervisor)
end

# lib/my_app/pub_sub_scope.ex
defmodule MyApp.PubSubScope do
  def child_spec(_) do
    %{
      id: __MODULE__,
      start: {:pg, :start_link, [:my_app_pubsub]},
      type: :worker,
      restart: :permanent
    }
  end
end

Step 2: Monitor Group Membership Health

# lib/my_app/pg_monitor.ex
defmodule MyApp.PgMonitor do
  use GenServer
  require Logger

  @scope :my_app_pubsub
  @monitored_groups [:order_events, :user_notifications, :system_alerts]
  @poll_interval :timer.seconds(30)

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

  def init(state) do
    schedule()
    {:ok, state}
  end

  def handle_info(:poll, state) do
    report_scope_health()
    report_group_health()
    schedule()
    {:noreply, state}
  end

  defp report_scope_health do
    scope_pid = :pg.which_groups(@scope) |> then(fn _ -> Process.whereis(@scope) end)

    alive = is_pid(scope_pid) and Process.alive?(scope_pid)

    :telemetry.execute(
      [:my_app, :pg, :scope],
      %{alive: if(alive, do: 1, else: 0)},
      %{scope: @scope, node: node()}
    )

    unless alive do
      Logger.error("[PgMonitor] :pg scope #{@scope} is not alive on #{node()}")
    end
  rescue
    e -> Logger.warning("[PgMonitor] scope check error: #{inspect(e)}")
  end

  defp report_group_health do
    for group <- @monitored_groups do
      all_members = :pg.get_members(@scope, group)
      local_members = :pg.get_local_members(@scope, group)
      remote_count = length(all_members) - length(local_members)

      :telemetry.execute(
        [:my_app, :pg, :group],
        %{
          total_members: length(all_members),
          local_members: length(local_members),
          remote_members: remote_count
        },
        %{scope: @scope, group: group, node: node()}
      )

      Logger.debug(
        "[PgMonitor] group #{group}: #{length(all_members)} total (#{length(local_members)} local)"
      )
    end
  rescue
    e -> Logger.warning("[PgMonitor] group health check error: #{inspect(e)}")
  end

  defp schedule, do: Process.send_after(self(), :poll, @poll_interval)
end

Step 3: Test Cross-Node Visibility

Add a probe that verifies a process joined on this node is visible from the node's perspective, and optionally verifies it from another node:

# lib/my_app/pg_probe.ex
defmodule MyApp.PgProbe do
  require Logger

  @scope :my_app_pubsub
  @probe_group :health_probe_group

  @doc "Join the probe group, verify local membership, then leave. Returns latency_ms or error."
  def local_roundtrip do
    start = System.monotonic_time(:millisecond)
    pid = self()

    :ok = :pg.join(@scope, @probe_group, pid)

    result =
      case :pg.get_local_members(@scope, @probe_group) do
        members when is_list(members) ->
          if Enum.member?(members, pid) do
            {:ok, System.monotonic_time(:millisecond) - start}
          else
            {:error, "joined group but self not in local members"}
          end

        error ->
          {:error, "get_local_members failed: #{inspect(error)}"}
      end

    :pg.leave(@scope, @probe_group, pid)
    result
  rescue
    e ->
      Logger.warning("[PgProbe] exception: #{inspect(e)}")
      {:error, inspect(e)}
  end

  @doc "Check group visibility from other nodes via RPC."
  def cross_node_visibility(group) do
    local_count = length(:pg.get_local_members(@scope, group))
    all_count = length(:pg.get_members(@scope, group))
    remote_count = all_count - local_count

    %{
      local: local_count,
      remote: remote_count,
      total: all_count,
      nodes_in_cluster: length(Node.list()) + 1
    }
  rescue
    e -> %{error: inspect(e)}
  end
end

Step 4: Build a :pg Health Endpoint

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

  @scope :my_app_pubsub
  @critical_group :order_events

  def index(conn, _params) do
    scope_check = check_scope()
    probe_check = check_probe()
    group_check = check_group(@critical_group)

    all_ok =
      scope_check.status == "ok" and
        probe_check.status == "ok" and
        group_check.status == "ok"

    http_status = if all_ok, do: 200, else: 503

    conn
    |> put_status(http_status)
    |> json(%{
      status: if(all_ok, do: "ok", else: "degraded"),
      node: node(),
      checks: %{
        pg_scope: scope_check,
        probe: probe_check,
        critical_group: group_check
      }
    })
  end

  defp check_scope do
    groups = :pg.which_groups(@scope)
    scope_pid = Process.whereis(@scope)

    case scope_pid do
      pid when is_pid(pid) ->
        %{status: "ok", group_count: length(groups)}

      nil ->
        %{status: "error", reason: "scope process not registered"}
    end
  rescue
    e -> %{status: "error", reason: inspect(e)}
  end

  defp check_probe do
    case MyApp.PgProbe.local_roundtrip() do
      {:ok, latency_ms} -> %{status: "ok", latency_ms: latency_ms}
      {:error, reason} -> %{status: "error", reason: reason}
    end
  end

  defp check_group(group) do
    visibility = MyApp.PgProbe.cross_node_visibility(group)

    case visibility do
      %{error: reason} ->
        %{status: "error", reason: reason}

      info ->
        %{status: "ok"}
        |> Map.merge(info)
    end
  end
end

Register the route:

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

Step 5: Heartbeat Monitor

# lib/my_app/pg_heartbeat.ex
defmodule MyApp.PgHeartbeat do
  use GenServer
  require Logger

  @interval :timer.minutes(1)

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

  def init(state) do
    schedule()
    {:ok, state}
  end

  def handle_info(:ping, state) do
    url = System.get_env("VIGILMON_PG_HEARTBEAT_URL")

    case MyApp.PgProbe.local_roundtrip() do
      {:ok, _latency_ms} ->
        if url do
          :httpc.request(:get, {String.to_charlist(url), []}, [{:timeout, 5_000}], [])
          Logger.debug("[PgHeartbeat] ping sent from #{node()}")
        end

      {:error, reason} ->
        Logger.error("[PgHeartbeat] skipping ping — probe failed: #{reason}")
    end

    schedule()
    {:noreply, state}
  end

  defp schedule, do: Process.send_after(self(), :ping, @interval)
end

Step 6: Set Up Monitoring in Vigilmon

HTTP Monitor (:pg health endpoint)

  1. Sign in at vigilmon.online
  2. Click New Monitor → HTTP
  3. Set URL to https://your-app.example.com/health
  4. Expected status: 200
  5. Body must contain: "pg_scope":{"status":"ok"}
  6. Check interval: 60 seconds

Heartbeat Monitor (:pg liveness)

  1. Click New Monitor → Heartbeat
  2. Name: :pg Scope Liveness — [node]
  3. Expected interval: 2 minutes
  4. Copy the URL → set as VIGILMON_PG_HEARTBEAT_URL in your environment

A missed heartbeat means the :pg scope can't process a join/lookup round-trip — any pub-sub features on this node are broken.

Step 7: Alerting

In Vigilmon, configure Notifications → New Channel:

Slack for :pg degradation:

🔴 DOWN: :pg Scope — MyApp
Monitor: /health returned pg_scope.status = "error"
Check: Process.whereis(:my_app_pubsub) in remsh on the affected node
Action: if scope is dead, OTP supervisor should restart it — check supervisor tree for repeated crashes

PagerDuty for heartbeat miss:

A missed :pg heartbeat means pub-sub delivery is broken on this node. This directly affects users if the application uses :pg for real-time features. Configure high-priority PagerDuty for this monitor.

Group membership alert:

Configure a second HTTP monitor checking for adequate group membership:

Alert: body contains '"total":0' for a critical group
Meaning: no subscribers in the group — broadcasts are being dropped silently
Action: check that subscriber processes are starting and joining the group on startup

Common :pg Issues and Fixes

Group membership not visible from newly joined node:

# :pg syncs membership lazily after node connection
# Add a brief delay or retry in startup code that depends on remote members:
defp await_members(scope, group, min_count, timeout_ms) do
  deadline = System.monotonic_time(:millisecond) + timeout_ms
  await_members_loop(scope, group, min_count, deadline)
end

defp await_members_loop(scope, group, min_count, deadline) do
  if System.monotonic_time(:millisecond) > deadline do
    {:error, :timeout}
  else
    case length(:pg.get_members(scope, group)) >= min_count do
      true -> :ok
      false ->
        Process.sleep(100)
        await_members_loop(scope, group, min_count, deadline)
    end
  end
end

Messages silently dropped to dead PIDs:

# Add process monitoring for critical group members
defmodule MyApp.SafeBroadcast do
  def broadcast(scope, group, message) do
    scope
    |> :pg.get_members(group)
    |> Enum.map(fn pid ->
      if Process.alive?(pid) do
        send(pid, message)
        :ok
      else
        :dead
      end
    end)
    |> Enum.frequencies()
  end
end

Scope process crash causing membership loss:

# :pg auto-restores membership when the scope restarts — processes that are still alive
# will re-join if they monitor the scope and re-join on scope restart:
def init(state) do
  join_group()
  # Monitor the scope process
  scope_pid = Process.whereis(:my_app_pubsub)
  if scope_pid, do: Process.monitor(scope_pid)
  {:ok, state}
end

def handle_info({:DOWN, _ref, :process, _pid, _reason}, state) do
  # Scope restarted — rejoin after brief delay
  Process.send_after(self(), :rejoin_group, 500)
  {:noreply, state}
end

def handle_info(:rejoin_group, state) do
  join_group()
  {:noreply, state}
end

High group churn causing membership lag:

# For high-throughput scenarios, use process pools instead of individual join/leave:
# One long-lived pool member per node, fanning out to local worker processes
# This reduces :pg traffic from N join/leave/sec to nearly zero

What You've Built

| What | How | |------|-----| | Scope liveness monitoring | Check Process.whereis/1 on each poll | | Group membership polling | get_members/2 and get_local_members/2 with telemetry | | Local join/lookup probe | Probe process joins group and verifies visibility | | Cross-node visibility check | Compare local vs total member counts per group | | :pg health endpoint | /health with scope, probe, and critical group checks | | Liveness heartbeat | GenServer pinging Vigilmon only on successful probe | | HTTP and heartbeat monitors | Vigilmon checks endpoint + heartbeat liveness per node |

:pg gives your Erlang/Elixir cluster a built-in, zero-dependency pub-sub mechanism. Vigilmon makes sure messages are actually reaching all subscribers when your application depends on distributed event delivery.

Next Steps

  • Export :pg telemetry to Grafana for group membership history and join/leave rate dashboards
  • Add per-group heartbeat monitors for your highest-stakes groups (e.g., payment events, alert broadcasts)
  • Use Vigilmon incident history to correlate :pg membership drops with cluster events or network incidents
  • Consider subscribing to :pg membership change notifications via notify_join/notify_leave in OTP 25+ for event-driven membership tracking instead of polling

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