Azure Functions の Durable Functions の node 版がリリースされたので遊んでみました

Durable Functions は、個人的に Azure で一番好きな機能なのですが、それが node でも使えるようになりました。 これまでもプレビューであったけど、今回は正式版ということで実践投入行ける感じですね。

これまで、Durable Functions を使おうと思ったら C# でしたが、これからは node でも OK。個人的には C# の方がなじむけど、好きな方が使える状態なのは大事ですね。

使ってみよう

では行ってみましょう。適当なフォルダで func init か VS Code で Create project して JavaScript を選びます。npm init -y もしておきます。そして npm i durable-functions でライブラリを入れておきます。

Durable Functions の拡張機能もコマンドで有効にしておきましょう。

func extensions install -p Microsoft.Azure.WebJobs.Extensions.DurableTask -v 1.7.0

続けて関数を作っていきます。Durable Functions を作るときは、オーケストレーターを起動するための関数（HTTPやQueueなど）と、色んな処理を呼び出したりフローを管理するオーケストレーター関数と、実際の処理を行うアクティビティ関数の最低 3 つをよく作ります。アクティビティ関数は、やりたい処理の数だけできるので実際にはもっと沢山の関数を定義します。

Durable Functions のいいところは、これらの沢山の関数をサーバーレスアーキテクチャーらしい呼び出し方をしてくれてるのに、書き味は普通のプログラムと変わらないというところが素晴らしいところです。 例えば、処理を 3 つシーケンシャルに呼ぶときには、サーバーレスとかだと Queue を間に挟んで 1 つ 1 つを細かくわけてやることが多いです。まぁ本当に小さい処理なら以下のように

funcA();
funcB();
funcC();

とすればいいのですが、そこそこ大きな処理になってくると長い実行時間だと色々困ることがあるサーバーレスのプラットフォーム上で動かすときは

Client(処理のお願いを QueueA に投げる) -> QueueA -> funcA(QueueA をトリガーに処理をして結果を QueueB に投げる) -> QueueB -> funcB(QueueB をトリガーに処理をして結果を QueueC に投げる) -> QueueC -> funcC(QueueC をトリガーに処理をして結果を QueueD に投げる) -> QueueD -> Client(QueueD を監視して結果を受け取る)

ツライ。この辛さを解消してくれて、さらに有り余るメリットを与えてくれるのが Durable Functions だと思ってます。

脱線しましたが、気を取り直して関数を作成します。まずはトリガーとなる関数ですね。今回は HttpTrigger で作成します。 func new か VS Code で New Function を選んで作ります。

function.json に追記して Durable Functions の機能を使えるようにします。

{
  "disabled": false,
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "name": "starter",
      "type": "orchestrationClient",
      "direction": "in"
    }
  ]
}

type が orchestrationClient の定義がそれにあたります。index.js ではオーケストレーターを呼び出す処理を書きます。

const df = require('durable-functions');
module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = await client.startNew("OrchestrationFunction", undefined, req.body);
    return client.createCheckStatusResponse(req, instanceId);
};

戻り値は、状態を確認したりするための情報をクライアントに返すようにていします。後で使いますが、これも便利。OrchestrationFunction には HTTP のリクエストのボディを入力として渡しています。

では、オーケストレーターの関数を作りましょう。HttpTrigger の関数を OrchesrationFunction という名前で作ります。そして function.json を以下のように編集します。

{
  "disabled": false,
  "bindings": [
    {
      "name": "context",
      "type": "orchestrationTrigger",
      "direction": "in"
    }
  ]
}

そして、処理を書いていきます。このオーケストレーターの関数は、実は何回も呼び出されて裏で実行履歴と突き合わせをして動く特殊な関数なので、ちょっと書くときにお作法がいります。要は何回動かしても同じ結果になるようなものしか使えません。（乱数とかはダメ、そういうのがしたい場合は Durable Functions が提供してる代替関数を使う）

