Amazon EKSでgRPCサーバを運用する

以前の記事でも紹介した通り、一休では、gRPCを使ったサービスを導入し始めています。

user-first.ikyu.co.jp

この記事では、このサービスをAmazon EKSで提供するための設計や気をつけたポイントについて紹介します。

背景

一休では、ウェブアプリケーションの実行環境としてAWS Elastic Beanstalkを採用しています。 そして、この4月からElastic BeanstalkをAmazon EKSへ移行するプロジェクトを進めています。 このgRPCサービスもElastic Beanstalkで運用をしていましたが、以下の問題を抱えていました。

• 適切にロードバランシングできない。
  • Elastic BeanstalkでgRPCサービスを運用しようとするとNetwork Load Balancer(NLB)を使うことになります。NLBはレイヤ4のロードバランサです。一方で、gRPCはhttp/2で動作し、ひとつのコネクションを使いまわします。このため、NLBでは、特定のサーバに負荷が偏ってしまう場合があります。

また、EC2サーバのdockerdのcpu利用率が突然上昇する、アプリケーションのデプロイに時間がかかるというような問題も起き始めていたため、Amazon EKSへ移行することでこれらの問題の解決を試みました。

KubernetesでgRPCサーバを動かす場合の注意点

Amazon EKSは、AWSが提供するマネージドなKubernetesです。KubernetesでgRPCサーバ動かす場合の注意点は上述のNLBの場合と同様、以下の点になります。

• KubernetesのServiceはL4のロードバランサであり適切に負荷分散できない。

このため、KubernetesのServiceではなく別の仕組みを使って負荷分散を実現する必要あります。

Envoyを導入する

多くの事例でEnvoyを使ってこの問題を解決しています。

www.envoyproxy.io

AWSのApplication Load Balancer(ALB)がhttp/2に対応しているのはクライアントとALB間の通信でありALBからバックエンドへの通信はhttp/2に対応していない、だから、gRPCサーバの場合はNLBを使う必要がある。そしてNLBを使うと上述の問題にぶつかる、、、ということでしたが、Envoyは、公式サイトに書かれている通り、クライアントとEnvoyの間もEnvoyとバックエンドサーバの間も両方ともHTTP/2とgRPCをサポートします。一休でもEnvoyを使うことに決めました。

構成

GitHub - GoogleCloudPlatform/grpc-gke-nlb-tutorial: gRPC load-balancing on GKE using Envoy

このGKEでの事例を参考にしつつ、以下のような配置構成にしました。

Kubernetesクラスタ外部からのアクセスを受ける必要があるため、type: LoadBalancer のService経由でEnvoyが外からのアクセスを受けます。EnvoyのバックエンドはHeadless Serviceにします。Headless Serviceを使うと背後にあるPodに直接アクセスできるレコードがクラスタ内部DNSに作成されます。EnvoyはこのDNSレコードからPodのIPを取得し、Envoy側の構成にしたがって負荷分散を行います。

Envoyの設定

Envoyの構成情報はKubernetesのComfigMapで設定します。内容な以下の通り。

apiVersion: v1
kind: ConfigMap
metadata:
  name: grpc-service-proxy
data:
  envoy.yaml: |
    static_resources:
      listeners:
      - address:
          socket_address:
            address: 0.0.0.0
            port_value: 50051
        filter_chains:
        - filters:
          - name: envoy.http_connection_manager
            config:
              access_log:
              - name: envoy.file_access_log
                config:
                  path: "/dev/stdout"
              codec_type: AUTO
              stat_prefix: ingress_http
              route_config:
                name: local_route
                virtual_hosts:
                - name: https
                  domains:
                  - "*"
                  routes:
                  - match:
                      prefix: "/"
                    route:
                      cluster: grpc-service
              http_filters:
              - name: envoy.health_check
                config:
                  pass_through_mode: false
                  headers:
                  - name: ":path"
                    exact_match: "/healthz"
                  - name: "x-envoy-livenessprobe"
                    exact_match: "healthz"
              - name: envoy.router
                config: {}
      clusters:
      - name: grpc-service
        connect_timeout: 0.5s
        type: STRICT_DNS
        dns_lookup_family: V4_ONLY
        lb_policy: ROUND_ROBIN
        drain_connections_on_host_removal: true
        http2_protocol_options: {}
        hosts:
        - socket_address:
            address: grpc-service.default.svc.cluster.local
            port_value: 50051
        health_checks:
          timeout: 3s
          interval: 5s
          unhealthy_threshold: 2
          healthy_threshold: 2
          grpc_health_check: {
            service_name: Check
          }
    admin:
      access_log_path: /dev/null
      address:
        socket_address:
          address: 127.0.0.1
          port_value: 8001

注意点としては、以下の通りです。

• drain_connections_on_host_removal: true を設定すること。
  • この設定がない場合、バックエンド側のServiceからPodが外れても、そのPodのヘルスチェックが失敗するまで、Envoyは外れたPodにトラフィックを流し続けます。その結果、デプロイ時にリクエストを落としてしまう可能性が発生します。
• grpc_health_check を指定すること。
  • バックエンド側がgRPCサービスなのでこれを設定します。

gRPCサービス側の構成

gRPCサービスはElastic BeanstalkでもDockerを使って動かしていました。したがって、EKSに動かすために大きな変更は必要ありませんでした。唯一必要だったのがヘルスチェックです。 上述の通りNLBではgRPCサービスのヘルスチェックをすることができません。NLBのヘルスチェックはTCPのポートの疎通確認ですが、厳密にいえばこれはアプリケーションの死活監視になりません。一方、Kubernetesではヘルスチェック( livenessprobeとreadinessprobe)は重要な役割を果たしますので、設定する必要があります。 gRPCサービスに対するヘルスチェックについては以下の記事に方法が書いてあります。

Health checking gRPC servers on Kubernetes - Kubernetes

この記事に従って、以下の2点を行いました。

• gRPCサービスにヘルスチェックのエンドポイントを実装する。
• Dockerイメージ内にgrpc-health-probeを同梱する。

そして、gRPCサービスのPodの定義を以下のようにしました。

...略...
spec:
      containers:
        - name: grpc-service
          image: ["ImageName"]
          imagePullPolicy: IfNotPresent
          ports:
            - containerPort: 50051
          readinessProbe:
            exec:
              command: ["/bin/grpc_health_probe", "-addr=:50051","-service=Check"]
            initialDelaySeconds: 5
          livenessProbe:
            exec:
              command: ["/bin/grpc_health_probe", "-addr=:50051","-service=Check"]
            initialDelaySeconds: 10

