builderscon 2018 に行ってきました

bulderscon 2018 tokyo に参加してきました。

今年は前後に CEDEC(登壇) と ISUCON(出題) があったので、発表する余裕はなさそうだなということで応募しなかったんですが、やっぱりカンファレンスは聞いているだけだと発表したくなりますね。来年は応募したい。

名札

バージョンアップ #builderscon pic.twitter.com/DgAYjpzq8v

— fujiwara (@fujiwara) September 6, 2018

結局会期中には画像転送しかできなかったので後日何とかしたいと思います。(酔って帰宅後に自宅Wifi繋ぐところでうまくいかないなと思ったら寝落ちしていた) 気象ビーコンからデータ引っ張ってきて電子ペーパーに表示したい！

聞いたトーク

Envoy internals deep dive

Envoy、まだ若いプロジェクトなのに一気に広まっててすごい。内部構造の話、いかにロックしないでパフォーマンスを出すかの話が面白かった。

Algorithms in React

React は4年ぐらい前に流行はじめたところで小さいアプリをひとつ書いてそれっきり、だったので興味深く聞きました。 同期非同期を使い分けて処理をキャンセルできるところとできないところを区別する、なるほどーというかんじ。

知らなかった、時に困るWebサービスのセキュリティ対策

セキュリティ事故、実際起こった身としては思い出したくもないということが多いんですが、それを組織としてリソースを割いて向き合って改善していこうという姿勢が真摯だなと。

発表にあったTLS太郎のような脆弱性のscan、うちは Zabbix でスクリプトを実行して TLS version とか Heartbleed の検知とかを証明書の有効期限チェックとまとめてやってたんですが、なかなか新規のを追加して維持するのが面倒なので、なにかいいサービスはないかなあ。

機械学習を用いず数学でゲーム内の需要予測をする

会社の同僚の発表。数式をできる限り抑えたところが、途中で数学愛が溢れて板書をはじめちゃったところが面白すぎた。

Understanding Microservices with Distributed Tracing

最近話題の分散トレーシングの話。自分も Perl から X-Ray を使うために AWS::XRay とか Plack::Middleware::XRay をこの前書いたので興味深いところ。(エントリ書いてない…)

あらゆる外部への通信を Envoy 経由にすることで、言語を問わず tracing するのは賢い。HTTP, gRPC みたいな通信じゃなく、たとえば MySQL や Memcached みたいな TCP の通信でどうやって trace-id を認識するのかがちょっと分からなかったので調べよう。

lld − 開発ツールの主要コンポーネントの1つをスクラッチから作成した話

似たようだけどちょっとだけ違うコードを無理に抽象化しないほうがよい、二度目は一度目よりよく書ける。長年プログラミングしているとほんとそうですね、という話なんだけど、実例が圧倒的なパフォーマンスを出しているので説得力が半端ない。

次世代通信プロトコルにおけるセキュリティ・プライバシー保護・パフォーマンス

kazuho さんの安定の素晴らしい話。いままで QUIC とか正直よく分からん、という感じだったんですが、内部構造の話が分かりやすくて最高です。

全人類に使われるプロトコルがまさに策定されている最前線で戦っているのは格好いいですね。

あなたの知らないデータベースのロギングの世界

セキュリティ事故を受けて DB へのクエリを全部ロギングしたい話。ProxySQL は知らなかったんですが、既存アプリケーションにあとからロギングのためだけに入れるにはアプリケーションへの影響が結構大きくて大変そう、という感想。

自分ならどうするか。単純な TCP Proxy を書いてそこで MySQL へ行く通信だけすべて記録して、解析はあとで何とかするとか、それこそ Envoy みたいなやつを通したらそこで capture できたりしないかな、とかいろいろ考えるところがありました。

業務時間で書いたパッチは誰のもの？ OSS 活動にまつわる罠

サイボウズさんの OSS ポリシー策定の話。カヤックも OSS は勝手にやっていいよ、というスタンスなんですが、権利関係とかいろいろ取り決めて個人にも会社にも不利益ないようにしていかないとな、と思っていたところなので大変ためになりました。参考にさせて頂きます。

Building Self-Hosted Kubernetes

