makoto-developer's テックブログ

Elixir OTPで分散システムを設計する方法(私の考え方を紹介)

#Elixir #分散システム

Elixir/Erlang OTPは、分散システム構築に最適なプラットフォームである。しかし、その設計には独特のパターンと考え方が必要である。本記事では、10年以上のElixir開発経験から得た、分散システム設計の実践的なアプローチを解説する。

OTPの基本思想

Let it crash

従来の考え方:
  エラーを予測し、すべてに対処する

OTPの考え方:
  エラーは発生するもの。
  クラッシュさせて、Supervisorが再起動する

これは、防御的プログラミングから回復可能なシステムへの転換である。

Supervision Tree

Application
  ├─ Supervisor
  │   ├─ Worker 1
  │   ├─ Worker 2
  │   └─ Supervisor
  │       ├─ Worker 3
  │       └─ Worker 4
  └─ Supervisor
      └─ Worker 5

システム全体を木構造として設計する。

GenServerの設計パターン

1. 状態管理

defmodule MyApp.Counter do
  use GenServer

  # Client 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

  # Server Callbacks
  def init(initial_value) do
    {:ok, initial_value}
  end

  def handle_call(:increment, _from, state) do
    {:reply, :ok, state + 1}
  end

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

原則: GenServerは単一の責務を持つ。

2. 非同期処理

defmodule MyApp.Logger do
  use GenServer

  def log(message) do
    GenServer.cast(__MODULE__, {:log, message})
  end

  def handle_cast({:log, message}, state) do
    IO.puts("[LOG] #{message}")
    {:noreply, state}
  end
end

原則: 応答が不要ならcastを使う。

3. タイムアウト

defmodule MyApp.Cache do
  use GenServer

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

  def handle_info(:cleanup, state) do
    new_state = remove_expired(state)
    schedule_cleanup()
    {:noreply, new_state}
  end

  defp schedule_cleanup do
    Process.send_after(self(), :cleanup, :timer.minutes(5))
  end
end

原則: 定期的な処理はhandle_infoで実装する。

分散システムの設計

1. ノード間通信

# ノード起動
iex --name node1@127.0.0.1 --cookie secret

# 別のターミナルで
iex --name node2@127.0.0.1 --cookie secret

# node1から接続
Node.connect(:"node2@127.0.0.1")

# node2のプロセスを呼び出し
:rpc.call(:"node2@127.0.0.1", MyApp.Counter, :increment, [])

原則: cookieが一致するノードのみ通信可能。

2. 分散レジストリ

defmodule MyApp.Registry do
  use GenServer

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

  def register(name, pid) do
    GenServer.call({:global, __MODULE__}, {:register, name, pid})
  end

  def lookup(name) do
    GenServer.call({:global, __MODULE__}, {:lookup, name})
  end

  def handle_call({:register, name, pid}, _from, state) do
    {:reply, :ok, Map.put(state, name, pid)}
  end

  def handle_call({:lookup, name}, _from, state) do
    {:reply, Map.get(state, name), state}
  end
end

原則: :globalを使うと、クラスタ全体で一意の名前を持てる。

3. ネットワーク分断への対処

defmodule MyApp.NodeMonitor do
  use GenServer

  def init(_) do
    :net_kernel.monitor_nodes(true)
    {:ok, %{}}
  end

  def handle_info({:nodeup, node}, state) do
    IO.puts("Node connected: #{node}")
    # 再同期処理
    sync_data(node)
    {:noreply, state}
  end

  def handle_info({:nodedown, node}, state) do
    IO.puts("Node disconnected: #{node}")
    # フェイルオーバー処理
    failover(node)
    {:noreply, state}
  end

  defp sync_data(node) do
    # データ同期ロジック
  end

  defp failover(node) do
    # フェイルオーバーロジック
  end
end

原則: ネットワーク分断は必ず発生する。対処を設計に組み込む。

実例: 分散カウンター

要件

  • 複数ノードで動作
  • ノードが落ちても、他のノードで継続
  • カウンターの値は、ノード間で同期

実装

defmodule MyApp.DistributedCounter do
  use GenServer

  # Client API
  def start_link(_) do
    GenServer.start_link(__MODULE__, 0, name: {:global, __MODULE__})
  end

  def increment do
    case :global.whereis_name(__MODULE__) do
      :undefined -> {:error, :not_found}
      pid -> GenServer.call(pid, :increment)
    end
  end

  def get do
    case :global.whereis_name(__MODULE__) do
      :undefined -> {:error, :not_found}
      pid -> GenServer.call(pid, :get)
    end
  end

  # Server Callbacks
  def init(initial_value) do
    # クラスタ内の他のノードに通知
    broadcast_state(initial_value)
    {:ok, initial_value}
  end

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

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

  def handle_info({:sync, value}, _state) do
    {:noreply, value}
  end

  defp broadcast_state(value) do
    for node <- Node.list() do
      send({__MODULE__, node}, {:sync, value})
    end
  end
end

起動

# ノード1
iex --name counter1@127.0.0.1 --cookie secret -S mix

# ノード2
iex --name counter2@127.0.0.1 --cookie secret -S mix

