Azure Functions で Domain-Driven Design (DDD) の Domain Event を実装する

この記事は Sansan Advent Calendar 2017 の 25 日目、すなわち最終日の記事です *1。ネタもないので参加する気はなかったのですが、とある社内バー的なもので酔っ払っていたところ会社の新卒君である id:kanjirz50 が目を輝かせて「最終日を飾ってください！！」と言ってきたのであまりのピュアネスが眩しくて二つ返事で快諾してしまいました。というわけでネタに困っています。

なんでそんなにネタに困っているかというと、実はここ 2 年弱程、開発者ではなくて開発マネージャーをやっていた関係で、技術ネタがないのです。この 11 月から改めて開発者になったので、今はリハビリ中でして、技術ネタは徐々に溜まってくると思います。 Advent Calendar と言っても必ずしも技術ネタである必要はなくて、 Sansan Advent Calendar 2017 では 5 日目とか 20 日目がジョブチェンジとか社内転職とかをネタにしているので、開発マネージャーの立場から開発者に復帰した経緯とか、開発マネージャーとして経験したことや学んだこととかを書くという手はあるんですが、復帰後の話を書くにはまだ少し時期尚早な事情があるので、そういうネタは今後のために取っておこうと思います。

さて、

結論

から書きますよ。 Azure Functions で Domain Event を実装する手法のいち提案は以下。

• Service Bus Topic / Subscription を使う手がある
• 現時点では Subscription から Queue への転送も使うと良さそう

はい、ツラツラと書いていきます。

Domain Event って何？

発想自体は GoF のデザインパターンにおける Observer パターンと一緒だと思ってます。 Observer パターンそれ自体はインプロセスでクラス同士の関連で実装されることが多いですが、 Domain Event ではこの表現形式が時にプロセスやネットワーク境界をまたぎます。今回この記事でテーマにしているのは、ネットワーク越しの Domain Event です。

Domain Event は、 Domain-Driven Design (DDD) 界隈では定番となっている設計手法です。あんまり頑張って説明しなくても世の中に良いドキュメントが溢れているのでそちらを参照されたいです。とだけ言うのもあんまりにも突き放し過ぎな気もするので適当にググって出てきたページへのリンクでも張ろうと思ったら、マイクロソフトの機械翻訳が笑えたのでそのページへのリンクを張ります。

ドメインのイベントです。設計と実装

話は脱線しますが最近のマイクロソフトさんはアーキテクチャ設計に関するドキュメントをドッサリ出していてすごいですね。読みきれないです。上記は Microservices に関する一連の重量級のドキュメント「マイクロサービスで DDD と CQRS パターンを使ってビジネスの複雑さに取り組む」の 1 ページです。昨今の流行りをもうこれでもかというぐらい盛り込んでいてタイトルだけでお腹いっぱい。昔は Web で読めるのはマイクロソフトテクノロジーの使い方的なドキュメントに留まっていた印象があるのですが、良い時代になりました。あとは機械翻訳さえ洗練されれば言うことなしですね。

さて Domain Event ですが、実は DDD の原著であり聖典たる Eric Evans の DDD 本ではたしか出てきていない手法で、

Domain-Driven Design: Tackling Complexity in the Heart of Software

それより 10 年も後に出た Vaughn Vernon による名著 IDDD で示されたものだった気がします。

Implementing Domain-Driven Design (English Edition)

その辺りの経緯は詳しく知りませんが、きっと Vaughn Vernon 自身が一から考えだしたものというわけでもなくて、 DDD Community の 10 年間の積み重ねのなかで、どこかで生まれたものなのではないかと想像しています。

なぜ Domain Event？

私の理解では「依存関係を逆転させるため」です。広い意味で DIP (Dependency Inversion Principle: 依存性逆転の原則)、と言ったら DIP を拡大解釈しすぎかもしれませんが、私は本質的に似通っているように感じています。以前に AWS Dev Day Tokyo 2017 というところで登壇した時の自分の資料にちょうどいい図があったので埋め込みます。

Sansanがメッセージング (SQS) でスケーラビリティを手に入れた話 from Atsushi Kambara

