Durable Functions の Activity を安全に呼び出すためのプロキシを作った

Durable Functions が相変わらず凄く便利なので、いろんな部分で使っています。しかしヘビーに使うにつれて、アクティビティ関数を呼び出す部分で事故ったり、非常に冗長だと感じるようになってきました。

新しい Function をテンプレートから作成すると、以下のようなオーケストレータが生成されると思います。

public static class Function1
{
    [FunctionName("Function1")]
    public static async Task<List<string>> RunOrchestrator(
        [OrchestrationTrigger] DurableOrchestrationContext context)
    {
        var outputs = new List<string>();

        // Replace "hello" with the name of your Durable Activity Function.
        outputs.Add(await context.CallActivityAsync<string>("Function1_Hello", "Tokyo"));
        outputs.Add(await context.CallActivityAsync<string>("Function1_Hello", "Seattle"));
        outputs.Add(await context.CallActivityAsync<string>("Function1_Hello", "London"));

        // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
        return outputs;
    }

    [FunctionName("Function1_Hello")]
    public static string SayHello([ActivityTrigger] string name, ILogger log)
    {
        log.LogInformation($"Saying hello to {name}.");
        return $"Hello {name}!";
    }
}

単純なオーケストレータですが CallActivityAsync とアクティビティ関数名が都度出てきて、少し冗長に感じると思います。そして戻り値の型は呼び出し側で指定しないと、正しい値は返って来ません。

このくらい小規模であれば我慢できるのですが、以下のコードのように複雑なオーケストレータと多くのアクティビティで構成される場合には大変です。既に何回かパラメータの渡しミスと、戻り値の型の指定ミスを経験しています。

特にアクティビティ関数のシグネチャを変更した場合は致命的で、呼び出されている部分を全て確認して、正しい型になってるか調べる必要があります。更に関数名を nameof で指定していない場合は CodeLens も使えないので詰みます。

こういうのは本来はコンパイラや IDE にやらせるべき仕事です。なので安全にアクティビティを呼び出すために、簡単なライブラリを作りました。

考え方はインターフェースとアクティビティを用意して、それに従って動的にプロキシクラスを作成するので、プロキシ経由でアクティビティを実行するという簡単なものです。

呼び出しはインターフェースに定義されたメソッドのシグネチャを元に行われるので、正しい型で値を渡したり、戻り値の受け取りが出来るわけです。

Activity Proxy を使って書き換える

上のサンプルを Activity Proxy を使って書き換えてみます。まずはインターフェースを用意します。

public interface IHelloActivity
{
    Task<string> SayHello(string name);
}

文字列を受け取って、文字列を返すだけの単純なメソッドだけ持ちます。今のところ制約としてメソッドは 1 つのパラメータが必須かつ、戻り値の型は Task か Task<T> が必須です。

パラメータを受け取らないメソッドを書きたいところですが、今は ActivityTrigger を指定するためにはパラメータが最低 1 つ必要なので、Durable Functions 側を弄らないと対応は難しいでしょう。

上のインターフェースを元にアクティビティ関数を実装します。Task が必須なので async を使わないメソッドの場合は少し冗長に見えますが、現実的には同期で終わるアクティビティを書かないので良いでしょう。

public class HelloActivity : IHelloActivity
{
    [FunctionName(nameof(SayHello))]
    public Task<string> SayHello([ActivityTrigger] string name)
    {
        return Task.FromResult($"Hello {name}!");
    }
}

ポイントとしては nameof を使って FunctionName を指定している部分と ActivityTrigger の指定です。最初はコード生成で上手く隠したかったのですが、大掛かり過ぎたのでこの形に落ち着きました。

ここまででアクティビティは実装出来たので、後はオーケストレータから呼び出すだけです。DurableOrchestrationContext に CreateActivityProxy という拡張メソッドが増えているはずなので、インターフェースを型引数に指定して呼ぶとプロキシのインスタンスが返って来ます。

プロキシを作ってしまえば、後は普通にメソッドを呼び出す感覚でコードを書けば良いです。

public class Function1
{
    [FunctionName("Function1")]
    public async Task<List<string>> RunOrchestrator(
        [OrchestrationTrigger] DurableOrchestrationContext context)
    {
        var outputs = new List<string>();

        var proxy = context.CreateActivityProxy<IHelloActivity>();

        // Replace "hello" with the name of your Durable Activity Function.
        outputs.Add(await proxy.SayHello("Tokyo"));
        outputs.Add(await proxy.SayHello("Seattle"));
        outputs.Add(await proxy.SayHello("London"));

        // returns ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
        return outputs;
    }
}

最初のサンプルコードと比べて、かなりスッキリしたと思います。型情報を保ったまま呼び出せるので、間違った値で呼び出そうとしてもコンパイル時に発見できます。

勿論ちゃんと実行出来ます。特に面白みがない実行結果ですが、意図したとおりの挙動です。

Retry 付きで Activity を実行する

Durable Functions は組み込みでアクティビティのリトライ機能を持っているので、よくあるリトライ処理を書く必要がないので便利です。

基本的に冪等であることを意識して書く必要があるので、全体として安全にリトライが行えます。

アクティビティ関数のリトライ付き実行は CallActivityWithRetryAsync を使って行っていましたが、Activity Proxy の場合は RetryOptions 属性をインターフェースに付けて行うようにしました。

public interface IHttpGetActivity
{
    // 5 秒のウェイトを入れつつ 10 回までリトライする
    [RetryOptions("00:00:05", 10)]
    Task<string> HttpGet(string path);
}

属性を使っているので実行時に変更が出来ないですが、大抵はアクティビティ単位でリトライ設定が行えれば、固定値で困らないことが多いのでこのようにしています。

リトライ用に属性を設定しても、アクティビティの実装は同じです。制約にだけ注意が必要ですが、通常の Azure Functions なので DI を使ってコンストラクタへのインジェクションも行えます。

public class HttpGetActivity : IHttpGetActivity
{
    public HttpGetActivity(HttpClient httpClient)
    {
        _httpClient = httpClient;
    }

    private readonly HttpClient _httpClient;

    [FunctionName(nameof(HttpGet))]
    public async Task<string> HttpGet([ActivityTrigger] string path)
    {
        var response = await _httpClient.GetAsync(path);

        response.EnsureSuccessStatusCode();

        var content = await response.Content.ReadAsStringAsync();

        return content;
    }
}

この Azure Functions での DI はかなり便利なので、既に大半を DI ベースに書き換えています。

プロキシがリトライ用の設定があれば、自動的に呼び出すメソッドを変えてくれるので、オーケストレータは何も意識する必要はありません。

public class Function2
{
    [FunctionName("Function2")]
    public async Task<string> RunOrchestrator(
        [OrchestrationTrigger] DurableOrchestrationContext context)
    {
        var proxy = context.CreateActivityProxy<IHttpGetActivity>();

        // ブチザッキのタイトルを取る
        var content = await proxy.HttpGet("https://blog.azure.moe/");

        var match = Regex.Match(content, @"<title>(.+?)<\/title>");

        return match.Success ? match.Groups[1].Value : "";
    }
}

このオーケストレータを実行すると、ちゃんとブチザッキのタイトルが返ってきます。ネットワークエラーが発生した場合でも、自動でリトライを行ってくれます。

大したことのないサンプルではありますが、どこかのタイミングで Let's Encrypt 更新くんを全面的に Activity Proxy を使って書き換えようと思っています。

パラメータ無しのメソッドを使えるようにしたいところなので、解決したらプレリリースを外します。