kubernetes という単語が何回出てきたか分からないぐらい連呼して噛まないのがすごい。Self-Hosted は大変そうだけど、オンプレでやるならやるしかないところですね。自宅で全台落ちるとクラスタ崩壊、つらい。

1日約70万ビルド: DockerとNomadが支えるCI/CDプラットフォーム

CircleCI 2.0 でスケジューラとして Nomad を採用したという話。いつもお世話になっております。

Hashicorp プロダクト、複雑すぎず、システムとして見通しがよいところが好きです。しかしクラスタはいつか崩壊する定めなんでしょうね。

懇親会とか

前夜祭、前夜祭のあとHUB、懇親会、アフターパーティーでHUB、とよく飲みました。ひとり源泉徴収票ナイト(正確には h29syotoku.pdf ナイト)を開催してしまったのでいつか他の人のも見せてもらわないと…

水曜日のスピーカーディナーから含めると4日間の長丁場で、主催、スタッフ、スポンサー、スピーカーの皆様には大変お世話になりました。楽しいカンファレンスをありがとうございました！

ハッカーズチャンプルーに参加して Goとコンテナで作るWebアプリケーションベンチマーカー という話をしてきた #hcmpl

ハッカーズチャンプルー2018 で Go と ISUCON について話してもらえませんか? というお誘いをいただきまして、20年ぶりに沖縄に渡っております (注 まだ帰ってないので現在進行形)。

発表資料はこちらです。

speakerdeck.com

前日〜前夜祭

早めの飛行機で昼前に着いたので、瀬長島という空港から車で15分ぐらいの島へ渡って、沖縄らしいもの…ということでハンバーガーを。

ヤバい pic.twitter.com/MDSxfaSFRu

— fujiwara (@fujiwara) June 29, 2018

氾濫バーガー チムフガス というお店の「氾濫バーガー」すごい。1.5cm厚のほぼ脂身なベーコンとパティ2枚、40過ぎの人間が食べるものではないのでは…と思ったけど意外としつこくなくて完食。

寝不足の状態で沖縄の凶悪な太陽を浴びたため、その後ホテルで昼寝して、起きたら前夜祭の時間でした。

時間になっても10人ぐらいしかいなくて、なるほどこれがウチナータイム、と思いつつビールを飲んでいると、だんだん人が集まってきて LT 開始、飛び入り LT が引きも切らず、いいコミュニティだなあ…とひたすらにビールを開けてました。

カンファレンス当日

朝、台風の影響か南国らしい土砂降りの洗礼を受けつつ、スタッフの方に会場まで車で送ってもらいました。

チャンプルーの名の通り、特定の技術要素によらないトーク、LT があって飽きないですね。とくに motemen さんのいい話、あーいい話だなーって聞いてました。人に歴史あり。

懇親会でオリオンビールを無限にのみ、沖縄の若者に ISUCON を仕込んでいる さぼ (@saboyutaka) さんと話せたのもよかったですね。

台風で帰れない

既にカンファレンス中から台風の影響で日曜日の飛行機が飛ばなくて帰れなそう、という話がでしたが案の定欠航が決まり、月曜日に振り替えようとしたけど既に満席で、火曜まで帰れないことが確定…

懇親会二次会中に欠航を知り、この時点で日月の宿を確保しておくべきだったのだけど (飛行機は振り替えできた)、眠くて寝てしまって翌朝。

既に那覇のホテルがほとんどすべて埋まっていて、booking.com でひたすらリロードして空き部屋 (おそらく飛行機が飛ぶ人がキャンセルしたんでしょう) を見つけたら争奪戦を何回か挑んでやっと確保して現在に至ります。

「ブログを書くまでがハッカーズチャンプルーです」ということなのですが、帰宅できない人たちでこれから後夜祭があるようなので行ってきます。まだハッカーズチャンプルーが終わりません！

maprobe - Mackerel のホスト情報と連携する外形監視エージェントを作った

監視を Zabbix から Mackerel に移行しています。そこで困ったことを OSS を書いて解決しようシリーズのお時間です。

ホストのダウン検知を早くしたい

Mackerel の監視は Push 型と呼ばれるもので、mackerel-agent が Mackerel サーバに対してメトリクスを送信する形態です。そのため、agent を稼働させているホストがダウンしたという事象は、「一定時間サーバに対して情報が送られてこない」ことによって、組み込みのアラート "connectivity" として検知されます。