# ノード1から接続
Node.connect(:"counter2@127.0.0.1")

# ノード1でカウンター起動
MyApp.DistributedCounter.start_link(nil)

# ノード1で操作
MyApp.DistributedCounter.increment()  # => 1
MyApp.DistributedCounter.increment()  # => 2

# ノード2で確認
MyApp.DistributedCounter.get()  # => 2

耐障害性の設計

1. 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 = [
      {MyApp.Counter, 0},
      {MyApp.Logger, []},
      {MyApp.Cache, %{}}
    ]

    # 戦略を選択
    Supervisor.init(children, strategy: :one_for_one)
  end
end

戦略:

  • :one_for_one: 1つのWorkerがクラッシュしたら、そのWorkerのみ再起動
  • :one_for_all: 1つのWorkerがクラッシュしたら、すべてのWorkerを再起動
  • :rest_for_one: 1つのWorkerがクラッシュしたら、それ以降のWorkerを再起動

2. 再起動の制限

Supervisor.init(children,
  strategy: :one_for_one,
  max_restarts: 3,
  max_seconds: 5
)

原則: 5秒以内に3回以上クラッシュしたら、Supervisor自身もクラッシュする。

3. DynamicSupervisor

defmodule MyApp.WorkerSupervisor do
  use DynamicSupervisor

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

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

  def start_worker(args) do
    spec = {MyApp.Worker, args}
    DynamicSupervisor.start_child(__MODULE__, spec)
  end

  def stop_worker(pid) do
    DynamicSupervisor.terminate_child(__MODULE__, pid)
  end
end

原則: 動的にWorkerを追加・削除する場合はDynamicSupervisorを使う。

データの一貫性

1. CRDTs (Conflict-free Replicated Data Types)

# Horde ライブラリを使用
defmodule MyApp.Application do
  use Application

  def start(_type, _args) do
    children = [
      {Horde.Registry, [name: MyApp.Registry, keys: :unique]},
      {Horde.DynamicSupervisor, [name: MyApp.Supervisor, strategy: :one_for_one]}
    ]

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

原則: CRDTsを使えば、ノード間で自動的にデータが同期される。

2. イベントソーシング

defmodule MyApp.EventStore do
  def append_event(stream, event) do
    # イベントを追記
    EventStore.append_to_stream(stream, event)
  end

  def replay_events(stream) do
    # イベントを再生して状態を復元
    EventStore.read_stream_forward(stream)
    |> Enum.reduce(%{}, fn event, state ->
      apply_event(state, event)
    end)
  end

  defp apply_event(state, %{type: "user_created", data: data}) do
    Map.put(state, :user, data)
  end

  defp apply_event(state, %{type: "user_updated", data: data}) do
    Map.update!(state, :user, &Map.merge(&1, data))
  end
end

原則: 状態ではなく、イベントを保存する。

パフォーマンス最適化

1. ETS (Erlang Term Storage)

defmodule MyApp.FastCache do
  def start_link do
    :ets.new(:my_cache, [:set, :public, :named_table])
    {:ok, self()}
  end

  def put(key, value) do
    :ets.insert(:my_cache, {key, value})
  end

  def get(key) do
    case :ets.lookup(:my_cache, key) do
      [{^key, value}] -> {:ok, value}
      [] -> {:error, :not_found}
    end
  end
end

原則: 高速な読み込みが必要なら、ETSを使う。

2. :persistent_term

# 起動時に設定
:persistent_term.put(:config, %{api_key: "secret"})

# どこからでも高速に読み込み
config = :persistent_term.get(:config)

原則: 不変のデータは:persistent_termに保存する。

モニタリング

1. Telemetry

defmodule MyApp.Counter do
  def increment do
    start_time = System.monotonic_time()

    result = GenServer.call(__MODULE__, :increment)

    duration = System.monotonic_time() - start_time

    :telemetry.execute(
      [:my_app, :counter, :increment],
      %{duration: duration},
      %{result: result}
    )

    result
  end
end

# ハンドラー登録
:telemetry.attach(
  "my-handler",
  [:my_app, :counter, :increment],
  fn _event, measurements, _metadata, _config ->
    IO.puts("Duration: #{measurements.duration}ns")
  end,
  nil
)

原則: すべての重要な操作を計測する。

2. Observer

iex -S mix
iex> :observer.start()

原則: 本番環境でも:observerでリアルタイム監視できる。

まとめ

Elixir/Erlang OTPで分散システムを設計する際の原則：

  1. Let it crash: エラーは発生するもの。回復可能な設計にする
  2. Supervision Tree: システムを木構造として設計する
  3. 単一責務: GenServerは1つのことだけを行う
  4. ネットワーク分断への対処: 必ず発生する。設計に組み込む
  5. CRDTs: データの一貫性を自動的に保つ
  6. イベントソーシング: 状態ではなくイベントを保存
  7. 計測: すべての重要な操作を計測する

OTPは、30年以上の通信システム開発の知見が詰まったプラットフォームである。その設計思想を理解し、適切に活用すれば、高い耐障害性を持つ分散システムを構築できる。

記事一覧に戻る