makoto-developer's テックブログ

Elixir OTPで実装する自動復旧アーキテクチャ - Supervisor、GenServer、リトライ戦略の実践

#Elixir #OTP #アーキテクチャ #フォールトトレランス

Elixir OTPの「OTP」とは「Open Telecom Platform」の略です。

OTPに付属するビヘイビア(他の言語でいうインターフェースと訳して大丈夫)を使って、自動復旧アーキテクチャの実装方法について知っていることを考えながら書き下します。
(見直しせず投稿するつもりなので矛盾する箇所があるかもしれないですがスルーしてください)

なるべく、プロセスの監視、自動再起動、リトライ戦略を実践的なコード例を多めに記載する。

なぜ自動復旧が重要なのか

従来の問題

多くのプログラミング言語では、エラーが発生した際の復旧処理を手動で実装する必要がある。

Javaの例:

// 手動でのリトライ実装
public Response callAPI() {
    int maxRetries = 3;
    for (int i = 0; i < maxRetries; i++) {
        try {
            return httpClient.get("https://api.example.com/data");
        } catch (Exception e) {
            if (i == maxRetries - 1) {
                throw e;
            }
            Thread.sleep(1000 * i);
        }
    }
}

問題点:

  • リトライロジックがビジネスロジックに混在
  • エラーハンドリングのボイラープレート
  • プロセス全体のクラッシュを防ぐのが困難
  • 状態の復旧が複雑

一番はコードが煩雑になる。リトライ処理自体がコードを読みにくくするのが問題だと思う。

外野の声 AIがあるから気にすんな
私 そんな野暮なことは言わないで、コードにこだわらなければプログラマじゃない(と個人的に思う)

OTPの解決策

Elixir OTPでは、以下の機能が標準で提供される：

  • Supervisor: プロセスの監視と自動再起動
  • GenServer: 状態を持つプロセスの実装
  • 再起動戦略: 障害の伝播制御
  • リンクとモニター: プロセス間の依存関係管理

利点:

  • ビジネスロジックとエラーハンドリングの分離
  • 宣言的なフォールトトレランス
  • 自動的な状態の初期化
  • スケーラブルなアーキテクチャ

OTPの哲学："Let it crash"

従来のアプローチ: Defensive Programming

# Python: 防御的プログラミング
def process_data(data):
    if data is None:
        return default_value
    if not isinstance(data, dict):
        return default_value
    if 'key' not in data:
        return default_value
    
    try:
        result = expensive_operation(data['key'])
        if result is None:
            return default_value
        return result
    except Exception as e:
        log_error(e)
        return default_value

問題点:

  • コードが冗長
  • エラー状態が隠蔽される
  • デバッグが困難

OTPのアプローチ: Let it crash

# Elixir: クラッシュさせて再起動
defmodule DataProcessor do
  use GenServer

  def process_data(data) do
    # バリデーションなし - 不正なデータはクラッシュ
    result = expensive_operation(data["key"])
    result
  end
end

