groupcache のピアを memberlist で自動管理


[groupcache‎ で組み込み型分散キャッシュ](http://qiita.com/yosisa/items/4980559f328fb31c3a8d)で groupcache‎ について説明しました。

groupcache を使うにはピア管理が課題になります。解決方法はいくつかありますが、今回は [serf](http://www.serfdom.io/) がメンバー管理に使っている [memberlist](https://github.com/hashicorp/memberlist) パッケージを使って自動化してみます。

## コード
```go
package main

import (
	"encoding/json"
	"fmt"
	"github.com/codegangsta/martini"
	"github.com/golang/groupcache"
	"github.com/hashicorp/memberlist"
	"log"
	"net/http"
	"os"
	"strings"
	"time"
)

const GroupcachePort = 8000

type eventDelegate struct {
	peers []string
	pool  *groupcache.HTTPPool
}

func (e *eventDelegate) NotifyJoin(node *memberlist.Node) {
	uri := e.groupcacheURI(node.Addr.String())
	e.removePeer(uri)
	e.peers = append(e.peers, uri)
	if e.pool != nil {
		e.pool.Set(e.peers...)
	}
	log.Print("Add peer: " + uri)
	log.Printf("Current peers: %v", e.peers)
}

func (e *eventDelegate) NotifyLeave(node *memberlist.Node) {
	uri := e.groupcacheURI(node.Addr.String())
	e.removePeer(uri)
	e.pool.Set(e.peers...)
	log.Print("Remove peer: " + uri)
	log.Printf("Current peers: %v", e.peers)
}

func (e *eventDelegate) NotifyUpdate(node *memberlist.Node) {
	log.Print("Update the node: %+v\n", node)
}

func (e *eventDelegate) groupcacheURI(addr string) string {
	return fmt.Sprintf("http://%s:%d", addr, GroupcachePort)
}

func (e *eventDelegate) removePeer(uri string) {
	for i := 0; i < len(e.peers); i++ {
		if e.peers[i] == uri {
			e.peers = append(e.peers[:i], e.peers[i+1:]...)
			i--
		}
	}
}

func initGroupCache() {
	eventHandler := &eventDelegate{}
	conf := memberlist.DefaultLANConfig()
	conf.Events = eventHandler
	if addr := os.Getenv("GROUPCACHE_ADDR"); addr != "" {
		conf.AdvertiseAddr = addr
	}

	list, err := memberlist.Create(conf)
	if err != nil {
		panic("Failed to created memberlist: " + err.Error())
	}

	self := list.Members()[0]
	addr := fmt.Sprintf("%s:%d", self.Addr, GroupcachePort)
	eventHandler.pool = groupcache.NewHTTPPool("http://" + addr)
	go http.ListenAndServe(addr, eventHandler.pool)

	if nodes := os.Getenv("JOIN_TO"); nodes != "" {
		if _, err := list.Join(strings.Split(nodes, ",")); err != nil {
			panic("Failed to join cluster: " + err.Error())
		}
	}
}

func main() {
	initGroupCache()
	heavy := groupcache.NewGroup("heavy", 64<<20, groupcache.GetterFunc(heavyTask))

	m := martini.Classic()
	m.Get("/_stats", func() []byte {
		v, err := json.Marshal(&heavy.Stats)
		if err != nil {
			panic(err)
		}
		return v
	})
	m.Get("/:key", func(params martini.Params) string {
		var result string
		if err := heavy.Get(nil, params["key"], groupcache.StringSink(&result)); err != nil {
			panic(err)
		}
		return result
	})
	m.Run()
}

func heavyTask(ctx groupcache.Context, key string, dst groupcache.Sink) error {
	time.Sleep(400 * time.Millisecond)
	dst.SetString("Value of " + key)
	return nil
}
```

ポイントは `memberlist.EventDelegate` インタフェースを実装して `memberlist.Config.Events` に代入することです。これにより、クラスタ構成に変化があったときにコールバックしてもらえます。

### 使い方
このプログラムは以下のポートを使います。

* 3000: サービス提供
* 7946: memberlist のクラスタ管理
* 8000: groupcache の内部 RPC

ポートを変更できるようにしていないので、VM や Docker を使って2台以上のホストを用意します。

1つめを以下のコマンドで起動します。

```bash
$ GROUPCACHE_ADDR=192.168.158.1 go run main.go
```

複数の IP アドレスを持つホストの場合、`GROUPCACHE_ADDR` に groupcache で使いたい IP アドレスを指定します。memberlist もこのアドレスを使って死活監視を行います。`GROUPCACHE_ADDR` が指定されなかった場合は自動判別を試みますが、うまくいかない場合は明示的に指定してください。

2つめ以降を以下のコマンドで起動します。

```bash
$ GROUPCACHE_ADDR=192.168.158.2 JOIN_TO=192.168.158.1 go run main.go
```

`JOIN_TO` で指定したホストに起動時に自動接続します。以降は memberlist で死活監視を行い、ピアの DOWN/UP を検知すると自動的に groupcache のピアリストが更新されます。

curl を使ってそれぞれのプロセスにリクエストを出せば、groupcache のクラスタが正しく構成されていることが確認できます。

```bash
$ curl 192.168.158.1:3000/foo  # 400ms
$ curl 192.168.158.2:3000/foo  # <1ms
```

## おわりに
groupcache と memberlist を組み合わせると、ビルドしたバイナリファイルを各ホストに置いて起動するだけで、キャッシュを共有しながらシステムをスケールアウトさせることができます。

死活監視も同じプロセス内で行っているため、非常にシンプルなシステム構成になっています。デプロイについてあれこれ考えなくてすむので、便利ですね。