左側の図 (NOT Domain Event) で言うところの Operation A は次に Operation B が実行されるべきだと知っています。これは A が B に依存している状態です。 Domain Event によって A は自身の完了後に B や C が実行されるべきだという知識を持たなくて良くなります。代わりに A の責務は、自身の完了を Event Aggregator に知らせることだけになります。これで A から B への依存はなくなりました。代わりに B から Domain Event への依存が発生しますが、 Domain Event の抽象度が十分に高ければ問題ないことが多いです。

よくある例だと、企業向け SaaS における設計だとして、 A が何らかの重要なデータの更新処理、 B が管理者ユーザーへのメール送信処理、とかですね。これは DDD 的には A が Domain Layer で、 B が Application Layer で起こっていることであり、 Layer をまたぐ例です。 Domain Layer は Application Layer に依存すべきではないので、 Domain Event により依存関係を逆転させます。

今回私が Domain Event を使ったのは Layer またぎではなくて、 DDD 的用語で言えば Bounded Context をまたぐケースでした。 Bounded Context とは何ぞや、という話をし始めるとキリがないのでまた別の機会に。ちょっと諸事情により今回の記事のネタになった実例を挙げづらいので無理やり別のところから例を持ってきます。 企業向け SaaS である Sansan では、企業の名刺データを一括管理できます。ユーザーが名刺をスキャンしたら独自のテクノロジーなんかを駆使してテキストデータ化され、テナント企業向けの名刺 DB に蓄積されていくわけなんですが、もちろん名刺 DB 構築以外にも機能があります。例えば名刺がテキストデータ化されたタイミングで、当該名刺データをもとに Salesforce 上にリードや取引先責任者をつくる機能があったりします。前述の例の A と B で言えば、 A が名刺のテキストデータ化、 B が Salesforce への書き込みです。 名刺のテキストデータ化処理の最後に Salesforce への書き込み処理の起動処理があるのは、名刺のテキストデータ化文脈から見ると不純物であり、不適切な依存であるというわけです。 Bounded Context をどの程度の粒度と定義するかにもよりますが、 A と B が異なる Bounded Context に属するケースの典型例ですね。 Domain Event はこういった不適切な依存関係の解消に役立ちます。

どうやって Domain Event？

息切れしてきたので駆け足で。 Azure の PaaS を組み合わせて実装するには、 Pub/Sub モデル用に設計されている Service Bus の Topic/Subscription が使えます。 Domain Event とは Observer パターンであり、 Observer パターンとは Pub/Subモデルですから、まさにおあつらえ向きというわけです。

Topic/Subscription と Queue を準備する

こんな感じで Topic を作って、、

Subscription をぶら下げます。 Topic : Subscription = 1 : N にできるので、上の方のスライドの右側の図のように、 A の処理後に B と C がトリガーされる、というのが容易に実現できます。

それでもって Queue も作っておきます。最初の結論で書いた、 Subscription から Queue への転送における転送先用です。

そして、ここがミソなんですが、 Subscription から Queue への転送を設定します。これが 2017/12/26 現在 Azure Portal からはできないようで、 Service Bus Explorer 等から設定することになります。

残念ながら ARM Template でも設定できないそうです。 もしくは ARM Template で subscription の forwardTo プロパティとして設定します (長らく古い情報のまま放置してしまいましたが 2018-11-19 に修正)。他には Azure CLI からとかならできる感じでしょうか、調べてませんが。

コードを書く

ここまでできたら後は簡単で、 Topic にメッセージを放り込むところと Queue からメッセージを読み出すところを実装するのみです。愚直に C# で書いてみます。

まずは Topic にメッセージを放り込むところ。 Visual Studio で HTTP Trigger の Function を新規作成しただけのコードに、放り込むコードを足しただけのものです。足したのは日本語でコメントした箇所です。放り込む型は何でも良いのですが、ここでは string にしています。任意の型を JSON にシリアライズして放り込んだりすれば良いのでないでしょうか。

using System.Linq;
using System.Net;
using System.Net.Http;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.Http;
using Microsoft.Azure.WebJobs.Host;
using Microsoft.Azure.WebJobs.ServiceBus;