...略...

リクエストを落とさずにデプロイをする

ここまで構築してリクエストを流してみると、ひとつのコネクションで多数のリクエストを送信しても、きちんと複数のPodにトラフィックが分散されることがわかりました。

あとは、EnvoyもgRPCサービスもリクエストを落とさずにデプロイできるように設定する必要があります。

Envoyの場合

Envoyの場合は、以下のissueがとても参考になりました。

Graceful HTTP Connection Draining During Shutdown? · Issue #7841 · envoyproxy/envoy · GitHub

このissueによれば、/healthcheck/fail というエンドポイントにPOSTリクエストをすることでEnvoyはそれ以後受け付けたリクエストの接続をクローズするようです。これを活用して、EnvoyのPodの定義のlifecycleに以下の通り書けばリクエストを落とさずにPodの入れ替えができそうです。

...略...
lifecycle:
            preStop:
              exec:
                command:
                  - /bin/sh
                  - '-c'
                  - >-
                    wget -qO- --post-data=''
                    http://127.0.0.1:8001/healthcheck/fail && sleep 10
...略...

preStopフックで wgetを使ってhttp://127.0.0.1:8001/healthcheck/failにPOSTリクエストを行い、その後10秒待っています。 127.0.0.1:8001 には Envoyの管理用のAPIが立ち上がっています。 この設定によって、Podが停止する流れは大まかに以下のようになります。

1. PodがTerminating 状態になる。同時にpreStopフックが動き、healthcheck/fail にPOSTリクエストがされたあと、10秒待つ。
2. Podには、リクエストは来るが、レスポンスを返すときにコネクションは閉じられる。
3. 10秒待っている間に、ServiceからPodが外れる。以後、新しいリクエストは来なくなる。接続中のリクエストも10秒間ですベて処理される、はず。
4. 10秒停止後、SIGTERMで、コンテナが停止される。すでに接続がない状態なので、リクエストは落とさない。

この設定をして、トラフィックを流しつつ、Envoyのデプロイを行ってみたところリクエストを落とさずにPodが入れ替わることが確認できました。

gRPCサーバの場合

注意点は以下のふたつです。

• 上述の通り、Envoyの構成で drain_connections_on_host_removal: true の設定をする。
  • この設定をしないとEnvoyはヘルスチェックが通り続ける限り、古いReplicaSetに属するPodにもトラフィックを流し続けます。
• アプリケーションはSIGTERMを受け取ったら、グレースフルにシャットダウンする処理を実行する。
  • 今回のサービスはGoのgRPCサービスなので、GracefulStop()を呼び出すようにしています。

このふたつをやっておけば、ローリングアップデートでもBlue/greenでもリクエストを落とさずにPodの入れ替えができることを確認できました。

まとめ

今回は、gRPCサービスのEKSでの動かした方を紹介しました。 一休では、現在、主要サービスのAmazon EKSへの移行を進めています。この移行の全体像については、別途、紹介するつもりです。

参考資料

文中に紹介したリンク以外にも以下のサイトを参考にさせて頂きました。

Kubernetes上でgRPCサービスを動かす | SOTA

kubernetesでgRPCするときにenvoy挟んでみたよ - Qiita

Kubernetes: 詳解 Pods の終了 - Qiita

spinnaker/solutions/bluegreen at master · spinnaker/spinnaker · GitHub

Manage Traffic Using Kubernetes Manifests - Spinnaker

---

この記事の筆者について

• システム本部CTO室所属の 徳武 です。
• サービスの技術基盤の開発運用、開発支援、SREを行なっています。

go-mssqldbでタイムゾーンが常にUTCになる

こんにちは。宿泊事業本部の宇都宮です。

一休では、基幹データベースにSQL Serverを使用しています。また、Goアプリケーションでは、go-mssqldbというライブラリを使用して、データベースとのやりとりを行っています。

このgo-mssqldbには、タイムゾーンに関して厄介な挙動があります。タイトルにもあるように、タイムゾーンが常にUTCになってしまうのです。本記事では、go-mssqldbのタイムゾーン関係の振る舞いと、go-mssqldbを使いつつ正しくタイムゾーンを扱うための対処法を紹介します。

go-mssqldbのタイムゾーン問題

go-mssqldbのタイムゾーン問題は、以下のコードで再現できます。

package main

import (
    "database/sql"
    "fmt"
    "log"
    "os"
    "time"

    // blank import必須
    _ "github.com/denisenkom/go-mssqldb"
)

func main() {
    conn, err := sql.Open("sqlserver", "sqlserver://user:password@dbhost:1433?ApplicationIntent=ReadWrite&MultiSubnetFailover=Yes&database=DB")
    if err != nil {
        fmt.Fprint(os.Stderr, err)
        os.Exit(1)
    }

    row := conn.QueryRow(`
select
    GETDATE()
`)

    var dbnow time.Time
    if err := row.Scan(&dbnow); err != nil {
        fmt.Fprint(os.Stderr, err)
        os.Exit(1)
    }

    now := time.Now()
    after1Hour := now.Add(1 * time.Hour)

    fmt.Println(after1Hour.After(now))
    fmt.Println(after1Hour.After(dbnow))

    fmt.Println(now)
    fmt.Println(after1Hour)
    fmt.Println(dbnow)
}

dbnowにはDBの現在時刻から取得した時刻が、nowにはサーバの現在時刻が入ります。after1Hourはサーバの現在時刻から1時間後です。したがって、 after1Hour.After(now) と after1Hour.After(dbnow) は、DBとサーバの時計が1時間以上ずれていない限り、trueになるはずです。

しかし、DBとサーバの両方のタイムゾーンがJST(+09:00)の状態でこのコードを実行すると、 after1Hour.After(now) はtrue、after1Hour.After(dbnow)はfalseになります。

原因は、このコードの出力を見ると分かります。

dbnow は日時は同じ（2019-08-23 15:54）ですが、タイムゾーンがUTCになっています。UTCでJSTと表面上の日時が同じと言うことは、実質9時間後の日時ということです。そのため、after1Hour.After(dbnow)はfalseになってしまうのです。

タイムゾーン問題の対処方法