これによって困ることとしては、以下があります。

• ホストが実際にダウンしてから7分程度経過しないとアラートが上がらない
  • あまり短時間で判断してしまうと誤検知が増えるからでしょうか
  • もうちょっと早く検知したいです
• "connectivity" アラートは Critical レベルしかなく、設定変更不可
  • 大抵のホストは多重化してあるため、1台ダウンするのは大きな問題にならない
  • アラートレベルの設定指針として、「即対応しないとサービスに影響が出るものは Critical」「翌営業日に確認・対処でいいものは Warning」としているので、Critical でない事象で Critical を上げたくない

内部リソースに対する外形監視をしたい

Mackerel の監視の基本は Push 型ですが、URL外形監視機能 というものがあり、これは Mackerel のサーバから HTTP(S) で対象にアクセスして監視する機能です。

外部に公開している HTTP のリソースについてはこれで問題はないのですが、足りないところとしては以下があります。

• サービスの内部ネットワークに存在するロードバランサーなどが監視できない
• HTTP 以外のプロトコル (ICMP ping, TCP 等) での外形監視ができない

---

ということで、そのあたりを補完する外形監視エージェント、maprobe を作りました。

github.com

maprobe がやること

maprobe は次のように動作します。

1. Mackerel API を叩いてホスト情報を取得
   • Service, Role でフィルタリング
2. 各ホストに対して、probe（ping、tcp、http、command）を実行
3. 得られた結果をホストメトリックとして Mackerel に送信
4. 60秒ごとに繰り返し

組み込みで ping, tcp, http の監視機能と、Mackerel agent plugin 形式で出力するコマンドを実行する機能があります。

監視対象のアドレス等は、Go の text/template 形式で、mackerel-client-go#Host の持っている値を展開することで指定します。

設定ファイルの例

probes:
  - service: production
    role: server
    ping:
      address: '{{ .Host.IPAddresses.eth0 }}'

  - service: production
    role: InternalELB
    http:
      url: 'http://{{ .Host.CustomIdentifier }}/api/healthcheck'
      post: POST
      headers:
        Content-Type: application/json
      body: '{"hello":"world"}'
      expect_pattern: 'ok'

  - service: production
    role: redis
    tcp:
      host: '{{ .Host.IPAddress.eth0 }}'
      port: 6379
      send: "PING\n"
      expect_pattern: "PONG"
      quit: "QUIT\n"
    command:
      command: "mackerel-plugin-redis -host {{ .Host.IPAddress.eth0 }} -tempfile /tmp/redis-{{ .Host.ID }}"

このように、複数の Service / Role に所属しているホストに対して外形監視を行い、ホストメトリックを送信することができます。 (なお、チェック監視ではなくメトリック形式なのは、チェック監視のような 0/1/2 でアラートを上げる形は過去の遺産でしかなく、今後はなるべく使いたくないという思想によるものです)

たとえば ping であればこのようなグラフが得られるので、この値に対して適宜アラートを設定する、ということになります。

Critical がなく Warning だけのアラートも設定できますし、式機能を使えば特定の Role に所属するホストの 10% 以上に ping が疎通できなくなったらアラート、というような設定も可能になります。

コマンド実行の応用例

特定の Service / Role に存在するホスト(情報)に対して、任意のコマンドを実行できるため、「Mackerel から退役に失敗して残ってしまっているが、既に EC2 側にはインスタンスが存在しないホスト」の掃除をする例です。

probes:
  - service: production
    role: EC2
    command:
      command: 'cleanup.sh {{.Host.ID}} {{index .Host.Meta.Cloud.MetaData "instance-id"}}'

Mackerel のホストIDと EC2 のインスタンス ID を引数にして、既に terminate されているインスタンスなら mkr retire を実行して退役処理を行う script を実行します。

#!/bin/bash
set -u
host_id="$1"
instance_id="$2"
exec 1> /dev/null # dispose stdout
result=$(aws ec2 describe-instance-status --instance-id "${instance_id}" 2>&1)
if [[ $? == 0 ]]; then
  exit
elif [[ $result =~ "InvalidInstanceID.NotFound" ]]; then
   mkr retire --force "${host_id}"
fi