const df = require('durable-functions');
module.exports = df.orchestrator(function* (context) {
    context.df.setCustomStatus({ message: 'OrchestrationFunction started'});
    const output = [];
    output.push(yield context.df.callActivity('SayHello', context.df.getInput().first));
    context.df.setCustomStatus({ message: 'first activity is completed'});
    output.push(yield context.df.callActivity('SayHello', context.df.getInput().second));
    context.df.setCustomStatus({ message: 'second activity is completed'});
    output.push(yield context.df.callActivity('SayHello', context.df.getInput().third));
    context.df.setCustomStatus({ message: `third activity is completed, waiting accept event: ${JSON.stringify(output)}`});
    const accept = context.df.waitForExternalEvent('accept');
    const reject = context.df.waitForExternalEvent('reject');

    const event = yield context.df.Task.any([accept, reject]);
    if (event === accept) {
        return output;
    } else {
        return [];
    }
});

さて、見慣れない関数の連続ですが、やってることはこの関数に渡された入力の first, second, third を、SayHello というアクティビティに渡して結果を output に追加していっています。

途中経過を連絡するためにカスタムステータスを適時設定していってます。

最後に、外部からの accept か reject という名前のイベントを待って accept されたら実行結果を返して、reject されたら空の配列を返しています。

最後に SayHello 関数を作ります。これも HttpTrigger で作ったあとに function.json を書き換えます。（そのうち func new とかにテンプレートが追加されると思いますが今はない）

{
  "disabled": false,
  "bindings": [
    {
      "name": "name",
      "type": "activityTrigger",
      "direction": "in"
    }
  ]
}

index.js は以下のような感じ。

function sleep10s() {
    return new Promise(resolve => setTimeout(resolve, 10 * 1000));
}

module.exports = async function (context, name) {
    await sleep10s();
    return `Hello ${name}`;
};

10 秒待った後に Hello xxxx という文字を返します。

ということで、まとめると30秒くらい処理に時間がかかって、さらに外部からのイベントが来たら結果を返すという感じですね。SayHello は 1 つ 1 つが 10 秒なので、どれくらいの時間がかかるか読めますが外部からのイベントは Durable Functions が提供してくれる REST API を叩かないと発行出来ないので、どれくらい時間がかかるのかさっぱりわかりません。

そんな処理も Durable Functions でさくっと書けるのが素敵。

実行してみよう

local.settings.json で Azure Storage の接続文字列を入れます。macOS や Linux では公式では Azure にストレージアカウントを作って、その接続文字列を入れてって書かれてます。Windows の場合はエミュレーターでもいいので UseDevelopmentStorage=true でも OK です。

あと、Azure の Storage Emulator のバージョンによっては AzureWebJobsSecretStorageType を追加しないといけないみたいです。

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "node",
    "AzureWebJobsSecretStorageType": "files"
  }
}

Issue はこちら。

github.com

このブログから行きつきました。感謝。

qiita.com

さ・ら・に。ローカルで動かすときは以下の issue もあります。

github.com

ということで以下のように WEBSITE_HOSTNAME 環境変数に localhost:port番号 を追加します。デフォルトだとポート番号は 7071 なので以下のような感じになると思います。

では、func start で実行して HttpTrigger を叩きます。

実行結果に、statusQueryGetUri というのがあります。これを叩くと関数の状態をチェックできます。

関数実行直後に叩いた様子

数十秒後に叩いた様子

カスタムステータスがイベントを待ってるようなので、イベントを送りたいと思います。このときも最初に HttpTrigger を叩いた結果に sendEventPostUri というのがあります。 今回の場合はこんな URL です。

http://localhost:7071/runtime/webhooks/durabletask/instances/98fbec63c5124414ae363518bb7d27bc/raiseEvent/{eventName}?taskHub=DurableFunctionsHub&connection=Storage&code=mF4BzJbHO9xX51OE6JK76feSO8A4zsUewXm53e9lSUxo5LCmwaYe6w==

{eventName} というところに今回は accept か reject を指定して POST でボディが空の JSON のリクエストを送ると OK です。

こんな感じ。

この API を叩いたあとに結果を取得するとこの通りステータスが Completed になって output に結果が入っています。

ソース

書きながら作ったソースは以下に置いてます。

github.com