go用の mysqlドライバ のように、Data Source NameにDBサーバのロケーションを指定できる機能があればよいのですが、go-mssqldbにはそのような機能はありません。

この問題を解決しようとしているissueやプルリクエストはいくつか見つかりますが、近いうちに取り込まれそう、というステータスではありません。

そこで、ライブラリ側の解決を待つのではなく、利用者の側で対処する必要があります。

最も根本的な対応は、datetimeoffset型の使用です。datetimeoffset型はタイムゾーンも保持しているため、日時の保存にこの型を使っておけば、go-mssqldbを使っていても問題なくタイムゾーンを扱えます。

すでにdatetime型で日時を保存していて、datetimeoffset型での保存が難しい場合は、TODATETIMEOFFSET関数を使ってデータを取得する際に型変換しましょう。

   row := conn.QueryRow(`
select
    GETDATE()
  , TODATETIMEOFFSET(GETDATE(), '+09:00')
`)

    var dbnow time.Time
    var dbnowoffset time.Time
    if err := row.Scan(&dbnow, &dbnowoffset); err != nil {
        fmt.Fprint(os.Stderr, err)
        os.Exit(1)
    }

    now := time.Now()
    after1Hour := now.Add(1 * time.Hour)

    fmt.Println(after1Hour.After(now))
    fmt.Println(after1Hour.After(dbnow))
    fmt.Println(after1Hour.After(dbnowoffset))

    fmt.Println(now)
    fmt.Println(after1Hour)
    fmt.Println(dbnow)
    fmt.Println(dbnowoffset)

dbnowoffset は+0900(JST)になっているため、after1Hour.After(dbnowoffset) はtrueになります。意図したとおりの動作といえます。

C#のgRPCクライアントではChannelを再利用しよう

こんにちは。宿泊事業本部の宇都宮です。6月に、Go + gRPCという構成のサービスを運用開始したという記事を書きました。

Go + gRPCによるマイクロサービス構築 - 一休.com Developers Blog

本番運用開始から2ヶ月ほどたち、いくつかのトラブルがありつつ、現在も元気に稼働中です。

運用していく中で定常的に発生していたgRPCのタイムアウトエラーについて、その対処法がわかったので、紹介します。

なお、本記事の知見はC#でのgRPCクライアント実装においては有用でしたが、他の言語では適用できない可能性が高いです。各言語のドキュメントもあわせてご参照ください。

gRPCのタイムアウト

以前の記事で紹介したマイクロサービスは、30～60req/sec程度のリクエストを常時受け付けます。

社内のサービスの中で、このマイクロサービスに最も頻繁にリクエストを送るのは認証基盤で、全リクエストの9割以上を占めます。なお、認証基盤はC#(ASP.NET MVC)で実装されています。

この認証基盤のログをみると、1時間に5件程度、gRPCの「Deadline Exceeded」エラーが発生していることがわかりました。これは通信タイムアウト時に発生するエラーです。

認証基盤では、当初、gRPCのタイムアウトを5秒に設定していました。これは大幅に余裕を持たせた数字で、実際、99%のリクエストは0.2秒以内にレスポンスを受け取っていました。また、gRPCサーバでも処理時間のログを取っていますが、こちらには1秒以上レスポンスにかかっているようなログは出ていませんでした。

つまり、gRPCのタイムアウトはレスポンスの遅延に起因するものではなく、クライアント-サーバ間のコネクションの確立失敗に起因するのでは？　と仮説を立てました。

対応(1) タイムアウトの短縮とリトライ

当初、タイムアウトを短くして、代わりにリトライを行うようにすればいいのでは？　と考えました。

C#でPollyを使うと、こんな感じのコードになります：

using Polly;

...

var client = new MemberServiceClient() { DeadLineSeconds = 1 };
var response = Policy.Handle<Exception>()
    .Retry(3)
    .Execute(() => client.GetMember(memberSeq));

この実装では、タイムアウトは 改善しませんでした。リトライは確かに行われるのですが、一度失敗した状態からリトライしても、失敗し続けてしまい、何回リトライしても同じ、という状態でした。

対応(2) Channelの再利用

C#向けのgRPCライブラリのドキュメントはここにあります。中でも特に重要なのは、サーバとの接続を管理する Grpc.Core.Channel クラスのドキュメントでしょう：

https://grpc.github.io/grpc/csharp/api/Grpc.Core.Channel.html

このドキュメントの冒頭には、以下のように書いてあります。

Channels are an abstraction of long-lived connections to remote servers. More client objects can reuse the same channel. Creating a channel is an expensive operation compared to invoking a remote call so in general you should reuse a single channel for as many calls as possible.

要するに、Channelはコネクションプールなので、インスタンスを使い回した方がいいよ、ということです。

もとの実装では、gRPCクライアントをnewするたびに新しいChannelを作っていました。この実装を改め、Channelは一度作成した後は、static変数に保持してサーバのライフタイムに渡って使い回すようにしました。

using Grpc.Core;

...

// もとの実装(ClientのDisposeでChannelをShutdown)
using(var client = new MemberServiceClient(new Channel(GetServerHost(), GetServerPort(), ChannelCredentials.Insecure)))
{
  //
}

// Channelを使い回すようにした実装(Application_EndでChannelをShutdown)
var client = new MemberServiceClient(GetChannel());

...

private static Channel _channel { get; set; }
private static object channelLock = new object();

public static Channel GetChannel()
{
    if (_channel == null)
    {
        lock (channelLock)
        {
            _channel = new Channel(GetServerHost(), GetServerPort(), ChannelCredentials.Insecure);
        }
    }

    return _channel;
}

この変更をしたところ、認証基盤のgRPCと通信する部分のスループットが2倍になり、さらに、gRPCのタイムアウトもほとんど発生しなくなりました。タイムアウトが発生した場合でも、Channelを頻繁に再作成していた場合と異なり、リトライすれば成功するようになりました。

むすび

gRPCは、ハイパフォーマンスなサービスを作るための強力なスタックですが、ややクセがあります。特に、コネクション周りは実装（言語）依存の部分が大きいため、クライアントを書く言語向けのドキュメントには一通り目を通した方がいいな、と思いました。

社内マーケター向けの機械学習プラットフォームを作りました

はじめに

こんにちは。データサイエンス部の平田です。

