mercari/datastore実戦投入

Datastoreについて

みなさまGCPをお使いになっているでしょうか。

GCPにはバックエンドのDBとしてCloudSQLというRDBと、NoSQLであるDatastoreというのがあります。

周囲の事例を聞く限りは、マスターデータなど変化が少なかったり、seedデータ的なものを用意しなければならないものをのぞいて、基本的にDatastoreを利用している印象です。

また、GAEで開発する場合はinternalなAPIからDatastoreを利用できる一方、GKEなどからは、GCPの外部向けAPIを呼び出すことでデータを送受信します。

さて、自分はいつもGoでAPIを書くのがいいよ！と触れ回ってるわけですが、GCP上で開発を進める際には前述の通りDBにはDatastoreを使っています。

このDatastoreですが、そのまま使うと自前でキーを発行する必要があったり、値をキャッシュしたりする際にごちゃごちゃした実装になりがちです。なので、ラッパーを自前で用意するか、クライアントパッケージを用いるのが基本となります。

GAEであれば、以下のgoonというパッケージを使うと、キーの発行からmemdへの自動キャッシュなど、便利機能の恩恵に預かることができます。構造体をPutすれば勝手にデータも入るし、IDも勝手に発行されるは速度が上がるわ、便利便利。

mjibson/goon

一方、GKEで開発してる際にはAPIを呼ぶわけですが、これに関しては僕は自前のラッパーを用意していて、以下のパッケージを使っていました。

timakin/gosto

こちらはgoonをパクっているのですが、個人的には不自由なくやれてました。ただ、GAEだからこそできる機能というのもいくつかあったり（memdへの自動キャッシュなど）、同じDBを使ってるのにクライアントパッケージが分かれる気持ち悪さ、メンテナンスへのモチベーションなど様々な面で問題を抱えていました。

mercari/datastoreについて

そんな課題を抱える中、先日@vvakameさんが作成された、mercari/datastore というパッケージが爆誕しました。

mercari/datastore

これは何かというと、AppEngineとCloud APIの共用Datastoreクライアントパッケージです。

インターフェースもgoonなどの既存の主流パッケージを利用していた方のために、ほぼ同一のものを提供してくれています。

結論から書くと、非常に有用ですし、基本的にGAEと他のCloud APIのDatastoreのパッケージが違うと、双方のメンテナンス度合いが違うことに不安を感じていた昨今の問題意識を払拭してくれるので最高です。みなさん導入しましょう。

僕は普段AppEngineでgoonを使っていましたが、それを捨ててmercari/datastoreに移行することにしました。さしあたっては、同じように利用したいけど、

• 移行コストどんなもんよ？
• 今までの実装をどう変えるのよ？
• 不安点、改善点などはあるのか？

ということについて、実戦投入目線でかいてみます。

移行コスト

結論、GCPUGのslackでvvakameさんに直接質問しながら進めたものの、３、４日かかりました。

これは後述しますが、IDの発行やトランザクションの扱い方について、微妙に不明点があって手間取りました。

とはいえコードの書き方自体はそんなに大きく変更はなく、超大幅なインターフェースの変更というのはなかったので、少なくともgoonとか、redisのクライアントライブラリに慣れてる方は速やかに利用を始めることができると思います。

実装の変更点

次に、実装の方法と、これまでgoonなどを使ってきた人でも注意しなければいけない点などをかいていきます。なお、今回はAppEngineの例に限ります。

AE/Cloudのどちらかを利用するか

mercari/gatastoreは、AppEngineとCloud PlatformのAPI両方に対して使えますが、ということはどちらに向けたクライアントとして利用するのかをどこかで宣言しなければなりません。

今回はAppEngineを利用するのですが、その場合はmainのimport節あたりで、aedatastoreとclouddatastoreという、どちらの実装を初期化するか宣言します。書き方は次の通りです。

package main

import (
  ...
  _ "go.mercari.io/datastore/aedatastore"
  ...
)