namespace FunctionApp2
{
    public static class Function1
    {
        [FunctionName("Function1")]
        public static async Task<HttpResponseMessage> Run(
            [HttpTrigger(AuthorizationLevel.Function, "get", "post", Route = null)]HttpRequestMessage req,
            // Topic にメッセージを放り込むための引数
            [ServiceBus("function1-completed", EntityType = EntityType.Topic, Connection = "ServiceBusConnectionString")] IAsyncCollector<string> asyncCollector,
            TraceWriter log)
        {
            log.Info("C# HTTP trigger function processed a request.");

            // parse query parameter
            string name = req.GetQueryNameValuePairs()
                .FirstOrDefault(q => string.Compare(q.Key, "name", true) == 0)
                .Value;

            // Get request body
            dynamic data = await req.Content.ReadAsAsync<object>();

            // Set name to query string or body data
            name = name ?? data?.name;

            // ここで放り込む
            await asyncCollector.AddAsync(name);

            return name == null
                ? req.CreateResponse(HttpStatusCode.BadRequest, "Please pass a name on the query string or in the request body")
                : req.CreateResponse(HttpStatusCode.OK, "Hello " + name);
        }
    }
}

そして受け取る側のコードはこんな感じ。 ServieBusTrigger のところが Subscription でなく Queue をバインドしているところがポイントです。上の方で Subscription から Queue への転送を設定しているので、ちゃんとメッセージがやってきます。

using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Host;
using Microsoft.ServiceBus.Messaging;

namespace AnotherBoundedContext
{
    public static class Subscriber1
    {
        [FunctionName("Subscriber1")]
        public static void Run(
            [ServiceBusTrigger("subscriber1", AccessRights.Listen, Connection = "ServiceBusConnectionString")]string name,
            TraceWriter log)
        {
            log.Info($"C# ServiceBus queue trigger function processed message: {name}");
        }
    }
}

なぜ Subscription から Queue に転送するか

最後に、なぜ Subscription を直接バインドせず、一手間加えて Queue への転送を設定するかです。今回この判断をした直接の理由は、 Functions と Service Bus の連携において、 2017/12/26 現在ではまだ Exponential Backoff 的なものが実装されていないことです。

例えば上の例で Subscriber1 が外部サービスの Web API 呼び出しを含むとします。そして外部サービスが (REST だとして) 500 系のレスポンスコードを返してきたとします。 500 系のレスポンスコードは一時的な (transient) エラーを表しますから、少し間を置いて同じリクエストを再度送ると、その時には成功する可能性があります。このリトライは、インメモリでループ的に実行することもできますが、せっかく Service Bus を利用しているなら、 Service Bus 自身が備えているメッセージの再実行機構に乗っかることができます。一般的にメッセージングを使う設計においては、 1 つのメッセージの処理は長くなりすぎないようにした方が良いとされていますが、ループ的にリトライ処理をするとこれに反してしまうので、今回はメッセージの再実行機構に乗っかってリトライすることにします。

ただ、何の制御もしないと場合によっては 1 回目のエラーの直後に 2 回目のメッセージ処理が走る可能性があります (Queue に他のメッセージが溜まっていない場合そうなります)。呼び出し先の外部サービスはエラーを返したのだから、さすがに少し間を置いてあげた方が良いと思われます。また、 1 回目の失敗から 2 回目の実行の間隔より 2 回目の失敗から 3 回目の実行の間隔は、長くした方が良いでしょう。このようにリトライ回数が増えるごとに実行間隔を長くする設計手法を Exponential Backoff と言います。

Functions と Service Bus を連携する場合、組み込みの Exponential Backoff は 2017/12/26 時点でサポートされておらず、実装しようとしたらおそらく、 Functions の実行自体はエラー終了ではなく成功とし、自前で ScheduledEnqueueTimeUtc を設定した BrokeredMessage を入り口と同じ Queue に放り込むようなコードを書くことになります。そういうことをしようとした時に、 Subscription を直接使っていると不都合が生じます。 Subscription を直接使っていると、リトライのために BrokeredMessage を放り込む先は当該 Subscription が結び付けられた Topic になりますが、そこに放り込んだメッセージは当該 Topic に結び付く全ての Subscription に配信されます。エラーになったのはそのうち 1 つの Subscription の処理ですから、これは不適切です。

というわけで、いったん Queue に転送することで取り回しが楽になりそうです。 Dead Letter が出た場合なんかにも、同じことが言えそうです。

まとめ

書いてるうちにとても長くなりましたが、この記事はこんなところで。使ってみると分かりますが Functions はまだまだ発展途上感があります。今後の進化に期待ですね。

*1:書いているうちに 26 日になってしまいました…