一休でのデータ分析はJupyter NotebookやJupyter Labを用いてDWHにアクセスして行われることが多いですが、サービスそのものと分析環境が乖離していることにより、分析結果を継続的にサービスに取り込むのが難しい状況でした。 また、マーケティング部の方々がJupyterを使用して分析した結果に基づいて継続的に施策を行おうとしても、Airflowに組み込む際のエンジニアの負担はそこそこありますし、修正するたびに依頼をしなければならないなどコミュニケーションコストも発生します。 さらに、マーケティングに機械学習を取り入れたい場合でもairflow側で全部やってしまうと密結合になってしまいます。

そこで、Airflowから別の場所にあるJupyterを直接実行することによりエンジニアの負担は最小限にとどめ、自由に施策を打てるような仕組みとして機械学習プラットフォーム、通称ml-jupyterが生まれました。

模式図

AirflowからJupyterをキックする

①まず、日次で実行しているAirflowからml-jupyter上のAPIをキックする関数を作ります。

キックするコードは以下のようになります。 大まかな流れとしては、カーネル起動→コードを取得→ウェブソケットで通信→カーネル終了となります。

Jupyter notebookでは一つのnoteごとにカーネルが割り当てられるため、まずその起動から始まります。

import json
import requests
import datetime
import uuid
from work.dags.base_taskset import BaseTaskset
from websocket import create_connection

class TriggerJupyterNotebook(BaseTaskset):

    @classmethod
    def execute(cls, **kwargs):
        ipynb_file = kwargs['ipynb_file']
        folder_name = kwargs['folder_name']

        # 施策ごとのノートブックファイル (Airflowから実行されるもの)
        notebook_path = f'/{folder_name}/{ipynb_file}.ipynb' 
        host = [host_name]
        base = f'http://{host}'

        # カーネルを起動
        url = base + '/api/kernels'
        response = requests.post(url)
        kernel = json.loads(response.text)
        print('kernel_id:', kernel['id'])

        # コードを取得
        url = base + '/api/contents' + notebook_path
        response = requests.get(url)
        file = json.loads(response.text)

        # セルのコードのみを抽出
        code = [c['source'] for c in file['content']['cells'] if len(c['source']) > 0 and c['cell_type'] == 'code']

        # WebSocketのオープン
        ws = create_connection(
            f'ws://{host}/api/kernels/' + kernel['id'] + '/channels'
        )

        # WebSocket上でメッセージを送る

        # カレントディレクトリを施策用のディレクトリに変更
        # (パッケージ等もカレントディレクトリを参照する)
        # 最後にカーネルの処理が完了したことを知るために、特定の文字列を出力する
        terminated_signal_str = uuid.uuid1().hex
        code = ['import os', f"os.chdir('/tf/{folder_name}')"] \
        + code + ["print('" + terminated_signal_str + "', end='')"]

        for c in code:
            msg_type = 'execute_request'

            content = {
                'code': c,
                'silent': False
            }

            hdr = {
                'msg_id': uuid.uuid1().hex,
                'username': 'airflow',
                'session': uuid.uuid1().hex,
                'data': datetime.datetime.now().isoformat(),
                'msg_type': msg_type,
                'version': '5.0'
            }

            ws.send(json.dumps({
                'header': hdr,
                'parent_header': hdr,
                'metadata': {},
                'content': content
            }))

        # WebSocketのレスポンスを取得
        error_flag = False
        while True:
            msg_type = ''
            while msg_type != "stream":
                rsp = json.loads(ws.recv())
                msg_type = rsp['msg_type']
                if rsp['content'].get('status') == 'error':
                    print('jupyter notebook error:', rsp['content']['evalue'])
                    error_flag = True
                    break

            # エラーを返却した場合、WebSocketをクローズして、処理を終了
            if error_flag:
                ws.close()
                break

            # 特定の文字列を含む場合、WebSocketをクローズして、処理を終了
            if terminated_signal_str == rsp['content']['text']:
                ws.close()
                break

        # カーネルを終了
        url = base + f"/api/kernels/{kernel['id']}"
        response = requests.delete(url)
        response.status_code  # 204ならOK

        return not error_flag

②PythonOperatorを作り、この関数を実行します。

PythonOperator(
    task_id='[task_id]',
    provide_context=True,
    python_callable=TriggerJupyterNotebook.execute,
    dag=subdag,
    op_kwargs={'folder_name': '[folder_name]', 'ipynb_file': '[file_name]'}
)

③新たな施策をAirflow上で組み込むときは、エンジニア側は「Jupyter上の〇〇ファイルを実行してほしい」 「〇〇が終わった後や〇〇テーブルが更新された後に実行してほしい」などの要望を聞き、上のPythonOperatorをコピペして少し変え、適切な順番で実行されるように実装するだけになります。

ただ、Airflowから無事キックできるようにするにはいくつか決めなければならないことがあります。

ライブラリ競合問題

施策ごとにライブラリのバージョンが異なる可能性がある場合、それぞれその施策独自のライブラリを見に行く必要があります。 Jupyter上で、

!pip install [hogehoge] -t .

とすることで、カレントディレクトリにライブラリがインストールされるようになります。

施策ごとにフォルダを切り、上記を実行することでそれぞれのフォルダ内で独立してライブラリをインストールできます。 Airflowからキックするコードにchdirコマンドを追加しています。（第1項）

code = ['import os', f"os.chdir('/tf/{folder_name}')"] \
+ code + ["print('" + terminated_signal_str + "', end='')"]

さらに、pandas, boto3など共通としてよさそうなライブラリは別途インストールしています。

終了検知問題

Jupyterでの実行終了を検知するのは結構難しい問題です。若干泥臭くなりますが、Airflowでランダムに生成した特定文字列をJupyter側で最後に出力させることで終了とみなすようにしました。もっといい方法があれば教えてほしいです。（第3項）

code = ['import os', f"os.chdir('/tf/{folder_name}')"] \
+ code + ["print('" + terminated_signal_str + "', end='')"]

また、Jupyter自体がエラーで終了することもあり得るのでフォローしています。

ml-jupyter上からできること

サービスAPIとマーケターをつなぐ

基本的に一休のCRM施策はDWHのデータを元に、SQLで対象者データやコンテンツ情報を抽出し、Ikyu Marketing Cloudという社内ツールから各種チャネルに配信しています。

ただし、DWHのデータは基幹DBの当日2:00までのデータであり、リアルタイムに基幹DBから情報を取得・追加・更新することができません。 上記のDB分離問題からマーケティング側の課題として、特定のユーザーに対して何らかのアクション(クーポンの付与や会員ランクの変更)をフレキシブルに行えないという問題がありました。