こうすると、初期化処理が走って、Client InterfaceのAppEngine用の実装を使えるようになります。

boom

API呼び出しをするためのパッケージに加え、goonを使っていた人向けにさらに直感的に使える、boom というパッケージが用意されています。例えば、以下のように初期化します。

func BoomFromContext(ctx context.Context) (*boom.Boom, error) {
	return boom.FromContext(ctx)
}

ここで、あえてboomの初期化をメソッド化しているのは、あとあとオプションなどがついたときのことを考えて、共通化しておきたかったからです。オプションの例は、memdキャッシュを利用するかなどです。今までgoonを使っていた人はほぼその機能を利用すると思うので、統一的にオプションを管理するために別メソッドに切り出しておきましょう。

ちなみに、先ほど記載したimportによる初期化をせずに boom.FromClient を呼ぶと、参照できる実装がなくてpanicするので注意しましょう。

CRUD

では、boomを使ってCRUDに関わる処理をかいていきましょう。

例えば、僕は自前の動画配信APIをかいていたりしますが、仮に動画のカテゴリを取得するには、以下のような構造体とメソッドを用意します。（一部フィールドを省略）

boomに自動でIDを振ってほしい場合は、フィールドタグにboomを指定しましょう。

type Category struct {
	ID   int64    `json:"id" datastore:"-" boom:"id"`
	Name string   `json:"name"`
}

IDをキーに取得する場合、以下のようなメソッドを用意すれば良いです。IDだけ入れた構造体を用意して、それをboomのGetというメソッドに渡してあげればフィールドが充足された構造体が返ってきます。

func (repo streamingRepository) GetCategoryByID(ctx context.Context, cID int64) (*entity.Category, error) {
	c := entity.Category{
		ID: cID,
	}
	b, err := library.BoomFromContext(ctx)
	if err != nil {
		return nil, err
	}
	if err := b.Get(&c); err != nil {
		return nil, err
	}

	return &c, nil
}

逆にCategoryを登録、削除する場合は以下のようになります。IDが入っていない状態で、Nameだけ入れてPutすると、自動でIDが採番されます。Deleteも同様の手順できます。

func (repo authRepository) UpsertCategory(tx *boom.Transaction, c *entity.Category) (*entity.Category, error) {
	if _, err := tx.Put(c); err != nil {
 		return nil, err
 	}
 	return c, nil
}

なお、Deleteする場合、goonはdatastoreのKeyを渡してあげる必要がありました。しかし、boomでは構造体をそのまま渡せば、内部でKeyを取得し、消してくれます。ここは痒いところに手が届くところです。

トランザクションとIDの採番

ユーザー登録などでは、１つのトランザクションで関連するKindをまとめて登録することで、Rollback可能にしなければならないこともあると思います。その場合は、RunInTransaction を呼びましょう。具体的に、ユーザーの基本情報とデバイスの情報を別々に保存する場合を想定すると、次のようになると思います。（こちらも簡略化のために大幅にフィールドを削っております）

b, err := library.BoomFromContext(ctx)
if err != nil {
	return nil, err
}
uKey, err := b.AllocateID(u)
if err != nil {
	return nil, err
}
_, err = b.RunInTransaction(func(tx *boom.Transaction) error {
	now := time.Now().Unix()
	u.ID = uKey.ID()
	u.Enabled = true
	u.CreatedAt = now
	u.UpdatedAt = now
	user, err = s.repo.UpsertUserTx(tx, u)
	if err != nil {
		return err
	}
	
    u.Device.UserID = u.ID
	u.Device.CreatedAt = now
	u.Device.UpdatedAt = now
	u.Device.OSTypeID = entity.OSNameToTypeID[u.Device.OSName]
	d, err := s.repo.UpsertDeviceTx(tx, u.Device)
	if err != nil {
		return err
	}
    user.Device = device

	return nil
})

