How to Monitor Circuits.GPIO with Vigilmon
Circuits.GPIO is the Elixir library for controlling GPIO (General Purpose Input/Output) pins on embedded Linux systems and Nerves devices. It provides a clean API for reading and writing digital pin states, setting pin directions, configuring pull-up/pull-down resistors, and attaching interrupt handlers that fire when a pin changes state. Whether you're toggling an LED, reading a door contact sensor, debouncing a button, or sampling a digital output from a hardware sensor, Circuits.GPIO is the interface between your Elixir application and the physical world.
Hardware is unreliable in ways software is not. A GPIO pin that worked yesterday may stop responding because of a loose connector, a power-supply glitch, a kernel driver reload, or a hardware fault on the GPIO controller. Interrupt handlers that fire too frequently signal electrical noise or a sensor in a failure mode. Pins that stop firing interrupts may indicate a stuck sensor or a silent crash in the GPIO subsystem. Without monitoring, these hardware events are invisible until a physical system stops working.
Vigilmon gives your embedded application observability: track pin state changes, interrupt rates, and read latency, and alert when hardware behavior deviates from expected patterns.
Why Monitor GPIO in Embedded Systems?
GPIO pins bridge software and the physical world — failures at this boundary are uniquely hard to detect:
- Stuck pins — a pin that should change state on sensor events stops changing; without interrupt monitoring you don't know the sensor is stuck until a physical inspection
- Interrupt flood — electrical noise on an input pin triggers hundreds of interrupts per second, consuming CPU and hiding the real signal
- Driver-level failures — the Linux GPIO subsystem or a specific GPIO controller driver can crash or become unresponsive; application-level reads return stale values without error
- Power supply instability — voltage fluctuations cause digital pins to read intermediate values, creating unexpected signal transitions that can trigger incorrect interrupt events
- Debounce failures — mechanical switches emit multiple transitions per physical press; missing or misconfigured debounce logic produces spurious events that can cause incorrect application behavior
- Sensor dropout — a digital sensor that outputs high-frequency pulses stops pulsing; pulse-counting code then silently reports zero when it should be reporting an error
Key Metrics to Track
| Metric | What it tells you | |--------|-------------------| | Interrupt count by pin per time window | Whether interrupt rates are normal or indicate noise/stuck pins | | Pin state read latency | Whether GPIO subsystem is responsive | | State transition count by pin | How often each pin changes state — baseline vs. anomaly | | Error count by operation (read/write/interrupt) | GPIO subsystem health | | Time since last interrupt by pin | Detects silent sensor dropout or stuck-state conditions | | GPIO handle open/close events | Whether GPIO resources are being allocated and released correctly |
Step 1: Add Circuits.GPIO to Your Project
# mix.exs
defp deps do
[
{:circuits_gpio, "~> 1.0"}
]
end
Basic GPIO usage:
# lib/my_app/gpio_controller.ex
defmodule MyApp.GpioController do
@doc "Read the current state of a digital input pin."
def read_pin(pin_number) do
with {:ok, gpio} <- Circuits.GPIO.open(pin_number, :input),
value <- Circuits.GPIO.read(gpio) do
Circuits.GPIO.close(gpio)
{:ok, value}
end
end
@doc "Write a digital value to an output pin."
def write_pin(pin_number, value) when value in [0, 1] do
with {:ok, gpio} <- Circuits.GPIO.open(pin_number, :output),
:ok <- Circuits.GPIO.write(gpio, value) do
Circuits.GPIO.close(gpio)
:ok
end
end
end
Long-lived GPIO with interrupt handling (preferred for production):
# lib/my_app/gpio_monitor.ex
defmodule MyApp.GpioMonitor do
use GenServer
require Logger
@pins %{
door_sensor: {4, :input, :both},
motion_sensor: {17, :input, :rising},
status_led: {27, :output, nil},
alarm_relay: {22, :output, nil}
}
def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
def init(_) do
gpio_handles =
Map.new(@pins, fn {name, {pin, direction, interrupt}} ->
{:ok, gpio} = Circuits.GPIO.open(pin, direction)
if interrupt do
:ok = Circuits.GPIO.set_interrupts(gpio, interrupt)
end
{name, %{gpio: gpio, pin: pin, last_value: nil, last_change: nil}}
end)
{:ok, %{gpios: gpio_handles}}
end
def handle_info({:circuits_gpio, pin, _timestamp, value}, state) do
# Find the pin name by pin number
{name, _} = Enum.find(state.gpios, fn {_, %{pin: p}} -> p == pin end)
Logger.info("GPIO interrupt", pin: name, pin_number: pin, value: value)
updated =
Map.update!(state.gpios, name, fn gpio_state ->
%{gpio_state | last_value: value, last_change: System.system_time(:second)}
end)
{:noreply, %{state | gpios: updated}}
end
def terminate(_reason, state) do
Enum.each(state.gpios, fn {_, %{gpio: gpio}} ->
Circuits.GPIO.close(gpio)
end)
end
end
Step 2: Instrument GPIO Operations with Telemetry
# lib/my_app/gpio_telemetry.ex
defmodule MyApp.GpioTelemetry do
@moduledoc """
Telemetry instrumentation for Circuits.GPIO operations.
Wrap GPIO reads, writes, and interrupt events to emit metrics.
"""
require Logger
@doc "Instrument a GPIO pin read."
def measured_read(gpio, pin_name) do
start = System.monotonic_time()
value = Circuits.GPIO.read(gpio)
duration = System.monotonic_time() - start
:telemetry.execute(
[:my_app, :gpio, :read],
%{duration: duration, count: 1},
%{pin: pin_name, value: value}
)
value
rescue
e ->
:telemetry.execute(
[:my_app, :gpio, :error],
%{count: 1},
%{pin: pin_name, operation: :read, error: e.__struct__}
)
Logger.error("GPIO read failed", pin: pin_name, error: inspect(e))
reraise e, __STACKTRACE__
end
@doc "Instrument a GPIO pin write."
def measured_write(gpio, pin_name, value) do
start = System.monotonic_time()
result = Circuits.GPIO.write(gpio, value)
duration = System.monotonic_time() - start
:telemetry.execute(
[:my_app, :gpio, :write],
%{duration: duration, count: 1},
%{pin: pin_name, value: value, success: result == :ok}
)
result
rescue
e ->
:telemetry.execute(
[:my_app, :gpio, :error],
%{count: 1},
%{pin: pin_name, operation: :write, error: e.__struct__}
)
Logger.error("GPIO write failed", pin: pin_name, value: value, error: inspect(e))
reraise e, __STACKTRACE__
end
@doc "Record a GPIO interrupt event."
def record_interrupt(pin_name, value) do
:telemetry.execute(
[:my_app, :gpio, :interrupt],
%{count: 1},
%{pin: pin_name, value: value}
)
end
@doc "Record a GPIO open/close lifecycle event."
def record_lifecycle(pin_name, pin_number, event) when event in [:open, :close] do
:telemetry.execute(
[:my_app, :gpio, :lifecycle],
%{count: 1},
%{pin: pin_name, pin_number: pin_number, event: event}
)
end
end
Update your GPIO monitor to use instrumented calls:
# lib/my_app/instrumented_gpio_monitor.ex
defmodule MyApp.InstrumentedGpioMonitor do
use GenServer
require Logger
@pins %{
door_sensor: {4, :input, :both},
motion_sensor: {17, :input, :rising},
status_led: {27, :output, nil},
alarm_relay: {22, :output, nil}
}
def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
def init(_) do
gpio_handles =
Map.new(@pins, fn {name, {pin, direction, interrupt}} ->
{:ok, gpio} = Circuits.GPIO.open(pin, direction)
MyApp.GpioTelemetry.record_lifecycle(name, pin, :open)
if interrupt do
:ok = Circuits.GPIO.set_interrupts(gpio, interrupt)
end
initial_value = if direction == :input, do: Circuits.GPIO.read(gpio), else: nil
{name, %{
gpio: gpio,
pin: pin,
direction: direction,
last_value: initial_value,
last_interrupt: nil,
interrupt_count: 0
}}
end)
{:ok, %{gpios: gpio_handles}}
end
def handle_info({:circuits_gpio, pin, timestamp, value}, state) do
{name, pin_state} = Enum.find(state.gpios, fn {_, %{pin: p}} -> p == pin end)
MyApp.GpioTelemetry.record_interrupt(name, value)
updated_pin = %{pin_state |
last_value: value,
last_interrupt: timestamp,
interrupt_count: pin_state.interrupt_count + 1
}
{:noreply, put_in(state, [:gpios, name], updated_pin)}
end
def handle_call({:read, name}, _from, state) do
pin_state = state.gpios[name]
value = MyApp.GpioTelemetry.measured_read(pin_state.gpio, name)
{:reply, {:ok, value}, state}
end
def handle_call({:write, name, value}, _from, state) do
pin_state = state.gpios[name]
result = MyApp.GpioTelemetry.measured_write(pin_state.gpio, name, value)
{:reply, result, state}
end
def terminate(_reason, state) do
Enum.each(state.gpios, fn {name, %{gpio: gpio, pin: pin}} ->
MyApp.GpioTelemetry.record_lifecycle(name, pin, :close)
Circuits.GPIO.close(gpio)
end)
end
end
Step 3: Detect Interrupt Flood and Sensor Dropout
# lib/my_app/gpio_anomaly_detector.ex
defmodule MyApp.GpioAnomalyDetector do
use GenServer
require Logger
@interrupt_flood_threshold 50
@dropout_timeout_seconds 300
@window_seconds 60
@poll_interval :timer.seconds(30)
@monitored_pins [:door_sensor, :motion_sensor]
@expected_interrupt_pins [:motion_sensor]
def start_link(_),
do: GenServer.start_link(__MODULE__, %{interrupts: %{}, last_seen: %{}}, name: __MODULE__)
def init(state) do
:telemetry.attach(
"gpio-anomaly-detector",
[:my_app, :gpio, :interrupt],
&handle_interrupt_event/4,
nil
)
schedule()
{:ok, state}
end
def handle_interrupt_event(_event, %{count: 1}, %{pin: pin}, _) do
GenServer.cast(__MODULE__, {:record_interrupt, pin, System.system_time(:second)})
end
def handle_cast({:record_interrupt, pin, ts}, state) do
interrupts = Map.update(state.interrupts, {pin, ts}, 1, &(&1 + 1))
last_seen = Map.put(state.last_seen, pin, ts)
{:noreply, %{state | interrupts: interrupts, last_seen: last_seen}}
end
def handle_info(:check, state) do
now = System.system_time(:second)
cutoff = now - @window_seconds
# Detect interrupt flood
recent_by_pin =
state.interrupts
|> Enum.filter(fn {{_, ts}, _} -> ts > cutoff end)
|> Enum.group_by(fn {{pin, _}, _} -> pin end, fn {_, count} -> count end)
|> Map.new(fn {pin, counts} -> {pin, Enum.sum(counts)} end)
Enum.each(recent_by_pin, fn {pin, count} ->
:telemetry.execute(
[:my_app, :gpio, :interrupt_rate],
%{count: count},
%{pin: pin, window_seconds: @window_seconds}
)
if count > @interrupt_flood_threshold do
Logger.error("GPIO interrupt flood detected",
pin: pin,
count: count,
window_seconds: @window_seconds
)
end
end)
# Detect sensor dropout
Enum.each(@expected_interrupt_pins, fn pin ->
last = Map.get(state.last_seen, pin)
silent_seconds = if last, do: now - last, else: nil
cond do
is_nil(last) ->
Logger.warning("GPIO pin has never fired an interrupt — possible sensor not connected",
pin: pin
)
silent_seconds > @dropout_timeout_seconds ->
Logger.error("GPIO sensor dropout detected",
pin: pin,
silent_seconds: silent_seconds,
threshold_seconds: @dropout_timeout_seconds
)
:telemetry.execute(
[:my_app, :gpio, :sensor_dropout],
%{silent_seconds: silent_seconds},
%{pin: pin}
)
true ->
:ok
end
end)
# Prune old interrupt records
pruned = Enum.filter(state.interrupts, fn {{_, ts}, _} -> ts > cutoff end) |> Map.new()
schedule()
{:noreply, %{state | interrupts: pruned}}
end
defp schedule, do: Process.send_after(self(), :check, @poll_interval)
end
Step 4: Define Telemetry Metrics
# lib/my_app/telemetry.ex
def metrics do
[
# GPIO read operations
counter("my_app.gpio.read.count",
tags: [:pin, :value]
),
distribution("my_app.gpio.read.duration",
unit: {:native, :microsecond},
reporter_options: [buckets: [10, 50, 200, 1000, 5000]],
tags: [:pin]
),
# GPIO write operations
counter("my_app.gpio.write.count",
tags: [:pin, :value, :success]
),
distribution("my_app.gpio.write.duration",
unit: {:native, :microsecond},
reporter_options: [buckets: [10, 50, 200, 1000, 5000]],
tags: [:pin]
),
# Interrupt events
counter("my_app.gpio.interrupt.count",
tags: [:pin, :value]
),
# Error events
counter("my_app.gpio.error.count",
tags: [:pin, :operation, :error]
),
# Anomaly detection
last_value("my_app.gpio.interrupt_rate.count",
tags: [:pin, :window_seconds]
),
last_value("my_app.gpio.sensor_dropout.silent_seconds",
tags: [:pin]
),
# Lifecycle
counter("my_app.gpio.lifecycle.count",
tags: [:pin, :pin_number, :event]
)
]
end
Step 5: Self-Test Health Check
Verify GPIO infrastructure is working with a round-trip write/read test on an output pin:
# lib/my_app/gpio_health.ex
defmodule MyApp.GpioHealth do
@moduledoc """
GPIO infrastructure health check.
Performs a write/read round-trip on a designated test pin to verify
that the GPIO subsystem is functional.
"""
@test_pin Application.compile_env(:my_app, :gpio_health_test_pin, 27)
def check do
start = System.monotonic_time(:millisecond)
with {:ok, gpio} <- Circuits.GPIO.open(@test_pin, :output),
:ok <- Circuits.GPIO.write(gpio, 1),
:ok <- Circuits.GPIO.write(gpio, 0),
:ok <- Circuits.GPIO.close(gpio) do
latency_ms = System.monotonic_time(:millisecond) - start
{:ok, %{gpio_subsystem: :ok, latency_ms: latency_ms}}
else
{:error, reason} ->
{:error, %{gpio_subsystem: :error, reason: inspect(reason)}}
end
rescue
e -> {:error, %{gpio_subsystem: :error, reason: Exception.message(e)}}
end
end
Step 6: Health Endpoint
# lib/my_app_web/controllers/health_controller.ex
defmodule MyAppWeb.HealthController do
use MyAppWeb, :controller
def index(conn, _params) do
gpio_check = MyApp.GpioHealth.check()
checks = %{
gpio: format_gpio_check(gpio_check),
database: check_db()
}
status = if match?({:ok, _}, gpio_check), do: 200, else: 503
conn
|> put_status(status)
|> json(%{
status: if(status == 200, do: "ok", else: "degraded"),
checks: checks
})
end
defp format_gpio_check({:ok, details}), do: Map.put(details, :status, "ok")
defp format_gpio_check({:error, details}), do: Map.put(details, :status, "error")
defp check_db do
case Ecto.Adapters.SQL.query(MyApp.Repo, "SELECT 1", []) do
{:ok, _} -> %{status: "ok"}
_ -> %{status: "error"}
end
rescue
_ -> %{status: "error"}
end
end
Step 7: Create Monitors in Vigilmon
HTTP monitor for the health endpoint:
- Sign in at vigilmon.online
- Click New Monitor → HTTP
- Set URL to
https://your-device.local/health(or the device's external IP/hostname) - Set interval to 60 seconds
- Alert condition: status
200and body contains"gpio":{"status":"ok"}
Heartbeat monitor for GPIO subsystem liveness:
# lib/my_app/gpio_heartbeat.ex
defmodule MyApp.GpioHeartbeat do
use GenServer
@heartbeat_url System.get_env("VIGILMON_GPIO_HEARTBEAT_URL")
@interval :timer.minutes(5)
def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
def init(state) do
schedule()
{:ok, state}
end
def handle_info(:ping, state) do
case MyApp.GpioHealth.check() do
{:ok, _details} ->
ping_vigilmon()
{:error, details} ->
require Logger
Logger.error("GPIO health check failed — not pinging Vigilmon", details: details)
end
schedule()
{:noreply, state}
end
defp ping_vigilmon do
if @heartbeat_url do
:httpc.request(:get, {String.to_charlist(@heartbeat_url), []}, [{:timeout, 5000}], [])
end
end
defp schedule, do: Process.send_after(self(), :ping, @interval)
end
Add to your application supervision tree:
# lib/my_app/application.ex
children = [
MyApp.Repo,
MyApp.InstrumentedGpioMonitor,
MyApp.GpioAnomalyDetector,
MyApp.GpioHeartbeat,
MyAppWeb.Endpoint
]
Create a Heartbeat monitor in Vigilmon with an 8-minute expected interval. If the GPIO subsystem crashes or the device loses network connectivity, the heartbeat stops and Vigilmon alerts.
Step 8: Alerting
In Vigilmon, configure Notifications → New Channel:
Slack for GPIO health failure:
🚨 CRITICAL: GPIO Subsystem Failure — MyDevice
Device: production-unit-03
Error: GPIO read failed — /dev/gpiochip0: no such device
Action: Check device power, GPIO controller, and kernel driver status
PagerDuty for missed heartbeat (device offline or GPIO crash):
Page on-call immediately if the heartbeat monitor misses two consecutive expected pings. A missed heartbeat from an embedded device usually means the device is offline, crashed, or has lost network access — all critical for physical systems.
PagerDuty for sensor dropout:
- alert: GpioSensorDropout
expr: my_app_gpio_sensor_dropout_silent_seconds{pin="motion_sensor"} > 300
for: 1m
annotations:
summary: "Motion sensor on pin {{ $labels.pin }} has been silent for {{ $value }}s — possible sensor fault"
Grafana alert for interrupt flood:
- alert: GpioInterruptFlood
expr: my_app_gpio_interrupt_rate_count > 50
for: 1m
labels:
severity: warning
annotations:
summary: "GPIO interrupt flood on {{ $labels.pin }} — {{ $value }} interrupts/min. Possible electrical noise."
Common GPIO Issues and Fixes
Interrupt flood from electrical noise:
# Add software debounce in the interrupt handler
defmodule MyApp.DebouncedGpioMonitor do
use GenServer
@debounce_ms 50
def handle_info({:circuits_gpio, pin, _ts, value}, state) do
# Cancel any pending debounce timer for this pin
if timer = state.debounce_timers[pin] do
Process.cancel_timer(timer)
end
# Schedule the actual event processing after the debounce window
timer = Process.send_after(self(), {:debounced_event, pin, value}, @debounce_ms)
{:noreply, put_in(state, [:debounce_timers, pin], timer)}
end
def handle_info({:debounced_event, pin, value}, state) do
# This fires only if no new interrupt arrived within @debounce_ms
MyApp.GpioTelemetry.record_interrupt(pin_name(pin), value)
process_pin_change(pin, value, state)
end
end
GPIO handle leak (opening without closing):
# Always use with/rescue to ensure GPIO handles are closed
def one_shot_read(pin_number, pin_name) do
gpio = nil
{:ok, gpio} = Circuits.GPIO.open(pin_number, :input)
value = MyApp.GpioTelemetry.measured_read(gpio, pin_name)
Circuits.GPIO.close(gpio)
{:ok, value}
rescue
e ->
if gpio, do: Circuits.GPIO.close(gpio)
{:error, Exception.message(e)}
end
GPIO subsystem unresponsive after kernel update:
# Add a watchdog that detects stale read latency and alerts
defmodule MyApp.GpioWatchdog do
use GenServer
require Logger
@read_timeout_ms 1000
@check_interval :timer.minutes(1)
def handle_info(:check, state) do
task = Task.async(fn -> MyApp.GpioHealth.check() end)
case Task.yield(task, @read_timeout_ms) || Task.shutdown(task) do
{:ok, {:ok, _}} ->
:ok
{:ok, {:error, reason}} ->
Logger.error("GPIO health check returned error", reason: inspect(reason))
nil ->
Logger.error("GPIO health check timed out after #{@read_timeout_ms}ms — possible kernel driver hang")
end
Process.send_after(self(), :check, @check_interval)
{:noreply, state}
end
end
Pin stuck high/low after power cycle:
# On startup, read and log all input pin states to detect pins
# that initialized in an unexpected state
defmodule MyApp.GpioStartupAudit do
require Logger
@expected_initial_states %{
door_sensor: 0,
motion_sensor: 0
}
def audit(gpio_handles) do
Enum.each(@expected_initial_states, fn {name, expected} ->
case gpio_handles[name] do
%{gpio: gpio} ->
actual = Circuits.GPIO.read(gpio)
if actual != expected do
Logger.warning("GPIO pin initialized in unexpected state",
pin: name,
expected: expected,
actual: actual
)
end
nil ->
Logger.error("Expected GPIO pin not initialized", pin: name)
end
end)
end
end
What You've Built
| What | How | |------|-----| | Instrumented GPIO operations | Telemetry on every read, write, interrupt, and lifecycle event | | Interrupt flood detection | Rolling-window counter with threshold alerting | | Sensor dropout detection | Time-since-last-interrupt tracking per expected-interrupt pin | | GPIO round-trip health check | Write/read self-test verifying GPIO subsystem responsiveness | | Structured health endpoint | JSON health response reporting GPIO subsystem status | | Liveness heartbeat | GenServer pinging Vigilmon after successful GPIO health check | | Real-time alerting | Vigilmon HTTP + heartbeat monitors with PagerDuty on-call routing |
Circuits.GPIO connects your Elixir application to the physical world. Vigilmon ensures that connection stays healthy — alerting you the moment a sensor drops out, a pin floods with noise, or the GPIO subsystem becomes unresponsive, before a physical system failure reaches your users or operators.
Next Steps
- Combine Circuits.GPIO with Circuits.I2C or Circuits.SPI for sensor buses and instrument them with the same telemetry patterns
- Use Nerves.Leds for status LED feedback alongside Vigilmon heartbeat monitoring for devices without network connectivity
- Add device-level dashboards in Grafana showing all GPIO pins by state over time for hardware debugging
- Set up Vigilmon public status pages for industrial or commercial deployments where hardware uptime is customer-visible
Get started free at vigilmon.online.