そこで、ml-jupyter上から基幹側の各種APIを呼べるようにライブラリを整備することで、各Notebookのインスタンス上でpythonを用いてAPIをコールするなど自由自在にクーポン付与等の施策を行えるようになりました。

機械学習のアプローチを用いたデータ生成

今まではMarketing Cloud上で配信する対象者等のデータ抽出については、SQLを用いて複雑なクエリやストアドプロシージャを組んで行っていましたが、SQL自体の言語仕様が貧弱なところもあり、機械学習のアプローチを用いてデータの作成をするのが困難でした。

ml-jupyterではJupyter Notebook上で、pandasやscikit-learn等の強力なpythonライブラリを用いた機械学習のアプローチで各種データの生成ができるようになり、より一歩進んだCRM施策が行えるようになりました。

宿泊のリマーケティングクーポンの施策は機械学習のアプローチにてデータが作られ、Marketing Cloud上で日々配信が行われており、現状かなりの数値のリフトが見られています。

細かい話

マーケティングの方々はおそらくバージョン管理に興味ありません。（偏見）

そこで、jupyter notebookをcronでこっそり自動pushしています。

5年間の改善を経て、現在の一休がどうなっているのかを7/4(木)にお話します

レストラン事業本部の田中( id:kentana20 )です。

先週末にDevLOVE Xというイベントで開発組織改善の取り組みについて5年間の取り組みと今後、というテーマでお話しました。

5年間でどれくらい一休の開発組織が変わったのか

• 技術面
• 組織面

それぞれで実施した改善について、改善の裏側で起こっていたことや自分の所感も含めてお話しました。

現在の一休について7/4(木)にお話します

来週開催するエンジニア向けの説明会で、上記の5年間の取り組みを経て、現在の一休がどうなっているのかをCTOの伊藤がお話します。

ikyu.connpass.com

説明会と書いていますが、会社・事業や開発体制のお話だけでなく

• 現在の開発組織やプロダクトの状況がどうなっているか
• 今後どのように改善を進めていくのか

を含めてお話する予定です。

一休で働くことに興味がある方だけでなく

• いまの現場を改善するためのヒントがほしい
• 改善を進めている一休のエンジニアと直接お話したい

などなど、どなたでもご参加いただけます。

また、現場のエンジニアと個別にお話する時間も設けていますので、ご興味のある方はぜひご参加ください！

みなさんとお会いできることを楽しみにしています。

hrmos.co

Go + gRPCによるマイクロサービス構築

こんにちは。宿泊事業本部の宇都宮です。

最近、とあるマイクロサービスをローンチしました。このアプリケーションの業務的な役割は諸事情により省略しますが、以下のような特性をもっています。

• 社内の多くのサービスから利用される
  • 一休.com
  • 一休.comレストラン
  • 一休.comギフト
  • 一休.com海外
• このサービスが落ちると、主要サービスの予約処理が止まる 😱

想定されるリクエスト数は、平常時で30req/sec、ピーク時には60req/sec程度になります。行う処理はシンプルで、DBにいくつかSELECT文を投げて、ビジネスロジックに沿った結果を返すことです。

また、基盤系のアプリケーションなので、各開発者の開発環境（WindowsとMacが混在）でも動作する必要があります。

したがって、このアプリケーションに求められる要件は、

• 高パフォーマンス
• 高信頼性
• クロスプラットフォームで動作すること

などになります。

このアプリケーションを構築するにあたって採用した技術と、開発の際に気をつけたこと、運用開始してから直したことなどを紹介します。

採用した技術

Go

開発言語としてはGoを採用しました。

一休のサーバサイドで広く使われている言語/ランタイムは以下の通りです。

• VB, C#/.NET Framework: 一休.com等
• C#/.NET Core: メール配信基盤等
• Python: 一休.comレストランの新アーキテクチャ、一休.comスパ
• VBScript: 一休.comレストランの旧アーキテクチャ
• JavaScript/Node.js: 一休.comレストランのBFF
• Go: 行動ログ収集API

これらのうち、Go以外の言語が選外になった理由は以下の通りです。

• VBScript, .NET Framework: 新機能の追加が止まっており、将来性がない
• Python, Node.js: 開発環境にランタイムのインストールが必要なのが微妙（バイナリを配布するだけにしたい）
• .NET Core: 要件は全て満たしているが、開発者の採用面が…

.NET CoreとGoはどちらも要件を満たしているので、.NET Frameworkの経験を活かしやすい.NET Coreか、採用の強そうなGoかは迷いました。が、As-isではなくTo-beで考えた方がいいかなということで、Goを選定しました。

gRPC

データのやりとりにはgRPCを採用しました。今回のアプリケーションの場合、RESTでも十分だと思っていましたが、より高速に動作しそう、という理由でgRPCを採用しました。

先行して開発されていた別のアプリケーションでもgRPCを採用していて、そちらで技術検証が済んでいた、という点もポイントです。

ただ、一休ではVBScript等レガシーな技術も使っているので、どうしてもgRPCが使えなくなる事態もありうると思っていました。なので、REST APIも並行して開発しました。なお、結果的には、gRPCで問題なかったです。

余談ですが、現在、一休の一部サービスでは、VBScript(Classic ASP)からgRPCを叩いています。C#(.NET Framework)でgRPCクライアントを開発し、このdllをCOM経由で使っています。

擬似コードですが、大体こんな感じです:

Dim client
Dim response

Set client = Server.CreateObject("Ikyu.GrpcServiceClient")
Set response = client.GetMember(memberSeq)

開発の際に気をつけたこと

プレゼンテーションとドメインの分離

前述したとおり、gRPCでいくか、RESTでいくか、迷っていました。gRPCかREST(JSON over HTTP)かは、結局プレゼンテーションの問題なので、プレゼンテーションとドメインをきっちり分けて、プレゼンテーションを簡単に切り換えられるように気をつけました。

擬似コードですが、gRPCのrpcは以下のような実装になっています。

func (s *server) GetMember(ctx context.Context, in *pb.GetMemberRequest) (*pb.Member, error) {
    member, err := memberService.FindByMemberSeq(in.MemberSeq)
    if err != nil {
        // エラー処理
    }
    return &pb.Member{
        MemberSeq: member.MemberSeq,
    }, nil
}