# Supervisorが自動的に再起動
defmodule MyApp.Supervisor do
  use Supervisor

  def start_link(init_arg) do
    Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  def init(_init_arg) do
    children = [
      {DataProcessor, []}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end
end

利点:

  • コードがシンプル
  • エラーが明確
  • 自動的にクリーンな状態に復旧

OTPの基礎概念

プロセスモデル

Elixirでは、すべてが軽量プロセスとして実行される。

Application
  └── Supervisor (監視)
        ├── GenServer A (状態を持つワーカー)
        ├── GenServer B (状態を持つワーカー)
        └── Task.Supervisor (一時的タスクの監視)
              ├── Task 1
              └── Task 2

GenServer（汎用サーバー）

状態を持つプロセスの実装パターン。

defmodule Counter do
  use GenServer

  # クライアントAPI

  def start_link(initial_value) do
    GenServer.start_link(__MODULE__, initial_value, name: __MODULE__)
  end

  def increment do
    GenServer.call(__MODULE__, :increment)
  end

  def get do
    GenServer.call(__MODULE__, :get)
  end

  # サーバーコールバック

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

  @impl true
  def handle_call(:increment, _from, state) do
    new_state = state + 1
    {:reply, new_state, new_state}
  end

  @impl true
  def handle_call(:get, _from, state) do
    {:reply, state, state}
  end
end

Supervisor（監視者）

子プロセスを監視し、クラッシュ時に再起動する。

defmodule MyApp.Supervisor do
  use Supervisor

  def start_link(init_arg) do
    Supervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  @impl true
  def init(_init_arg) do
    children = [
      {Counter, 0}
    ]

    # 再起動戦略
    Supervisor.init(children, strategy: :one_for_one)
  end
end

再起動戦略

戦略 説明 用途
:one_for_one クラッシュしたプロセスのみ再起動 独立したワーカー
:one_for_all 1つがクラッシュすると全て再起動 相互依存するプロセス
:rest_for_one クラッシュ以降のプロセスを再起動 順序依存のプロセス 
# :one_for_one
# Worker A がクラッシュ → Worker A のみ再起動
# Worker B, C は影響なし

# :one_for_all
# Worker A がクラッシュ → Worker A, B, C すべて再起動

# :rest_for_one
# Worker B がクラッシュ → Worker B, C を再起動（A は影響なし）

基本的な自動復旧の実装

プロジェクトのセットアップ

# 新規プロジェクト作成
mix new auto_recovery --sup
cd auto_recovery

# 依存関係の追加

クラッシュするGenServerの実装

defmodule AutoRecovery.CrashableWorker do
  use GenServer
  require Logger

  # クライアントAPI

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def do_work do
    GenServer.call(__MODULE__, :do_work)
  end

  def crash do
    GenServer.call(__MODULE__, :crash)
  end

  def get_state do
    GenServer.call(__MODULE__, :get_state)
  end

  # サーバーコールバック

  @impl true
  def init(_opts) do
    Logger.info("Worker started with PID #{inspect(self())}")
    {:ok, %{work_count: 0, start_time: DateTime.utc_now()}}
  end

  @impl true
  def handle_call(:do_work, _from, state) do
    new_count = state.work_count + 1
    Logger.info("Doing work... count: #{new_count}")
    {:reply, :ok, %{state | work_count: new_count}}
  end

  @impl true
  def handle_call(:crash, _from, _state) do
    Logger.warning("Intentional crash!")
    # 意図的にクラッシュ
    raise "Boom!"
  end

  @impl true
  def handle_call(:get_state, _from, state) do
    {:reply, state, state}
  end

  @impl true
  def terminate(reason, state) do
    Logger.error("Worker terminating. Reason: #{inspect(reason)}, State: #{inspect(state)}")
    :ok
  end
end

Supervisorの実装

defmodule AutoRecovery.Application do
  use Application
  require Logger

  @impl true
  def start(_type, _args) do
    Logger.info("Starting AutoRecovery Application")

    children = [
      # クラッシュ可能なワーカー
      {AutoRecovery.CrashableWorker, []}
    ]

    opts = [strategy: :one_for_one, name: AutoRecovery.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

動作確認

# アプリケーションを起動
iex -S mix

# ワーカーの動作確認
iex> AutoRecovery.CrashableWorker.do_work()
:ok

iex> AutoRecovery.CrashableWorker.get_state()
%{work_count: 1, start_time: ~U[2026-02-11 10:00:00Z]}

# 意図的にクラッシュ
iex> AutoRecovery.CrashableWorker.crash()
# ** (EXIT) Boom!

# 自動的に再起動される（新しいPIDで起動）
iex> AutoRecovery.CrashableWorker.get_state()
%{work_count: 0, start_time: ~U[2026-02-11 10:00:05Z]}
# 状態がリセットされている

重要: クラッシュ後、Supervisorが自動的に新しいプロセスを起動する。状態は初期化される。

リトライ戦略の実装

1. 単純なリトライ

defmodule AutoRecovery.Retry do
  require Logger

  @doc """
  単純なリトライ（固定間隔）
  """
  def retry(func, max_attempts \\ 3, delay_ms \\ 1000) do
    retry_loop(func, max_attempts, delay_ms, 1)
  end

  defp retry_loop(func, max_attempts, delay_ms, attempt) do
    case func.() do
      {:ok, result} ->
        {:ok, result}

      {:error, reason} when attempt < max_attempts ->
        Logger.warning("Attempt #{attempt} failed: #{inspect(reason)}. Retrying in #{delay_ms}ms...")
        Process.sleep(delay_ms)
        retry_loop(func, max_attempts, delay_ms, attempt + 1)

      {:error, reason} ->
        Logger.error("All #{max_attempts} attempts failed. Last error: #{inspect(reason)}")
        {:error, :max_retries_exceeded}
    end
  end
end

2. 指数バックオフ

defmodule AutoRecovery.ExponentialBackoff do
  require Logger

  @doc """
  指数バックオフによるリトライ
  """
  def retry(func, opts \\ []) do
    max_attempts = Keyword.get(opts, :max_attempts, 5)
    base_delay_ms = Keyword.get(opts, :base_delay_ms, 1000)
    max_delay_ms = Keyword.get(opts, :max_delay_ms, 30_000)

    retry_loop(func, max_attempts, base_delay_ms, max_delay_ms, 1)
  end

  defp retry_loop(func, max_attempts, base_delay_ms, max_delay_ms, attempt) do
    case func.() do
      {:ok, result} ->
        {:ok, result}

      {:error, reason} when attempt < max_attempts ->
        delay_ms = calculate_delay(attempt, base_delay_ms, max_delay_ms)
        Logger.warning(
          "Attempt #{attempt}/#{max_attempts} failed: #{inspect(reason)}. " <>
          "Retrying in #{delay_ms}ms..."
        )
        Process.sleep(delay_ms)
        retry_loop(func, max_attempts, base_delay_ms, max_delay_ms, attempt + 1)

      {:error, reason} ->
        Logger.error("All #{max_attempts} attempts failed. Last error: #{inspect(reason)}")
        {:error, :max_retries_exceeded}
    end
  end

  defp calculate_delay(attempt, base_delay_ms, max_delay_ms) do
    # 指数バックオフ: base * 2^(attempt - 1)
    delay = base_delay_ms * :math.pow(2, attempt - 1)
    min(trunc(delay), max_delay_ms)
  end
end

3. サーキットブレーカー

defmodule AutoRecovery.CircuitBreaker do
  use GenServer
  require Logger

  @moduledoc """
  サーキットブレーカーパターンの実装

  状態:
  - :closed (正常) - リクエストを通す
  - :open (異常) - リクエストを遮断
  - :half_open (回復中) - 一部のリクエストを通してテスト
  """

  defmodule State do
    defstruct [
      :state,           # :closed | :open | :half_open
      :failure_count,   # 失敗回数
      :failure_threshold, # 失敗しきい値
      :timeout_ms,      # オープン状態の持続時間
      :last_failure_time # 最後の失敗時刻
    ]
  end

  # クライアントAPI

  def start_link(opts) do
    name = Keyword.get(opts, :name, __MODULE__)
    GenServer.start_link(__MODULE__, opts, name: name)
  end

  def call(circuit_breaker, func) do
    GenServer.call(circuit_breaker, {:call, func})
  end

  def status(circuit_breaker) do
    GenServer.call(circuit_breaker, :status)
  end

  # サーバーコールバック

  @impl true
  def init(opts) do
    state = %State{
      state: :closed,
      failure_count: 0,
      failure_threshold: Keyword.get(opts, :failure_threshold, 5),
      timeout_ms: Keyword.get(opts, :timeout_ms, 60_000),
      last_failure_time: nil
    }

    {:ok, state}
  end

  @impl true
  def handle_call({:call, func}, _from, state) do
    case state.state do
      :closed ->
        handle_closed(func, state)

      :open ->
        handle_open(func, state)

      :half_open ->
        handle_half_open(func, state)
    end
  end

  @impl true
  def handle_call(:status, _from, state) do
    {:reply, state.state, state}
  end

  # プライベート関数

  defp handle_closed(func, state) do
    case execute_function(func) do
      {:ok, result} ->
        # 成功: 失敗カウントをリセット
        new_state = %{state | failure_count: 0}
        {:reply, {:ok, result}, new_state}

      {:error, reason} ->
        # 失敗: カウントを増やす
        new_failure_count = state.failure_count + 1

        if new_failure_count >= state.failure_threshold do
          Logger.warning("Circuit breaker opened after #{new_failure_count} failures")
          new_state = %{state |
            state: :open,
            failure_count: new_failure_count,
            last_failure_time: System.monotonic_time(:millisecond)
          }
          {:reply, {:error, reason}, new_state}
        else
          new_state = %{state | failure_count: new_failure_count}
          {:reply, {:error, reason}, new_state}
        end
    end
  end

  defp handle_open(func, state) do
    current_time = System.monotonic_time(:millisecond)
    elapsed = current_time - state.last_failure_time

    if elapsed >= state.timeout_ms do
      # タイムアウト経過: half_openに移行
      Logger.info("Circuit breaker transitioning to half_open")
      new_state = %{state | state: :half_open}
      handle_half_open(func, new_state)
    else
      # まだタイムアウト前: リクエストを遮断
      {:reply, {:error, :circuit_open}, state}
    end
  end

  defp handle_half_open(func, state) do
    case execute_function(func) do
      {:ok, result} ->
        # 成功: 回復してclosedに戻る
        Logger.info("Circuit breaker closed (recovered)")
        new_state = %{state | state: :closed, failure_count: 0}
        {:reply, {:ok, result}, new_state}

      {:error, reason} ->
        # 失敗: 再びopenに戻る
        Logger.warning("Circuit breaker reopened")
        new_state = %{state |
          state: :open,
          last_failure_time: System.monotonic_time(:millisecond)
        }
        {:reply, {:error, reason}, new_state}
    end
  end

  defp execute_function(func) do
    try do
      result = func.()
      {:ok, result}
    rescue
      e ->
        {:error, e}
    catch
      :exit, reason ->
        {:error, reason}
    end
  end
end

実践例1: 外部API呼び出しの自動復旧

HTTPクライアントのラッパー

defp deps do
  [
    {:httpoison, "~> 2.0"},
    {:jason, "~> 1.4"}
  ]
end
defmodule AutoRecovery.ApiClient do
  use GenServer
  require Logger
  alias AutoRecovery.{ExponentialBackoff, CircuitBreaker}

  defmodule State do
    defstruct [:base_url, :circuit_breaker]
  end

  # クライアントAPI

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def fetch_weather(city) do
    GenServer.call(__MODULE__, {:fetch_weather, city}, 30_000)
  end

  # サーバーコールバック

  @impl true
  def init(opts) do
    base_url = Keyword.get(opts, :base_url, "https://api.weather.example.com")

    # サーキットブレーカーを起動
    {:ok, circuit_breaker} = CircuitBreaker.start_link(
      name: :api_circuit_breaker,
      failure_threshold: 3,
      timeout_ms: 30_000
    )

    state = %State{
      base_url: base_url,
      circuit_breaker: circuit_breaker
    }

    {:ok, state}
  end

  @impl true
  def handle_call({:fetch_weather, city}, _from, state) do
    result = ExponentialBackoff.retry(fn ->
      fetch_with_circuit_breaker(city, state)
    end, max_attempts: 3, base_delay_ms: 1000)

    {:reply, result, state}
  end

  # プライベート関数

  defp fetch_with_circuit_breaker(city, state) do
    CircuitBreaker.call(state.circuit_breaker, fn ->
      do_fetch(city, state.base_url)
    end)
  end

  defp do_fetch(city, base_url) do
    url = "#{base_url}/weather?city=#{URI.encode(city)}"
    Logger.info("Fetching weather for #{city} from #{url}")

    case HTTPoison.get(url, [], recv_timeout: 5_000) do
      {:ok, %HTTPoison.Response{status_code: 200, body: body}} ->
        case Jason.decode(body) do
          {:ok, data} ->
            {:ok, data}

          {:error, reason} ->
            Logger.error("JSON decode error: #{inspect(reason)}")
            {:error, :invalid_json}
        end

      {:ok, %HTTPoison.Response{status_code: status_code}} ->
        Logger.error("HTTP error: #{status_code}")
        {:error, {:http_error, status_code}}

      {:error, %HTTPoison.Error{reason: reason}} ->
        Logger.error("HTTP request failed: #{inspect(reason)}")
        {:error, reason}
    end
  end
end

Supervisorへの追加

defmodule AutoRecovery.Application do
  use Application
  require Logger

  @impl true
  def start(_type, _args) do
    children = [
      {AutoRecovery.CrashableWorker, []},
      {AutoRecovery.ApiClient, base_url: "https://api.weather.example.com"}
    ]

    opts = [strategy: :one_for_one, name: AutoRecovery.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

使用例

# APIクライアントを使用
iex> AutoRecovery.ApiClient.fetch_weather("Tokyo")
{:ok, %{"city" => "Tokyo", "temp" => 15, "condition" => "Sunny"}}

# エラー時は自動的にリトライ
iex> AutoRecovery.ApiClient.fetch_weather("InvalidCity")
# ログ: Attempt 1/3 failed... Retrying in 1000ms...
# ログ: Attempt 2/3 failed... Retrying in 2000ms...
# ログ: All 3 attempts failed.
{:error, :max_retries_exceeded}

実践例2: データベース接続の自動復旧

Ectoの設定

defp deps do
  [
    {:ecto_sql, "~> 3.10"},
    {:postgrex, "~> 0.17"}
  ]
end
import Config

config :auto_recovery, AutoRecovery.Repo,
  database: "auto_recovery_dev",
  username: "postgres",
  password: "postgres",
  hostname: "localhost",
  pool_size: 10

Repoの定義

defmodule AutoRecovery.Repo do
  use Ecto.Repo,
    otp_app: :auto_recovery,
    adapter: Ecto.Adapters.Postgres
end

データベースワーカー

defmodule AutoRecovery.DbWorker do
  use GenServer
  require Logger
  alias AutoRecovery.Repo

  # クライアントAPI

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def execute_query(query) do
    GenServer.call(__MODULE__, {:execute_query, query}, 10_000)
  end

  # サーバーコールバック

  @impl true
  def init(_opts) do
    Logger.info("DbWorker started")
    {:ok, %{}}
  end

  @impl true
  def handle_call({:execute_query, query}, _from, state) do
    result = AutoRecovery.ExponentialBackoff.retry(fn ->
      execute_with_timeout(query)
    end, max_attempts: 3, base_delay_ms: 500)

    {:reply, result, state}
  end

  # プライベート関数

  defp execute_with_timeout(query) do
    try do
      result = Repo.query(query, [], timeout: 5_000)
      {:ok, result}
    rescue
      Ecto.QueryError -> {:error, :query_error}
      DBConnection.ConnectionError -> {:error, :connection_error}
    catch
      :exit, {:timeout, _} -> {:error, :timeout}
    end
  end
end

接続プールの監視

Ectoの接続プールは、標準でSupervisorによって監視されている。接続が切れた場合、自動的に再接続が試みられる。

defmodule AutoRecovery.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      # Ecto Repo（接続プール）
      AutoRecovery.Repo,

      # その他のワーカー
      {AutoRecovery.DbWorker, []}
    ]

    opts = [strategy: :one_for_one, name: AutoRecovery.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

高度なパターン

1. DynamicSupervisor

動的に子プロセスを追加・削除できるSupervisor。

defmodule AutoRecovery.JobProcessor do
  use GenServer
  require Logger

  def start_link(job_id) do
    GenServer.start_link(__MODULE__, job_id)
  end

  @impl true
  def init(job_id) do
    Logger.info("Processing job #{job_id}")
    Process.send_after(self(), :process, 1000)
    {:ok, %{job_id: job_id}}
  end

  @impl true
  def handle_info(:process, state) do
    Logger.info("Job #{state.job_id} completed")
    {:stop, :normal, state}
  end
end
defmodule AutoRecovery.JobSupervisor do
  use DynamicSupervisor

  def start_link(init_arg) do
    DynamicSupervisor.start_link(__MODULE__, init_arg, name: __MODULE__)
  end

  @impl true
  def init(_init_arg) do
    DynamicSupervisor.init(strategy: :one_for_one)
  end

  def start_job(job_id) do
    spec = {AutoRecovery.JobProcessor, job_id}
    DynamicSupervisor.start_child(__MODULE__, spec)
  end
end

2. Task.Supervisor

一時的なタスクの実行と監視。

defmodule AutoRecovery.AsyncWorker do
  require Logger

  def perform_async(task_name, func) do
    Task.Supervisor.async_nolink(AutoRecovery.TaskSupervisor, fn ->
      Logger.info("Starting task: #{task_name}")
      result = func.()
      Logger.info("Task #{task_name} completed")
      result
    end)
  end

  def await_result(task, timeout \\ 5000) do
    case Task.yield(task, timeout) || Task.shutdown(task) do
      {:ok, result} ->
        {:ok, result}

      nil ->
        Logger.error("Task timed out")
        {:error, :timeout}
    end
  end
end
children = [
  # Task.Supervisorを追加
  {Task.Supervisor, name: AutoRecovery.TaskSupervisor},
  # ...
]

3. Registry

プロセスの名前解決と動的な検索。

defmodule AutoRecovery.UserSession do
  use GenServer
  require Logger

  # クライアントAPI

  def start_link(user_id) do
    GenServer.start_link(__MODULE__, user_id, name: via_tuple(user_id))
  end

  def get_session(user_id) do
    case Registry.lookup(AutoRecovery.Registry, user_id) do
      [{pid, _}] ->
        GenServer.call(pid, :get_session)

      [] ->
        {:error, :not_found}
    end
  end

  defp via_tuple(user_id) do
    {:via, Registry, {AutoRecovery.Registry, user_id}}
  end

  # サーバーコールバック

  @impl true
  def init(user_id) do
    Logger.info("User session started for user #{user_id}")
    {:ok, %{user_id: user_id, logged_in_at: DateTime.utc_now()}}
  end

  @impl true
  def handle_call(:get_session, _from, state) do
    {:reply, state, state}
  end
end
children = [
  # Registryを追加
  {Registry, keys: :unique, name: AutoRecovery.Registry},
  # ...
]

4. Telemetry

メトリクスの収集とモニタリング。

defp deps do
  [
    {:telemetry, "~> 1.0"},
    {:telemetry_metrics, "~> 0.6"},
    {:telemetry_poller, "~> 1.0"}
  ]
end
defmodule AutoRecovery.Telemetry do
  use Supervisor
  import Telemetry.Metrics

  def start_link(arg) do
    Supervisor.start_link(__MODULE__, arg, name: __MODULE__)
  end

  @impl true
  def init(_arg) do
    children = [
      {:telemetry_poller, measurements: periodic_measurements(), period: 10_000}
    ]

    Supervisor.init(children, strategy: :one_for_one)
  end

  def metrics do
    [
      # カウンター
      counter("auto_recovery.api.requests.total"),
      counter("auto_recovery.api.errors.total"),

      # ヒストグラム
      distribution("auto_recovery.api.request.duration",
        unit: {:native, :millisecond}
      ),

      # 最終値
      last_value("vm.memory.total", unit: {:byte, :megabyte}),
      last_value("vm.total_run_queue_lengths.total")
    ]
  end

  defp periodic_measurements do
    []
  end

  # イベントの発行
  def emit_api_request(duration_ms) do
    :telemetry.execute(
      [:auto_recovery, :api, :request],
      %{duration: duration_ms},
      %{}
    )
  end
end

テスト戦略

プロセスクラッシュのテスト

defmodule AutoRecovery.CrashableWorkerTest do
  use ExUnit.Case
  alias AutoRecovery.CrashableWorker

  setup do
    # テスト用に新しいプロセスを起動
    {:ok, pid} = CrashableWorker.start_link([])
    %{pid: pid}
  end

  test "worker handles work correctly", %{pid: _pid} do
    assert :ok = CrashableWorker.do_work()
    state = CrashableWorker.get_state()
    assert state.work_count == 1
  end

  test "worker restarts after crash" do
    # クラッシュ前の状態を取得
    CrashableWorker.do_work()
    state_before = CrashableWorker.get_state()
    assert state_before.work_count == 1

    # クラッシュさせる
    catch_exit(CrashableWorker.crash())

    # 少し待つ（再起動を待つ）
    Process.sleep(100)

    # 再起動後の状態を確認
    state_after = CrashableWorker.get_state()
    assert state_after.work_count == 0  # 状態がリセット
  end
end

リトライのテスト

defmodule AutoRecovery.RetryTest do
  use ExUnit.Case
  alias AutoRecovery.Retry

  test "retry succeeds on first attempt" do
    func = fn -> {:ok, "success"} end
    assert {:ok, "success"} = Retry.retry(func)
  end

  test "retry succeeds after failures" do
    # 最初の2回は失敗、3回目に成功
    agent = start_supervised!({Agent, fn -> 0 end})

    func = fn ->
      count = Agent.get_and_update(agent, fn state ->
        {state, state + 1}
      end)

      if count < 2 do
        {:error, :temporary_failure}
      else
        {:ok, "success"}
      end
    end

    assert {:ok, "success"} = Retry.retry(func, 3, 10)
  end

  test "retry fails after max attempts" do
    func = fn -> {:error, :permanent_failure} end
    assert {:error, :max_retries_exceeded} = Retry.retry(func, 3, 10)
  end
end

サーキットブレーカーのテスト

defmodule AutoRecovery.CircuitBreakerTest do
  use ExUnit.Case
  alias AutoRecovery.CircuitBreaker

  setup do
    {:ok, cb} = CircuitBreaker.start_link(
      failure_threshold: 3,
      timeout_ms: 100
    )
    %{circuit_breaker: cb}
  end

  test "circuit breaker opens after threshold", %{circuit_breaker: cb} do
    failing_func = fn -> raise "error" end

    # 最初の2回は失敗してもclosed
    assert {:error, _} = CircuitBreaker.call(cb, failing_func)
    assert {:error, _} = CircuitBreaker.call(cb, failing_func)
    assert :closed = CircuitBreaker.status(cb)

    # 3回目でopenになる
    assert {:error, _} = CircuitBreaker.call(cb, failing_func)
    assert :open = CircuitBreaker.status(cb)

    # openの間はリクエストが遮断される
    assert {:error, :circuit_open} = CircuitBreaker.call(cb, failing_func)
  end

  test "circuit breaker transitions to half_open", %{circuit_breaker: cb} do
    failing_func = fn -> raise "error" end

    # openにする
    CircuitBreaker.call(cb, failing_func)
    CircuitBreaker.call(cb, failing_func)
    CircuitBreaker.call(cb, failing_func)
    assert :open = CircuitBreaker.status(cb)

    # タイムアウトを待つ
    Process.sleep(150)

    # 次のリクエストでhalf_openに移行
    success_func = fn -> :ok end
    assert {:ok, :ok} = CircuitBreaker.call(cb, success_func)

    # 成功したのでclosedに戻る
    assert :closed = CircuitBreaker.status(cb)
  end
end

本番環境での運用

ロギング

import Config

config :logger, level: :info

config :logger, :console,
  format: "$time $metadata[$level] $message\n",
  metadata: [:request_id, :user_id, :module]
defmodule AutoRecovery.LoggerHelper do
  require Logger

  def log_with_metadata(level, message, metadata \\ []) do
    Logger.metadata(metadata)
    Logger.log(level, message)
    Logger.reset_metadata()
  end

  def log_error_with_stacktrace(error, stacktrace) do
    Logger.error("""
    Error: #{inspect(error)}
    Stacktrace:
    #{Exception.format_stacktrace(stacktrace)}
    """)
  end
end

メトリクスの収集

defmodule AutoRecovery.Metrics do
  use GenServer
  require Logger

  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  end

  def record_api_call(duration_ms, status) do
    GenServer.cast(__MODULE__, {:record_api_call, duration_ms, status})
  end

  def get_stats do
    GenServer.call(__MODULE__, :get_stats)
  end

  @impl true
  def init(_opts) do
    state = %{
      total_calls: 0,
      successful_calls: 0,
      failed_calls: 0,
      total_duration_ms: 0
    }
    {:ok, state}
  end

  @impl true
  def handle_cast({:record_api_call, duration_ms, status}, state) do
    new_state = %{
      state |
      total_calls: state.total_calls + 1,
      total_duration_ms: state.total_duration_ms + duration_ms
    }

    new_state = case status do
      :success -> %{new_state | successful_calls: new_state.successful_calls + 1}
      :failure -> %{new_state | failed_calls: new_state.failed_calls + 1}
    end

    {:noreply, new_state}
  end

  @impl true
  def handle_call(:get_stats, _from, state) do
    avg_duration = if state.total_calls > 0 do
      state.total_duration_ms / state.total_calls
    else
      0
    end

    stats = Map.put(state, :avg_duration_ms, avg_duration)
    {:reply, stats, state}
  end
end

デバッグ

# 実行中のプロセスを確認
iex> Process.list() |> length()
42

# 特定のプロセスの情報
iex> pid = Process.whereis(AutoRecovery.ApiClient)
iex> Process.info(pid)

# Supervisorの子プロセスを確認
iex> Supervisor.which_children(AutoRecovery.Supervisor)
[
  {AutoRecovery.ApiClient, #PID<0.200.0>, :worker, [AutoRecovery.ApiClient]},
  {AutoRecovery.CrashableWorker, #PID<0.199.0>, :worker, [AutoRecovery.CrashableWorker]}
]

# プロセスの状態を取得
iex> :sys.get_state(AutoRecovery.ApiClient)
%AutoRecovery.ApiClient.State{...}

まとめ

OTPの利点

  1. 自動復旧: Supervisorによる自動再起動
  2. 障害の局所化: プロセス単位での障害管理
  3. 状態の分離: 各プロセスが独立した状態を持つ
  4. 宣言的な設計: Supervisorツリーで構造を表現
  5. スケーラビリティ: 数百万のプロセスを管理可能

ベストプラクティス

  1. "Let it crash"を実践する

    • 防御的プログラミングを避ける
    • 異常な状態でクラッシュさせる
    • Supervisorに復旧を任せる

  2. 適切な再起動戦略を選択する

    • 独立したワーカー: :one_for_one
    • 相互依存: :one_for_all
    • 順序依存: :rest_for_one

  3. リトライ戦略を適切に選択する

    • 一時的なエラー: 指数バックオフ
    • 外部サービス: サーキットブレーカー
    • 重要な処理: デッドレターキュー

  4. 状態を最小限にする

    • GenServerの状態は必要最小限に
    • 永続化が必要なデータはデータベースへ
    • 状態の再構築が容易な設計

  5. 監視とロギングを充実させる

    • Telemetryでメトリクス収集
    • 適切なログレベルの使用
    • エラー時のスタックトレース記録

アンチパターン

  1. 過度な防御的プログラミング

    # ❌ 悪い例
def process(data) do
  if data != nil and is_map(data) and Map.has_key?(data, :key) do
    # ...
  else
    {:error, :invalid_data}
  end
end

# ✅ 良い例
def process(%{key: value}) do
  # パターンマッチで失敗させる
end

  2. Supervisorなしでの長時間実行プロセス

    # ❌ 悪い例: Supervisorなし
{:ok, pid} = GenServer.start_link(MyWorker, [])

# ✅ 良い例: Supervisorで監視
children = [{MyWorker, []}]
Supervisor.start_link(children, strategy: :one_for_one)

  3. 状態の過度な蓄積

    # ❌ 悪い例: すべてのリクエストを保存
def handle_call(:request, _from, state) do
  new_state = Map.update(state, :requests, [], fn list ->
    [request | list]  # メモリリーク
  end)
end

# ✅ 良い例: 必要な情報のみ保持
def handle_call(:request, _from, state) do
  new_state = %{state | last_request_at: DateTime.utc_now()}
end

  4. リトライのない外部呼び出し

    # ❌ 悪い例: リトライなし
HTTPoison.get(url)

# ✅ 良い例: リトライあり
ExponentialBackoff.retry(fn -> HTTPoison.get(url) end)

次のステップ

本稿で示した自動復旧アーキテクチャを基に、以下の発展的なトピックに取り組まれたい：

  • 分散システム: ノード間のプロセス監視
  • 永続化: データベースとの統合
  • メッセージキュー: RabbitMQ、Kafkaとの連携
  • クラスタリング: 複数ノードでの負荷分散
  • ホットコードアップグレード: 無停止でのデプロイ

Elixir OTPは、堅牢で保守性の高いシステムを構築するための強力な基盤を提供する。

記事一覧に戻る