なお、このような処理は毎分実行するには負荷が大きい場合があるので、maprobe once というサブコマンドで、処理を一回だけ実行できるようにもしてあります。このような掃除だったら、1時間に1回 cron で実行する、でも十分でしょう。

コンテナ環境への対応

DockerHub にイメージを用意してあります。https://hub.docker.com/r/fujiwara/maprobe/

maprobe のように複雑な設定ファイルが必要なものをコンテナにすると、設定ファイルをコンテナ内に配置するためだけにいちいち自前のイメージを作成する必要があって面倒です。

なので、-config 引数の値はローカルファイルだけではなく、HTTP(S)と Amazon S3 の URL を処理できるようにしました。 適当なところに設定ファイルを配置し、環境変数 CONFIG にその URL を指定すれば、公式イメージをそのまま動作させられて便利ですね。

config は更新を検知して自動で再読み込みするので、maprobe agent の再起動も不要です。

どうぞご利用ください。

sardine で mackerel-plugin の出力をサービスメトリックとして投稿する

全国三千万 Mackerel ユーザーの皆様こんにちは。

mackerel-plugin で生成した値を、サービスメトリックとして投稿したいと思ったことはないでしょうか。ありますよね。でも mackerel-agent ではホストメトリックしか投稿できません。

ということで、拙作の sardine に Mackerel のサービスメトリックとして投稿する機能を追加しました。(Thanks for @jet_zousan)

sardine についてはこちらをご覧ください。

sfujiwara.hatenablog.com github.com

mackerel-plugin 互換の出力を CloudWatch のメトリックとして投稿する agent です。 おもにコンテナ環境で、mackerel-agent を全部に立てたくはないけど使い慣れた mackerel-plugin で収集した値を集約して把握することを目的としています。Go で書かれていてバイナリ1つで動作するため、コンテナに同梱して動かすのも簡単です。

以下のような設定ファイルで起動すると

[plugin.metrics.ping]
command = "mackerel-plugin-ping -count 2 -host 8.8.8.8,8.8.4.4"
service = "Home"
destination = "Mackerel"

サービスメトリックを投稿して

$ sudo MACKEREL_APIKEY="XXX" sardine -config test.conf -debug
2018/03/27 13:05:36 [plugin.servicemetrics.ping] starting
2018/03/27 13:05:38 putToMackerel: {"Service":"Home","MetricValues":[{"name":"ping.rtt.8_8_8_8","time":1522123536,"value":3.379283},{"name":"ping.rtt.8_8_4_4","time":1522123536,"value":2.739579}]}
2018/03/27 13:06:38 putToMackerel: {"Service":"Home","MetricValues":[{"name":"ping.rtt.8_8_8_8","time":1522123596,"value":2.716042},{"name":"ping.rtt.8_8_4_4","time":1522123596,"value":3.67368}]}

こんな感じにグラフができます。便利ですね。

また、interval という設定で実行間隔を制御できるため、毎分取るほどでもない値を任意の間隔で投稿することもできます。(mackerel-agent では metric plugin の実行間隔は 1分で固定になっていて、現状変更できません)

どうぞご利用ください。

ECS のデプロイツール ecspresso と、環境変数を展開して YAML/JSON/TOML を読み込む go-config について

OSS紹介 Advent Calendar 2017 - Qiita 18日目の記事です。(一週間遅れ)

Amazon ECS へのデプロイツール ecspresso と、そこで使っている環境変数を展開しつつ複数の YAML/JSON/TOML を読み込む config loader である go-config の紹介をします。

ecspresso

github.com

「エスプレッソ」と読みます。Go で書かれた Amazon ECS 用のデプロイツールです。以下の3つのファイルを用いて ECS へのサービス、タスク定義作成、入れ換えを行います。