これに対して、RESTのハンドラ（MVCのコントローラー相当）は以下のような実装になっています（フレームワークはgin）。

func (h *MemberHandler) GET(c *gin.Context) {
    member, err := memberService.FindByMemberSeq(c.Param("member-seq"))
    if err != nil {
        // エラー処理
    }
    c.JSON(http.StatusOK, gin.H{
        "memberSeq":      member.MemberSeq,
    })
}

このように、ほとんどの業務処理はドメイン層で行い、プレゼンテーション層はドメインサービスから戻ってきた処理結果を表現することのみを行うようにしました。これによって、gRPCとRESTを容易に併存可能にし、開発終盤まで決断を遅らせられるようにしました。

ユニットテストとインテグレーションテスト

高信頼性のためには、コーナーケースも含めたテストが必要です。そこで、ドメイン層についてはユニットテストを徹底しました。

また、DBと通信するコードはインフラストラクチャー層として分離し、こちらについては実際のDB(SQL Server)をDockerに立ててテストを行うインテグレーションテストを行いました。高速な動作のために、ビジネスロジックをSQLのクエリで表現している部分もあるため、インテグレーションテストもある程度しっかりやっています。

Goは標準でテストのための仕組みが組み込まれているので、テストが書きやすい言語だと思います。

今後の運用を考えると、Consumer-Driven Contractsのようなテストも用意した方がいいのかな、と検討中です。

運用開始してから直したこと

sql.DB を頻繁にオープン・クローズしない

Goのイディオムとして、取得したリソースは defer Close() するというものがあります（Close() のエラーハンドリングが必要な場合を除く）。そのため、最初はDBと通信するコードを以下のように書いていました。

func (s *server) GetMember(ctx context.Context, in *pb.GetMemberRequest) (*pb.Member, error) {
    db, err := sql.Open("sqlserver", dsn)
    if err != nil {
        // エラー処理
    }
    defer db.Close()

    memberService := NewMemberService(db)
    ... 
}

RPCの始めでDB接続をオープンして、終わりでクローズしています。これでもそれなりの性能は出ていたのですが、ドキュメントを確認すると、気になる記述がありました。

The returned DB is safe for concurrent use by multiple goroutines and maintains its own pool of idle connections. Thus, the Open function should be called just once. It is rarely necessary to close a DB.

https://golang.org/pkg/database/sql/#Open

sql.Open() の結果として取得できる sql.DB 構造体は、一度Openした後はプログラム中でずっと使い回すべきで、Closeを呼ばなければならないことはまれである、とされています。

sql.DB はコネクションの実体ではなく、コネクションプールなので、サーバの起動時に一度だけOpenし、その後は使い回すのがよさそうです。

func main() {
    db, err := sql.Open("sqlserver", dsn)
    if err != nil {
        // エラー処理
    }
    // gRPCサーバの実体となる server 構造体にdbフィールドを持たせる
    s := grpc.NewServer()
    pb.RegisterMemberServiceServer(s, &server{db: db})
}

// RPCのハンドラでは server.db を使う
func (s *server) GetMember(ctx context.Context, in *pb.GetMemberRequest) (*pb.Member, error) {
    memberService := NewMemberService(s.db)
    ... 
}

このように変更した結果、低負荷時のレスポンスタイムが改善しました（10～20ms => 5ms）。

あわせて、コネクションプール周りの設定も行いました。こちらは、以下の記事を参考にさせていただきました。

DSAS開発者の部屋:Re: Configuring sql.DB for Better Performance

db.SetMaxOpenConns(50)
db.SetMaxIdleConns(50)                   // MaxOpenConns 以上の値にすること
db.SetConnMaxLifetime(100 * time.Second) // 最大接続数 × 秒数 が目安

コネクションプールの設定によって、高負荷に対する耐性も上がりました。負荷試験を行ったところ、当初の実装では300req/sec程度の負荷をかけるとタイムアウトが頻発するようになっていましたが、sql.DBの使い回しとコネクションプールの設定チューニングによって、300req/secでも正常に結果を返すことができるようになりました。

むすび

今回はGo + gRPCというスタックでアプリケーションを開発した際の知見をいくつか紹介しました。一休では、引き続き、Go言語によるWebアプリケーション開発を進めていきたいと考えています。

doda.jp

API Test ライブラリ Tavern のご紹介

こんにちは。 一休.comの開発基盤を担当しています、akasakasです。

今回は、Tavern という API Test ライブラリ を紹介したいと思います。

一休でAPI Test が必要になった背景

前回のブログでも少し触れましたが、APIのテストを無理やりSeleniumを使ってテストを続けた結果、E2Eが破綻しました。

適切なレイヤーで適切なテストをしようということで、APIに関してはちゃんと API Test ライブラリ の導入を決めました。

user-first.ikyu.co.jp

API Test を導入する上で考えたポイント

API Test を導入する上で考えたポイントは以下の点です。

• 開発者フレンドリー
• CI連携

開発者フレンドリー

前回のブログでも書きましたが、一休ではQA・テストエンジニアのようなポジションはいないので、開発者がテストも修正するようになってます。

開発者が気軽にテストを追加・修正することができるような API Test ライブラリが必要だと考えました。

CI連携

一休で扱っているAPIは外部の提携先に提供しているAPIが多いです。

もしリリース後、APIで障害が発生した場合、外部の提携先にも影響が及びます。

リリース前の障害事前検知のため、検証環境で定期的にAPI Testを流して、リリース後の障害を防ぎたいです。

その為には、API Test を CIで回せるようにする必要がありました。

Tavernのご紹介

上記の要件を満たせるAPI Test ライブラリを検討した結果、Tavernと出会いました。

Tavern はAPI Test に特化した PyTest のプラグインです。

テストはYAMLで記述できるので、シンプルでわかりやすく、メンテナンスが簡単ですので、開発者フレンドリーであると感じました。

また、PyTestのプラグインでコマンドラインでテストの実行ができるので、CI連携もしやすいです。

API Testのツールでいうと、PostmanやInsomniaがありますが、GUIとして使うのが一般的で、CI連携が難しかったです（PostmanのCLIでNewmanというのがありますが、テストの記述がTavernよりも難しいと感じたので、Newmanの導入も見送りました）。

Tavern のいいところ

Tavern のいいところとしては以下の3点があると思います。

