Azure DurableFuncitons v3 → v4 (isolated .NET7)

前回の記事の続きです。

v3 WebJobs DurableFunction を v4 isolated Durable Function に移行する記事です。

基本的な移行操作は前回の記事を参考にします。
追加で必要な作業だけこちらで書きます。

Microsoft Docsを参考にすればいいと思いますが、なんか不親切だったので自己流にまとめなおしますよ

作業概要

1.　通常の v3 → v4 バージョンアップを実施
2.　DurableFunctions に必要なパッケージ参照を追加
3.　attribute や、型の書き換え(地獄)

ざっと上記に記したような手順になるかと思います。
3つ目の項がこの記事のメインです。基本は機械的な作業になっていて、「こう書かれていたらこう書き換える」みたいな感じです。

作業詳細

概要に挙げたことを詳しく書いていきます。

1. 通常の v3 → v4 バージョンアップを実施

こちらは前回の記事を参考にします。
大まかに以下の作業を行います。

1. プロジェクト(*.csproj)の構成を、v4 .NET7用に書き換え
   • ※Microsoft.Azure.Functions.Worker.Extensions.○○○を含めないとダメ
2. local.setting.json を isolated 用に書き換え
3. Startup.cs があれば、Program.cs に移植・変換
4. 引数でlogger をもらっていたら、DIしてもらうように変更
   足りなければプロジェクトごとに適切な変更を行って下さい。

これだけだと全然ビルドも通らないし、Intellisense に怒られて赤い下線で埋め尽くされていると思います。

2. DurableFunctions に必要なパッケージ参照を追加

新しい方のパッケージを読み込むようにします。

<ItemGroup>
  <PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.0.0" />
</ItemGroup>

バージョンは各自書き換えてください。
NuGetパッケージマネージャからMicrosoft.Azure.Functions.Worker.Extensions.DurableTaskの最新版を検索してインストールするのがいいかと思います。

3. attribute や、型の書き換え(地獄)

ここまで来たら、存在する function と Orchestrator , Activity をゴリゴリ書き換えていきます。

Microsoft Docsにも書き換え表がありますが、こっちにも書いておきます。

上記のドキュメントの通りにバージョンアップを実施しようとすると、どうにもおかしいかもしれないので、こちらのDocsを参考にして書き換えます。

以下の手順で書き換えてゆきます。
a. Durable Orchestration Client (Starter)
b. Orchestrator
c. Activity
基本的にこの3項目があると思うので、順番に書き換えていきます。
内容は以下のように統一します。

• v3 のサンプルプログラム
• v4 に変換したプログラム
• 書き換え表

---

a. Durable Orchestration Client (Starter)

v3

using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class SampleStarter
    {
        [FunctionName(nameof(SampleStarter))]
        public static async Task Run(
            [QueueTrigger("queue")]string payload,
            [DurableClient] IDurableClient client,
            ILogger log)
        {
            var instanceId = await client.StartNewAsync(nameof(SampleOrchestrator), payload);
            log.LogInformation($"Started orchestration. payload => {payload}");
        }
    }
}

v4

using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;
using Microsoft.DurableTask.Client;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class SampleStarter
    {
        private readonly ILogger _logger;

        public SampleStarter(ILoggerFactory loggerFactory)
        {
            _logger = loggerFactory.CreateLogger<SampleStarter>();
        }

        [Function(nameof(SampleStarter))]
        public async Task Run(
            [QueueTrigger("queue")]string payload,
            [DurableClient] DurableTaskClient client)
        {
            var instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(SampleOrchestrator), payload);
            _logger.LogInformation($"Started orchestration. payload => {payload}");
        }
    }
}

DurableFunciton v3 → v4 isolated 書き換え表

v3	v4	備考
IDurableClient	DurableTaskClient	function task の書き換え
StartNewAsync	ScheduleNewOrchestrationInstanceAsync	DurableTaskClientのメソッドを書き換え

---

b. Orchestrator

v3

using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class SampleOrchestrator
    {
        private readonly RetryOptions _retryOptions;

        public SampleOrchestrator(IOptionsMonitor<DurableRetryOption> durableRetryOption)
        {
            var firstRetoryInterval = 60;
            var maxNumberOfAttempts = 30;
            var maxRetryIntervalMinute = 10;

            _retryOptions = new RetryOptions(TimeSpan.FromSeconds(firstRetoryInterval), maxNumberOfAttempts)
            {
                MaxRetryInterval = TimeSpan.FromMinutes(maxRetryIntervalMinute)
            };
        }

        [FunctionName(nameof(SampleOrchestrator))]
        public async Task<bool> RunOrchestrator(
            [OrchestrationTrigger] IDurableOrchestrationContext context
            , ILogger log)
        {
            var payload = context.GetInput<SamplePayload>();
            if (context.IsReplaying == false)
                log.LogInformation($"Executing SampleOrchestrator {payload}");

            // なにかする
            var result1 = await context.CallActivityWithRetryAsync<Sample1Result>(nameof(Sample1Activity), _retryOptions, payload);
            if (result1.Succeed == false)
            {
                _logger.LogError($"activity1 failed");
                return false;
            }

            // 1こめのactivityの結果からさらに何かする
            var result2 = await context.CallActivityWithRetryAsync<Sample2Result>(nameof(Sample2Activity), _retryOptions, result1.hogehoge);
            if (result2.Succeed == false)
            {
                _logger.LogError($"activity2 failed");
                return false;
            }
	    
	    .......
	    
            return true;
        }
    }
}

v4