UpsertUserTxというメソッドでユーザーを登録してます。トランザクションの中でPutなどをする際は、boom.Transactionに生えたPut, Deleteなどを使う必要があり、別で定義してみました。個人的には、トランザクションを作らないで同じような操作をする場合、似たような名前でメソッドを分けるようにしてます。

また、ここで注意すべきなのは、AllocateIDというメソッドで、トランザクションに入るまえにKeyを用意している点です。トランザクションが確定するまで、datastoreに保存される構造体のIDは振られず、0になります。なので、トランザクションに入る前に、APIをコールして、IDを採番します。また、この時振られるIDはAllocatedID（例: 10001001）で、goon使いがよく見るScatteredID(例: 939573199836491002)ではありません。なので、IDの見た目が少し変わりますが、実害はないと思います。

なお、このAllocateIDメソッドをわざわざ呼びたくない際には、即座にAllocateIDを呼んでくれる、AECompatibleTransaction というインターフェースもあります。とはいえpackageの最初にコメントしてあるのですが、IMPORTANT NOTICE: You should use *boom.Transaction.とあります。好みによりますが、個人的にはboom.Transactionをそのまま使って、AllocateIDは都度呼ぶようにしました。

Query, Cursor, Countなど

クエリの利用

今までFilter, Limitなどのクエリを書くときは、AppEngine公式のdatastoreパッケージにあるメソッドをそのまま呼んでいました。しかしboomではそれができません。datastore.Clientというのをboomの構造体が持っていて、それを経由して機能を呼び出します。たとえばこんな感じです。DUIDを元にUserを取得する処理です。

b, err := library.BoomFromContext(ctx)
if err != nil {
	return nil, err
}
var user = &entity.User{}
q := b.Client.NewQuery("Device").Filter("DUID = ", u.Device.DUID)

ページング

また、ページングする際はCursorというのを使いますが、これは通常文字列です。Decodeして読み込みを開始する位置を取得しますが、これもboom経由で呼び出せるDecodeCursorを使えます。

var vs []*entity.Video
g, err := library.BoomFromContext(ctx)
if err != nil {
	return nil, "", err
}
query := g.Client.NewQuery("Video").Limit(pt.Offset).Order("-CreatedAt")
if pt.Cursor != "" {
	cursor, err := g.DecodeCursor(pt.Cursor)
	if err != nil {
		return nil, "", err
	}
	query = query.Start(cursor)
}

var v entity.Video
co, err := g.Count(query)
if co == 0 || err != nil {
	return nil, "", err
}

it := g.Run(query)
for {
	_, err = it.Next(&v)
	if err != nil {
		break
	}
	v := v
	vs = append(vs, &v)
}

if err != iterator.Done {
	return nil, "", err
}

// Get the cursor for the next page of results.
nc, err := it.Cursor()
if err != nil {
	return nil, "", err
}
return vs, nc.String(), nil

また、goonではページングの最後まで到達したかをdatastore.Doneというエラーでチェックしていました。しかし、mercari/datastoreはtoWrapperErrorという関数でエラーを置き換えているようなのですが、上記のページングの終わりではiterator.Doneというエラーが返ってくるようになりました。

Count

そこまで書くこともないのですが、要素のCountはboom.Countというメソッドでそのままチェックできます。

不安点、改善点

不安点でいうと、今までScatteredIDだったのをAllocateIDにすると、IDが被らないかやや心配です。また、一番親の要素にAllocateIDでIDを割り当てても、それ以外の要素はScatteredIDで作成されるので、その不揃いに若干気持ち悪さはあります。とはいえサービス影響ないので、許容範囲かなと思います。

改善点でいうと、すぐ実装されると思いますが、まだgoonの目玉機能だったmemdへの自動キャッシュがないことです。

まとめ

以上、実際にmercari/datastoreを使ってみる場合の始め方を色々書きました。パッケージとしては構成がすごくしっかりしていて、GAEとGKE両方使ってGCPどっぷりで開発する場合はこのパッケージを使うことをオススメします。