• YAMLでテストを記述できる
• 前のテストの結果を保存できる・次のテストに使える
• CI連携

YAML でテストを記述できる

Github API を例にして Tavern を使って、API Test をしてみましょう。

Tavern で issue を作れることを確認します。

https://developer.github.com/v3/issues/#create-an-issue

先にテストからお見せします

test_name: Github API Test

includes:
  - !include common.yaml

stages:
  - name: Create Issue
    request:
      url: https://api.github.com/repos/{service.owner:s}/{service.repo:s}/issues
      headers:
        Authorization: "token {service.token:s}"
      method: POST
      json:
        title: "Issue From Tavern Test"
        body: "一休.comをご利用頂き、誠にありがとうございます。"
    response:
      status_code: 201
      body:
        title: "{tavern.request_vars.json.title}"
        body: "{tavern.request_vars.json.body}"
        state: "open"

YAMLで書かれているので、どんなことをしているのか、なんのテストをしているのかがイメージしやすいと思います。

• request
  • url でテスト対象のエンドポイントを指定
  • header で Authorization Header を設定
  • jsonでポストするデータを指定 etc
• reponse
  • 201 が返却されること
  • body の内容がrequestの内容と一致していること

などを記述していることがわかります。

共通のデータは common.yaml で記述しています。

common.yaml の内容としては以下のようなイメージです。

description: used for github api testing
name: test includes
variables:
  service:
    token: "token"
    owner: "リポジトリオーナー"
    repo: "リポジトリ"

テストの実行は

 tavern-ci test_github.tavern.yaml

で確認できます。

Tavern のお作法として、テストファイルのyamlは test_*.tavern.yaml という形で統一しているようです。

テストも通りました。

$ tavern-ci test_github.tavern.yaml
================================================== test session starts ==================================================
platform darwin -- Python 3.6.1, pytest-4.3.0, py-1.8.0, pluggy-0.9.0
plugins: tavern-0.22.1
collected 1 item                                                                                                        

test_github.tavern.yaml .                                                                                         [100%]

=============================================== 1 passed in 1.66 seconds ================================================

Tavern で Github API を使って、issueを作成することができました。

前のテストの結果を保存できる・次のテストに使える

Tavern の便利なところとして、前のテストの結果を保存して、それを次のテストに使うことができるというのもあります。

Github API を使って

• Issue を作成
• Issue の番号を取得
• その番号のissueを編集する

ということをやってみましょう。

先にテストからお見せします

test_name: Github API Test

includes:
  - !include common.yaml

stages:
  - name: Create Issue
    request:
      url: https://api.github.com/repos/{service.owner:s}/{service.repo:s}/issues
      headers:
        Authorization: "token {service.token:s}"
      method: POST
      json:
        title: "Issue From Tavern Test"
        body: "一休.comをご利用頂き、誠にありがとうございます。"
    response:
      status_code: 201
      save:
        body:
          issue_id: number

  - name: Edit Issue
    request:
      url: https://api.github.com/repos/{service.owner:s}/{service.repo:s}/issues/{issue_id}
      headers:
        Authorization: "token {service.token:s}"
      method: PATCH
      json:
        title: "Edit Issue From Tavern"
        body: "一休レストランをご利用頂き、誠にありがとうございます。"
    response:
      status_code: 200
      body:
        state: "open"

• Create issue のresponse body で issue_id を save
• Edit Issue の request url で issue_id を使って、テスト

ということをやっています。

Tavern で Github API を使って、issueの作成と編集をすることができました。

CI連携

tavern のインストール自体は、 pip install tavern のみで済みますし、

テストの実行自体も tavern-ci test_*.tavern.yaml で済むので、CI連携も容易です。

Tavern の実運用にまつわる細かい話

Tavern の実運用にまつわる細かい話としては

• 環境別のテストデータの設定

が気になるところなのかなと思います。

これに関しては、

https://tavern.readthedocs.io/en/latest/basics.html?#multiple-global-configuration-files

に書いてありますが、各環境別で使うYAML を用意します。

まとめ

今回は、 Tavern を使ったAPI Test について紹介しました。

「API Test ライブラリといえば、コレだ！」というものがない印象でしたが、

• テストを YAML で記述できて、イメージしやすい
• CI連携の容易

という点で、 Tavern は選択肢の一つとしてアリなのかなと思いました。

参考

github.com

taverntesting.github.io

tavern.readthedocs.io

E2EテストをSelenium Webdriver からCypress.io に移行した話

こんにちは。 一休.comの開発基盤を担当しています、akasakasです。

今回は、E2EテストをSelenium WebdriverからCypress.ioに移行した話をしたいと思います。

一休のE2Eテスト事情

一休では staging/production へのリリース完了をフックにして、主要導線に対してE2Eテストを実施しています。

これを実施している主な理由としては

• 検証環境での障害の事前検知
• リリース後も正常に予約ができるかどうかの確認
  • ECサイトで予約を止めるのは致命的なので、これを防ぐ

があります。

詳しくはこちらのスライドに書いてあるので、興味のある方はみてください。

speakerdeck.com

あれから、数年が経過して、、、

完全に動かなくなりました。悲しいです。

どうしてこうなった？？？

理由としては

• SelniumではSPAへの対応が難しくなってきた
• なんでもかんでもSeleniumに任せようとした弊害

がありました

SeleniumではSPAへの対応が難しくなってきた

一休ではSPA化が徐々に進んできています。

具体的な取り組みについては下記のエントリで紹介しているので、ご興味があれば、ご覧ください。

user-first.ikyu.co.jp

user-first.ikyu.co.jp

Selnieum Webderiver は画面遷移をしていくMPAに対して、効果を発揮するブラウザテストツールであり、

• 非同期リクエストや動的な画面の書き換え
• 画面遷移が発生しない

SPAでSelenium Webdriverを使って安定したテストを継続していくのが困難でした。

Wait処理などを上手く使えば、不可能ではないですが、一休ではQA・テストエンジニアのようなポジションはいなく、開発者がテストも修正するようになってます。

開発者にテストを書く負担を減らして、サービス開発に集中して欲しいというのも思いとしてありました。

なんでもかんでもSeleniumで頑張ろうとした弊害

一休.com ではUTが充実していないためか、「なんでもかんでもSeleniumでテストしよう」みたいな雰囲気がありました。

