Finagle の Service Discovery に Consul を利用する |

Blog

2016.12.22

Finagle の Service Discovery に Consul を利用する

こんにちは、AI Studio dev group の阿川です。

前回の記事で少し言及しましたが、私達のプロジェクトでは Finagle を使っていまして、今回は Finagle の Service Discovery に Consul を組み合わせた話をします。

Consul とプロジェクトにおける利用方法については「社内向け技術カンファレンス ADC2016 より「Axion meets HashiCorp」 | Tech Blog」で紹介しているので、そちらもご覧ください。

まずは Finagle の Name のおさらいです。

`# com.twitter.finagle.Name`

Finagle client を生成するとき、接続先サーバのネットワーク位置を Name 型で表現される形式で指定します。

Name 型には以下の 2 つの表現があります。

`case class Name.Bound(va: Var[Addr])`

“scheme!arg” 形式の文字列(例えば “inet!twitter.com:80”)は Resolver によって Name.Bound へ解決されます。
新しい scheme に対応させるには trait Resolver を実装します。

`case class Name.Path(path: Path)`

スラッシュから始まる “/$/namer/path..” 形式の文字列(例えば “/$/inet/twitter.com/80”)は Name.Path に解釈され、Namer によって最終的に Name.Bound へ解決されます。
新しい namer に対応させるには abstract class Namer を実装します。

文字列が Name に解決される過程に違いはあれど、どちらも最終的に物理的なネットワーク位置を表現する Var[Addr] へ行き着きます。これが Finagle における Service Discovery と言えます。

より正確には Finagle Service を Service registry へ登録する Announcer も Service discovery に深く関係しますが、今回は Service の解決に焦点を当てています。

余談になりますが、実際に Finagle を利用するとき “scheme!arg” と “/$/namer/path” のどちらで接続先を指定するべきか迷うことがあるかと思います(私は迷いました)。
これについては「Finagle client が接続する Service がさらに他の Service に依存していなければ (Service 内に他 Service へ接続する Finagle client が居なければ) 前者の Resolver 形式で十分、そうでなければ後者の Namer 形式を利用するべき」と認識しています。
なぜなら Namer 形式では Dtab (“Delegate TABle” の略らしいです) によって、最小粒度では Request 単位で Namer パスの解決方法をカスタマイズすることが可能なので、特定の Request だけ Backend Service を Canary 版に差し替える、など柔軟な運用が可能になるためです。

話を戻して Var[Addr] についてもおさらいします。

`# Var[Addr]`

Addr は仮想のネットワーク位置 Name を解決した結果として物理的なネットワーク位置を示す型で、以下の 4 つの状態があります。

`object Pending`

Name が未解決である状態です。例えば DNS の解決や ZooKeeper の操作の完了を待っている場合です。

`object Neg`

Name が無効である状態です。例えば DNS が NXDOMAIN ステータスを返した場合です。

case class Failed(cause: Throwable)

Name の解決に失敗した状態です。例えば DNS 問い合わせ時にネットワークのエラーが発生した場合などです。Neg との違いは、リトライすることで正常に解決できる可能性があることです。

`case class Bound(addrs: Set[Address])`

Name の解決が成功した状態です。 Set[Address] が物理的なエンドポイントの集合をあらわします。 Address は java.net.InetSocketAddress です。

Addr を内包しているcom.twitter.util.Var[+T] ですが、これは動的に変更される値のコンテナです。
Finagle は Var[Addr] の変更を検知して Client 側 Load Balancer の Pool を更新しています。

以上、 Finagle の Service discovery 機構のおさらいでした。

# ConsulResolver の実装

というわけで、 “consul!consul.local:8500/service-name” 形式を解決できる ConsulResolver のコードを Gistに載せました。コードは実際にプロジェクトで利用しているものです。
本来はちゃんとライブラリとして Publish したいところですが、Namer/Announcer を実装していなかったりと自分達に必要なユーズケースしか対応しておらず不完全なのと、メンテナンスする時間も取れないので見送っています。

以下、ポイントとなる場所を見ていきましょう。

## Var[Addr] の生成

def addrOf(query: ConsulQuery): Var[Addr]

1	def addrOf(query: ConsulQuery): Var[Addr]

が肝心の Var[Addr] を生成して返す関数となっています。Gist のコードはリトライ時のウェイトなどで入り組んでいるので、説明のために少し簡略化するとこのようになります。

private def addrOf(query: ConsulQuery): Var[Addr] =
  Var.async(Addr.Pending: Addr) { u: Updatable[Addr] =>
    def update(consulIndex: Option[String]): Future[Unit] = {
      readCatalog(query, consulIndex).flatMap {
        case (addr, nextIndex) =>
          u() = addr
          update(nextIndex)
      }
    }

    val f = update(None).interruptible()

    new Closable {
      override def close(deadline: Time): Future[Unit] = {
        f.raise(new FutureCancelledException)
        f
      }
    }
  }

private def addrOf(query: ConsulQuery): Var[Addr] =

Var.async(Addr.Pending: Addr) { u: Updatable[Addr] =>

def update(consulIndex: Option[String]): Future[Unit] = {

readCatalog(query, consulIndex).flatMap {

case (addr, nextIndex) =>

u() = addr

update(nextIndex)

}

val f = update(None).interruptible()

new Closable {

override def close(deadline: Time): Future[Unit] = {

f.raise(new FutureCancelledException)

}

readCatalog が Consul の HTTP API を問い合わせる関数ですが、これは Consul の Blocking Query を利用しており、Consul の Service の状態が変化したとき(もしくはタイムアウトしたとき)に Future が完了するようになっています。
Future の完了後には、その結果で Var の update メソッドによりVar[Addr] を変化させ、再帰で readCatalog をループさせています。

Future が末尾再帰になっていますが、 com.twitter.util.Future は末尾再帰の最適化を備えているためコールスタックの消費を心配する必要がありません。この特長は Scala 2.12 の scala.concurrent.Future にも輸入されていますね。

ちなみにこの Future が failure 状態になると Future 再帰によるループが止まってしまうので、 readCatalog 関数が関す Future は常に successful とし、失敗は Addr.Failed にエンコードするようにしています。

## ServiceLoader

Resolver は Java の ServiceLoader 機構によって検索されます。自分で実装した Resolver を Finagle に組み込むためにはリソースに “META-INF/services/com.twitter.finagle.Resolver” ファイルを含める必要があります。
このファイルには Resolver の具象クラスを完全名で書いておけば OK です。

# 他の Finagle + Consul ライブラリ

この実装を行うにあたり、以下のリポジトリのコードを参考にしました。

https://github.com/kachayev/finagle-consul

ただ、こちらのコードは Consul の Blocking Query を使っておらず一定間隔のポーリングとなっていたため、メンバー変更への追随に遅れが発生してしまうことから、自前で Blocking Query を利用するように実装した、という経緯があります。
そして、この記事を書くにあたりもう一度調べたところ、以下のリポジトリで Blocking Query の対応が入っていたので、みなさんはこちらを利用されても良いかと思います。

https://github.com/ncbi/finagle-consul/
- Use Long Polling for PUSH notification about service status updates

# まとめ

今回は Finagle に Consul の Service Discovery を統合する方法を紹介しました。実際に実装してみて、 Finagle における Service Discovery の部分がプラガブルで拡張可能になっていることでシンプルに独自 Resolver を実装でき、 Finagle の良さをあらためて感じました。
これをベースにすれば、任意の Service Discovery 機構をさほど手間なく Finagle へ統合することができるので、各自のニーズに合わせてカスタマイズしてください。

以上、ありがとうございました。

Author

koji