using DurableTask.Core;
using Microsoft.Azure.Functions.Worker;
using Microsoft.DurableTask;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class SampleOrchestrator
    {
        private readonly ILogger _logger;
        private readonly TaskOptions _retryOptions;

        public SampleOrchestrator(ILoggerFactory loggerFactory)
        {
            _logger = loggerFactory.CreateLogger<SampleOrchestrator>();
            
            var firstRetoryInterval = 60;
            var maxNumberOfAttempts = 30;
            var maxRetryIntervalMinute = 10;

            var retryPolicy = new RetryPolicy(
                maxNumberOfAttempts: maxNumberOfAttempts,
                firstRetryInterval: TimeSpan.FromSeconds(firstRetoryInterval),
                maxRetryInterval: TimeSpan.FromMinutes(maxRetryIntervalMinute)
                );

            _retryOptions = new TaskOptions(retryPolicy);
        }

        [Function(nameof(SampleOrchestrator))]
        public async Task<bool> RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, SamplePayload payload)
        {
            if (context.IsReplaying == false)
                _logger.LogInformation($"Executing SampleOrchestrator {payload}");

            // なにかする
            var result1 = await context.CallActivityAsync<Sample1Result>(nameof(Sample1Activity), payload, _retryOptions);
            if (result1.Succeed == false)
            {
                _logger.LogError($"activity1 failed");
                return false;
            }

            // 1こめのactivityの結果からさらに何かする
            var result2 = await context.CallActivityAsync<Sample2Result>(nameof(Sample2Activity), result1.hogehoge, _retryOptions);
            if (result2.Succeed == false)
            {
                _logger.LogError($"activity2 failed");
                return false;
            }

            .......

            return true;
        }
    }
}

※ なんでかわからないけど、TaskOrchestrationContext.GetInput<T>()を呼び出すと、そこで急にOrchestraorがExecutedになります。
なので、代2引数で受け取るようにします。なんで落ちちゃうんでしょうか・・・？

CallActivityAsync でリトライポリシーとペイロードを同時に渡す方法が記事で見つからず、ずっと悩んでましたが、TaskOptions のコンストラクタで Microsoft.DurableTask.RetryPolicy を渡したものを第三引数に渡せばできそうでした。

CallActivityAsync に RetryPolicy と input payload を両方渡す方法

var retryPolicy = new RetryPolicy(
    maxNumberOfAttempts: 60,
    firstRetryInterval: TimeSpan.FromSeconds(30),
    maxRetryInterval: TimeSpan.FromMinutes(10)
    );

var retryOptions = new TaskOptions(retryPolicy);

var result = await context.CallActivityAsync<Sample1Result>(nameof(Sample1Activity), payload, retryOptions);

DurableFunciton v3 → v4 isolated 書き換え表

v3	v4	備考
IDurableOrchestrationContext	TaskOrchestrationContext	Function の引数の型を書き換え
CallActivityWithRetryAsync	CallActivityAsync	引数の順番も変わっているので注意
RetryOptions	TaskOptions	詳しい書き方は上記のCallActivityAsync に RetryPolicy と input payload を両方渡す方法を参照
context.GetInput<T>();	Orchestrator の引数にバインドできるようになった

SubOrchestrator

v3 の頃は、Orchestrator内で、別のOrchestratorを呼び出すときに、CallActivityすればよかったのですが、v4 からは、CallSubOrchestratorAsync というメソッドが用意されているので、そちらから呼ぶようにしましょう。
ちなみに、CallActivityで呼ぶと、何もログを吐かずに死にました。やめてくれｗ

DurableFunciton v3 → v4 isolated 書き換え表

v3	v4	備考
CallActivityWithRetryAsync	CallSubOrchestratorAsync	忘れないようにね。ログに乗らない

c. Activity

v3

using Commerble.EC.Entities.Models;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;
using Microsoft.Extensions.Logging;
using System.Linq;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class Sample1Activity
    {
        [FunctionName(nameof(Sample1Activity))]
        public async Task<Sample1Result> GetEcProduct([ActivityTrigger] IDurableActivityContext context, ILogger log)
        {
            var input = context.GetInput<なんか型>();

            なんか処理

            if (なんかエラーチェック)
            {
                log.LogError("Error Sample1Activity");
                return new Sample1Result { Succeed = false, hogehoge = null };
            }

            return new Sample1Result { Succeed = true, hogehoge = "hogehoge" };
        }
    }
}

v4

using Commerble.EC.Entities.Models;
using Microsoft.Azure.Functions.Worker;
using Microsoft.Extensions.Logging;
using Microsoft.DurableTask;
using Microsoft.DurableTask.Client;
using System.Linq;
using System.Threading.Tasks;

namespace SampleDurable
{
    public class Sample1Activity
    {
        private readonly ILogger _logger;

        public Sample1Activity(ILoggerFactory loggerFactory)
        {
            _logger = loggerFactory.CreateLogger<Sample1Activity>();
        }

        [Function(nameof(Sample1Activity))]
        public async Task<Sample1Result> GetEcProduct([ActivityTrigger] なんか型 input)
        {
            なんか処理

            if (なんかエラーチェック)
            {
                _logger.LogError(msg);
                return new Sample1Result { Succeed = false, hogehoge = null };
            }

            return new Sample1Result { Succeed = true, hogehoge = "hogehoge" };
        }
    }
}

DurableFunciton v3 → v4 isolated 書き換え表

v3	v4	備考
[ActivityTrigger]IDurableActivityContext	[ActivityTrigger]受け取りたい型	context.GetInput<受け取りたい型>でinput payloadをうけとっていましたが、引数にバインドする形で受け取れるようになりました。