Erlang アドベントカレンダー 2014の23日目の記事です。
Erlang/OTPでアプリケーションを書いていると、システムを冗長化するために複数ノードでうまく協調動作するようにさせるために、Distributed Erlangの上に構築されたFailoverやTakeoverを使う場面がいずれ出てくる。しかし、これらの仕組みは、Riakのようにシステムをスケールアウトさせたい場合には不十分だ。スケールアウトするシステムの本質は
- アクセスしたいモノの物理的な位置を隠蔽して論理的な位置でアクセスできるようにする
- 物理的な位置が故障やスケールアウトのために変化しても常に追跡できて同じ論理的な位置でアクセスする
- アクセスしたいモノが偏らず、ほぼ均等に分散されている
の3点がサポートされていることだ。これだけだといろんなものが該当するが、 Riak風に翻訳すると
- アクセスしたいデータがどのノードのどのディスク、どのファイルに入っているかを隠蔽して、キーが分かればアクセスできる
- 故障やノード追加が行われても同じキーで同様にアクセスできること
- アクセスしたいデータが各ノードに均等に分散配置されいていること
ということになる。よくよく読むと、扱う対象が、実はKVSである必要がないということだ。 Riak というデータベースの構成は、 Riak Core という分散フレームワークの上に載った Riak KV というアプリケーションであるということだ。 Riak KV は Riak Core 上で動作することを前提としているが、 vnode の動作は単なるローカルのKVSである（ちょっとRiak Coreの機能を使っている）。つまり、 Riak Core という分散フレームワークを使えばあなたのアプリケーションをvnodeに乗せてスケールアウトできるようになる！というわけである。絵でいうとこんな感じだ。
ちょっと調べたところRiak以外でも（驚くべきことに！）いくつか使われているようで、最大のクラスタはOpenXの125台のもののようだ。
参考
- 簡単な解説から→Where To Start With Riak Core | Basho Technologies
- ちょっとフレームワークを使ってアプリを作ってみたい場合は→rzezeski/try-try-try · GitHub
- ソースを読みたい人は→basho/riak_core · GitHub
他にも、Riakの中だと riak_pipe というMapReduceのためのサブシステムがRiak Coreの上に分散アプリケーションとして載っている。
もうちょっと詳しく: 何ができるの？
もうちょっというと、いままで Erlang node というものにアプリをくっつけていろいろ扱ってきたものを、間にひとつ vnode というものを挟むことになる。 vnode にアプリをくっつけて分散配置するのだ。手書きの温もりあふれる絵でかくとこうなる。
リクエストのルーティング
Riak Core の用語では、リクエストは「 command 」になる。Erlangのコードの中で vnode に対してコマンドを投げるとそれを実行して、結果を返してくれるというイメージだ。どの vnode にコマンドを投げるべきかは、キーを渡してハッシュを計算すると vnode のIDが返ってくるようになっている。 "riak_core_util:chash_key/1" という関数がそれにあたる。
> DocIdx = riak_core_util:chash_key({spam, ham}). <<220,127,3,124,116,114,28,36,44,206,203,237,43,32,170,68,62,127,127,133>>
この関数の中は決定論的で特に副作用はないので*1、何度実行しても同じ値が返ってくる。Consitent Hashingでいうところのハッシュ値だ。これを使って Preference List をつくる。みっつめの引数は、 Riak Core アプリケーションの名前だ。ここではRiakを使っているので、 riak_kv という名前を使う。
> riak_core_apl:get_apl(DocIdx, 3, riak_kv). [{1278813932664540053428224228626747642198940975104, 'riak@127.0.0.1'}, {1301649895747835411525156804137939564381064921088, 'riak@127.0.0.1'}, {1324485858831130769622089379649131486563188867072, 'riak@127.0.0.1'}]
2番めの引数は何個冗長化したいかという（合計でなんこの vnode にアクセスしたいのか？という）意味だ。これで、3つの vnode のIDと、それを持っているノードが判明する。あとはこれをもとに、 riak_core_vnode_master:command/4 を使えば vnode の handle_command が呼び出されて、アプリケーション特有の処理をすることができる。
vnode の移動、転送、破棄、作成をサポート
いくつかのコールバックを作成することで、 vnode の各種操作ができるようになる。アプリケーション毎に冗長性や移動、転送などのセマンティクスは異なるだろうから、アプリ側で作りこんでやる必要がある。簡単なコールバックで抽象化されていることから、難しい作り込みをしなくて済むようになっている。
故障時のフォールバックの動作をサポート
図のように、たとえばあるノードが故障なりダウンなりしていてそのノードが持っている vnode にアクセスできなかったとしよう。Riak Coreでは、この故障を検知して一時的にリストから外して、代替の vnode を用意する仕組みが用意されている。図のように V' にコマンドをフォールバックして別のノードで処理を代替させることができる。そのときも、本来存在しないはずの vnode を一時的に作成してセマンティクスを変えないで済むようになっている。
この代替処理のせいで、ネットワーク切断なんかのときはデータがdivergeしてしまうのだけど、そういうのも上手く処理するためのベクタークロック（ vclock.erl ）も用意されているので利用することができる。
つまり、Riakの高可用性の中心的な機能はフレームワークとして切り出されているので、Riak Coreを使えば誰でも高可用な分散アプリケーションを書けるようになるというわけだ。たとえば分散キューをほしいと思っている人は一部にいるだろうけど、OpenXはRiak Coreを使って広告表示フレクエンシーカウンターを作っている。噂によるとかなりの大きさのクラスタになっている（推測が正しければ、上記で出てきた125台のクラスタのことだと思うんだけど…）。
ノードの追加、削除、変更、ステータスチェックなど
riak_core_console.erl をみれば分かるのだが、Riakの cluster コマンドはほぼ全てこのモジュールにある関数を呼んでいるだけである。つまり、 vnode に紐づくアプリを実装して、各種 handle_* のコールバックを実装していけば、自動的にノードの追加、削除、変更などの各種操作のコンソールも作成できるのである。これと clique と以前紹介した node_package を使えば、それっぽく動く分散アプリケーションを構築することができるようになっている。そいつのテストは riak_testで自動化 してしまえばよい。これで、スケールアウトや冗長化の難しいところ、面倒なところをほとんどがフレームワークで済ませられるので、開発者は分散まわり以外のアプリケーションの本質的な部分に集中できるという寸法だ。もちろん動作やアップデートは正しく理解して追っていかなければいけないが、自分でつくり上げて苦労するよりは随分楽だろう。
こんなのがほしかった（感想）
4年ほど前にJubatusというスケールアウトさせたいソフトウェアがあった。当時ぼくが直面していた問題は、機械学習のコアモジュールをどうやってDHTに乗せて分散させるかというものだった（BigTable風にヒエラルキアルな分散するシステムも同時期に開発していたが、わりとしんどかったのでDHTをやってみようと思っていた）。そのときコレを知っていれば（そしてC++と親和性があればナア）と、Bashoに入ってから歯痒い思いをしたものであった。
まとめ
Riakが利用している分散アプリケーション開発用のフレームワーク、Riak Coreの紹介をした。こういったことにどっぷり浸かりたい方は是非、わたしの職場で一緒にハマりましょう。
*1:いろんなコンテキスト的情報を利用するのでピュ〜ア〜ではない