• YAML の設定ファイル
• タスク定義のための JSON (aws ecs describe-task-definition 出力と互換)
• サービス定義のための JSON (オプション。aws ecs describe-services 出力の services セクションと互換`)

region: ap-northeast-1
cluster: default
service: app
task_definition: taskdef.json
service_definition: service.json
timeout: 10m

create (サービス作成)、deploy (新しいタスク定義を作成しサービスに対して入れ換える)、status (deployments, events をみる)、rollback (一つ前のタスク定義に差し替えてデプロイすることでロールバックする) の機能があります。

元々 aws-cli を shell script から叩いていたデプロイ script を実装し直したもので (一番最初のバージョンは Go から aws-cli をコマンド起動するものでした)、「それ ○○ (他の ECS に対応したデプロイツール)とどう違うの」といわれると答えに窮する代物ですが…

無理矢理特徴を挙げると

• Go で実装してあるのでシングルバイナリで環境依存少なく動作する
• タスクとサービスの定義ファイルには後述の go-config による、環境変数展開機能がある

ぐらいでしょうか。

go-config

github.com

こちらは、複数の YAML / JSON / TOML をマージしつつ、Go の text/template の記法で環境変数を展開して読み込むことができるパッケージです。もともと社内のとあるアプリケーションで使用するために開発されたものですが、ecspresso で使いたいがために (便利なので) 公開しました。

コンテナでアプリケーションやミドルウェアを動作させる場合、設定ファイル自体はコンテナに同梱しておくが、その中の値は環境変数で指定したい、というニーズがあります。ファイルにハードコードされていると設定値を書き換えるたびにコンテナのビルドが必要になりますが、環境変数であればコンテナの起動時に動的に指定することができるため、便利になりますね。

foo: {{ env "FOO" "default_foo" }}

このような YAML を、環境変数 FOO に bar が設定されている状態で読み込むと

foo: bar

となりますし、FOO が未設定であればデフォルト値として

foo: default_foo

として読み込まれます。

また、環境変数が設定されていなければ panic することで、読みこみ時に確実に設定されていることを強要する {{ must_env "FOO" }} という記法もあります。

やっていることは単純で、先に Go の text/template で環境変数を展開した上で YAML / JSON / TOML のパーサを通しているだけなので、ファイル自体が各フォーマットとして不正であっても展開後に正しい状態であれば、問題なく読み込むことができます。

とはいえエディタでの編集時にそれぞれのフォーマットとして正しく扱えてくれた方が楽なので、" " でクォートされた内部に展開する場合は、template の記法の方でバッククォートを用いるのがお薦めです。

{
  "foo": "{{ env `FOO` `default_foo` }}"
}

先述の ecspresso では、タスクとサービスの定義ファイル (JSON) の読みこみ時に go-config による環境変数展開が行われます。

{
  "taskDefinition": {
    "containerDefinitions": [
      {
         "image": "example.com/myapp:{{ must_env `IMAGE_TAG` }}",
         "environment": [
           {
             "name": "APP_SECRET",
             "value": "{{ must_env `APP_SECRET` }}"
           },
           {
             "name": "ENDPOINT",
             "value": "{{ env `ENDPOINT` `example.com` }}"
           }
         ]
      }
    ]
  }
}

典型的には、以下のような値を展開することを想定しています。

• デプロイごとに変わる可能性が高い値
  • Docker イメージのタグなど
• リポジトリには生で入れたくないクレデンシャルの類
  • アクセストークンなど

特にクレデンシャル類は、direnv 等で使用される .envrc というファイルに記述した上で暗号化してリポジトリに保存し、デプロイ時に復号、環境変数に設定した上でデプロイを行うと便利に扱えると思います。

# .envrc
export APP_TOKEN="raw_value_of_token"

AWS KMS で暗号化した値を .envrc.encrypted として出力、リポジトリにはこれをコミット。

$ aws kms encrypt --key-id mykey \
       --plaintext fileb://.envrc \
       --output text --query CiphertextBlob \
       --output text \
    | base64 --decode > .envrc.encrypted

デプロイ直前に KMS で .envrc を復号し、環境変数に設定してデプロイ。

$ aws kms decrypt --ciphertext-blob fileb://.envrc.encrypted \
        --output text
        --query Plaintext \
     | base64 --decode > .envrc
$ source .envrc
$ ecspresso --config app.yaml deploy

まとめ

• Amazon ECS 用のデプロイツール ecspresso を書きました
• go-config は実行時に環境変数を展開しつつ設定ファイルを読み込めて便利です

sardine - mackerel plugin のメトリクスを CloudWatch で集約する agent を書いた

OSS紹介 Advent Calendar 2017 - Qiita 22日目の記事です。

最近、監視を Zabbix から Mackerel に切り替えていっています。それと並行して、新規プロジェクトは Amazon ECS でコンテナで運用するようにもしていっています。そこで考えどころなのが、コンテナで動作するプロセスのモニタリングをどうするかです。

たとえば、コンテナで動作する nginx を mackerel-plugin-nginx で監視する場合、普通にやるとこんな感じになるのですが…

• nginx と mackerel-agent を同一タスク (ECS用語) に定義する
• mackerel-agent.conf で cloud_platform = "none" の設定をして、コンテナがホストの EC2 とは切り離された状態でホストとして認識されるようにする

すべてのタスクを Mackerel 上のホストとして認識させることになると、いくつか困ることがあります。

• 課金対象が増える
  • タスクは EC2 のホストなどよりかなり多くなるため、インパクトが大きい
  • 今後の Fargate 化を考えると、Fargate は最大4コアなので、多数のタスクを配置することになり EC2 で大きなインスタンスを動作させるよりもかなり数が増えそう
• タスクはデプロイごとに入れ替わるので、ホストの入れ替わりイベントが大量に発生する
  • 自動退役すればみなくてよいとはいえ…

複数コンテナのメトリクスは、集約した状態で一つの値がみられれば用が足りることが多いため、何らかの形で集約メトリクスを実現したかったのですが、現状の Mackerel では同一時刻に登録したサービスメトリックの値は上書きされてしまいます。

そこで、CloudWatch で集約することを考えました。

CloudWatch では同一時刻に複数回送信した値を、合計、最大、最低、平均などの統計(statistics)を指定して取得することができるため、nginx でいうと RequestCount の合計なら単位時間内の総クエスト数、平均なら1コンテナあたりの平均リクエスト数、のように、目的に合わせて値を読み取れます。

sardine

github.com

mackerel-plugin (互換) のコマンドを実行し、得たメトリクスを CloudWatch へ投稿する agent として、sardine を書きました。 Go で実装されたシンプルな agent です。

命名は、mackerel(鯖)より小さいのが群れているので sardine(鰯) ということで。

[plugin.metrics.nginx]
command = "mackerel-plugin-nginx --host nginx --port 80 --path /nginx_status"
dimensions = ["Service=staging", "Service=staging,Task=app"]

このような、mackerel-agent.conf に類似した設定ファイルで定義をします。各 ECS タスクに同梱して動作させる想定です。

デフォルトでは60秒ごとにプラグインをコマンドとして実行し、たとえば得られた出力が以下だった場合に

nginx.requests.requests 123.0 1512057958

CloudWatch のメトリクスとしては以下の値を投稿します。

• Namespace: nginx/requests
• MetricName: requests
• Value: 123.0
• Timestamp: 2017-12-01T16:05:58Z

これで複数の同一種類のコンテナ、タスクから投稿された値を、大づかみにして把握できます。

この集約されたメトリクスを、Fluentd を用いて (plugin-cloudwatch + plugin-macakrel) Mackerel に値を持っていくことで、Mackerel 上でグラフを並べてみることもできます。

まとめ

• コンテナ、タスク単位で Mackerel にホストを作らず、集約メトリクスを実現するために sardine を作りました
• Mackerel 本体だけで、このような集約メトリクスが実現できると嬉しいです

Amazon CloudSearch にドキュメントを取りこむ Lambda 関数 s32cs のご紹介

このエントリは OSS紹介 Advent Calendar 2017 - Qiita 16日目の記事です。穴が空いていたので拙作の紹介で穴埋めを。

s32cs という Amazon CloudSearch に対して S3 からのイベントドリブンでドキュメントを投入する Lambda を書いたのでご紹介します。

github.com

---

3行で

• Amazon CloudSearch にドキュメントを登録する Lambda を作った
• アプリケーションからは Fluentd にログを送るだけ
• Kinesis Firehose + S3 を利用することで、ストリームデータを 一定時間/一定サイズ に区切っての継続的なバッチ処理が容易かつ堅牢に

---

基礎知識

Amazon CloudSearch は AWS が提供している、フルマネージドな検索エンジンです。ドキュメント数が多くなったりすると勝手にスケール (アップ|アウト) してくれるので、キャパシティプランニングにそれほど気を遣うことなく全文検索を実装できます。

業務では数億(5より大きい)レコードを投入しているものもありますが、おおむね元気に動いてくれています。

CloudSearch へのデータの投入方法あれこれ

Amazon CloudSearch ドメインにデータをアップロード - Amazon CloudSearch

ここがちょっと癖があるところで、逐次レコードを登録するような、いわゆるストリーミングアップロード的なことはできず、投入するレコードが複数まとまった 5MBまでの JSON / XML ファイル (SDF) を用意してバッチ的に投げ込む API になっています。

素朴な方法

最初はものすごく素朴に、「5分ごとに DB から直近 5分のレコードを取得し、JSON に加工した上でアップロードする cron 処理」などで投入していました。

しかし、このような頻度が高い定期処理は、いろいろ面倒なことがあるのでできれば避けたいものです。

• 前回の処理が何らかの原因で終わってない場合の排他処理
  • 特に流量が多くなると、並列処理で分散させるのが面倒になります
• 失敗した場合のリカバリー
  • 前回処理が終わった最後のレコードを覚えていないといけない

また、レコードが削除されたというドキュメントも CloudSearch 側に反映しないといけないため、DBから物理削除してしまうと何らかの削除済みフラグを他に保存しておく必要があり、処理が複雑になります。

ken39arg/fluent-plugin-cloudsearch

そこで同僚の ken39arg が開発したのが fluent-plugin-cloudsearch です。Fluentd の output plugin として動作し、指定した時間/サイズごとにバッファを CloudSearch にアップロードするものです。

github.com

これによって、アプリケーションからはドキュメントの情報をログとして Fluentd に送信するだけでインデックスの更新ができるようになりました。ストリーミング処理ですね！

ただ、これもいくつか問題がありました。

• 複数台で動作させると、ある1レコードの作成と削除が別々の Fluentd のバッファに保存される可能性がある
  • バッファのフラッシュタイミングによっては、作成→削除 ではなく 削除(空振り)→作成 という順で処理され、削除されたはずのレコードが消えてくれない
• Fluentd のホストがダウンすると、バッファが失われる可能性がある
  • コンテナで動作させると、落ちた瞬間に失われるのでロストするリスクが大
• 投入した SDF はどこにも保存されずに消えてしまうので、問題があった場合のリカバリが難しい

s32cs

ということで開発したのが s32cs です。名前はまったくいいキラキラネームが思いつかなかったので、「S3 to CloudSearch」という意味です。 Go + Apex で実装されています。

S3 にオブジェクトが配置された時のトリガで Lambda を実行し、

• S3 からオブジェクトを取得
• 以下のような 1行 1レコードの JSON をアップロード用の SDF 形式に加工

{"id":"123","type":"add","fields":{"foo":"bar","bar":["A","B"]}}
{"id":"123","type":"delete"}

↓

[
  {"id":"123","type":"add","fields":{"foo":"bar","bar":["A","B"]}},
  {"id":"123","type":"delete"}
]

• CloudSearch にアップロード

という動作を行います。

S3 に JSON を配置するときに fluent-plugin-s3 で行うと、fluent-plugin-cloudsearch 同様に作成と削除が入れ替わる可能性があるため、Firehose を経由することでレコードの発生順序を(できるだけ)直列化するのがお薦めです。

1. アプリケーションは Fluentd に送信
2. Fluentd は Kinesis Firehose に逐次送信
3. Firehose は 5分 / 5MB 単位などで S3 にオブジェクトを保存
4. s32cs が S3 のオブジェクトを加工してアップロード

これによって、fluent-plugin-cloudsearch であったいくつかの問題も解消しました。

• 複数台で動作させると、ある1レコードの作成と削除が別々の Fluentd のバッファに保存される可能性がある
  • → Firehose へなるべく間隔をおかずに送信することでイベントの順序を保つ
• Fluentd のホストがダウンすると、バッファが失われる可能性がある
  • → 短時間で Firehose へ送り出されるので、ダウン時の影響が小さい
• 投入した SDF はどこにも保存されずに消えてしまうので、問題があった場合のリカバリが難しい
  • → S3 に元データが残っているので、再度別の箇所に投入したり、内容を確認することが容易

まとめ

Kinesis Firehose + S3 を利用することで、ストリームデータを 一定時間/一定サイズ に区切っての継続的なバッチ処理が Lambda で容易かつ堅牢に行えるようになります。

使いでのあるアーキテクチャパターンなので、今後も多用していきたいと思います。