具体的には、APIの疎通確認をしたいが為に、SwaggerUIのようなテスト用の画面を作成し、その画面をSeleniumを使って、APIの疎通確認を行っていました。

APIのテストをわざわざブラウザテストをする必要はないです。 ただでさえ、ブラウザテストは不安定で時間がかかるので、適切なレイヤーで適切なテストができていないというアンチパターンに陥っていました。

いざリプレイスへ・リプレイスをする上で気をつけたこと

上記の理由からSeleniumから別のブラウザテストツールの移行を検討しました

単純なツールの乗り換えだけだと、同じ過ちを繰り返す恐れがあったので、下記の点を注意しました。

• 開発者フレンドリー
• 安定性
• 然るべきレイヤーでテストする（何でもかんでもブラウザテストにしない）

開発者フレンドリー

Selneiumの課題として、セットアップが面倒というのがありました。

開発者にテストへの時間を軽減して、サービス開発に集中して欲しいというのも思いがあったので、下記の点を重視しました。

• セットアップの敷居が低いこと
• 開発者が容易にテストを作ることができる

安定性

言わずもがなですが、「移行したはいいが、テストが落ちまくっている」というのは有り得ないので、

• SPAでも安定してテストが動く

ということにフォーカスしました

然るべきレイヤーでテストする（何でもかんでもブラウザテストにしない）

前述でも書きましたが、

• APIのテストを無理やりSeleniumで書いていた

というのが、テストの安定性を損ねていた原因の一つでした。

この問題に関しては、APIテストライブラリを導入して、ブラウザテストとは切り分けました。

APIテストライブラリに導入については後日、どこかで書きたいと思います。

技術選定

ブラウザテストでSeleniumからどのツールを選ぼうかを考えた際に、以下の3つが選択肢としてありました

• WebdriverIO
• Puppeteer
• Cypress.io

どの技術を採用するかで重要視したポイントが「開発者フレンドリー」であるかです。

具体的には

• セットアップ
• 書きやすさ

の2点です。

セットアップという点だと、Puppeteer・Cypress.ioがいい印象でした。

書きやすさで見た場合、Cypress.ioの方がテストを書くことに集中できると思ったので、Cypress.ioを採用することに決めました。

Cypress.io とは？

JavaScript製のブラウザテストに特化したE2Eテストフレームワークです。

Seleniumはテストを書くこと以外にもスクレイピング等の用途で使うことができますが、 Cypress.ioはテストを書くことに特化したE2Eテストツールです。

Cypress.io のいいところ

Cypress.io の特徴は色々あると思いますが、個人的に感じるところとしては、次の3点が大きいと思います。

• セットアップが楽
• テストを書くことだけに集中できる
• CI連携が楽

セットアップが楽

Cypress.io はセットアップが非常に簡単です。

npm install cypress

これだけで終わりです。

SeleniumだとGeckodriverやChromedriverをインストールしたり、パス設定したりと、 少し手間がかかるので、セットアップの敷居が低いという点で、非常にありがたいです。

テストを書くことだけに集中できる

SeleniumやPuppeteerを選ぶと、

• テストランナーどれを選ぼう
• レポーティングはどれにしよう
• アサートのライブラリはどれにしよう

などといったところも考えると思います。

Cypress.io はオールインワンでサポートしているので、テストを書くことだけに集中することができます。

https://www.cypress.io/how-it-works/ で紹介されている、下記の図のようなイメージです。

CI連携が楽

CI連携が楽という点も個人的にはありがたかったです。

• DockerImageが用意されている
• 各CI Provider に対して、 example project が用意されていて、わかりやすい

こちらに詳細が書かれているので、興味のある方はご覧ください。

https://docs.cypress.io/guides/guides/continuous-integration.html

Cypress.io の頑張って欲しいところ

Cypress.io に対する不満はそんなにありませんが、あえて1点だけ挙げるなら

• クロスブラウザ未対応

という点です。

一休で、E2Eテストを実施している目的は

• 主要導線が正常に動くことを確認すること

なので、クロスブラウザで確認する必要性はないです（確認するに越したことはありませんが）

Cypress でもOpen Issue として挙げられているので、今後クロスブラウザ対応がされる日が来るかもしれません（いつになるのかはわかりませんが）

Proposal: Support for Cross Browser Testing · Issue #310 · cypress-io/cypress · GitHub

その他、移行に関しての細かい話

あと、移行に関する細かい話としては以下の3つがあります

• 重複テストケースの排除
• Page Object Design Patternで設計
• 移行に乗じて、CIもJenkinsからCircleCIに変更

重複テストケースの排除

既存のテストケースを見直すと、同じようなことをテストしている部分があったので、 移行の際にテストケースを精査して、必要最低限のテストケースを実施するようにしました。

Page Object Design Patternで設計

既存のSeleiumでもPage Object Design Pattern を採用しましたが、 画面変更に対して強い設計方法なので、ここは変えませんでした。

移行に乗じて、CIもJenkinsからCircleCIに変更

以前はCIのためにオンプレサーバのJenkinsを用意していましたが、Jenkins起因でE2Eテストが失敗することもしばしばありました。 テストの安定性を考えた場合、CIも乗り換えた方がいいと感じていたので、このタイミングでCircleCIで実行するように変更しました。

そして、E2Eは復活し、平和はおとずれた

かくかくしかじかありまして、E2Eテストが復活しました。めでたしめでたし。

まとめ

今回は、E2EテストをSelenium WebdriverからCypress.ioに移行した話をしました。

Seleniumがよくないとか、Cypress.ioがいいという話ではなく、 一休のサービス開発が進んでいった結果、SeleniumによるE2Eテストが難しくなり、今回Cypress.ioへの移行をしました。

Cypress.io の利点としては、上述でも書いた通り

• 開発者フレンドリー

であることだと感じます。

一休のようにQAやテストエンジニアがいなく、開発者がE2Eテストを修正するようなワークフローになっている開発現場ではCypress.ioを採用するのは選択肢の一つとして、ありなのかなと思います。

しかし、正直な話、この仕組みも数年後には破綻するかもしれません。 その時はまたサクッと捨てて、その時に一休のサービス・開発現場にマッチする新しい仕組みに乗り換えればいいと思います。 そういうことができるようにブラウザレベルのテストを極力書かないほうがいいのかもしれません。

また、この数年間、一休.com を守ってくれたSeleniumには感謝と敬意を払いながら、未来のために新システムに切り替えていきたいです。