ステートフルなサーバレスワークフローの構築を容易にするAzure Orchestrator

Azure Durable Functionsは、オーケストレータ機能という概念を導入して、より複雑なワークフローの定義を可能にすることで、サーブレスコンピューティングのパラダイムの拡張を目指すものだ。同じニーズを抱えた開発者のため、Microsoftは今回、サーバレスコンピューティングとオーケストレータ機能を新たに始める上で、まさに最適なウォークスルーを公開した。

Azure FunctionsはサーバレスコンピューティングにおけるMicrosoftの取り組みである。入力を取り、それを処理し、トリガによるアクション時に結果を返すという、基本的なメカニズムを提供する。Durable Functionsは、元来ステートレスであるそれに状態を追加するという、興味深い方法でそれを拡張したものだ。Durable Functionsは、いわゆるオーケストレータ関数を単一の論理トランザクションにラップした関数のセットであり、複雑なワークフローのコードによる記述と実行を可能にする。

Azure Functionsを支える２つの基本的な概念は、関数とトリガだ。関数は、静的クラスの単なるメソッドのひとつである。またトリガは、コードが応答するイベントで、その典型的なものとしては、ユーザからのHTTP要求に応答するHttpや、コードの実行をスケジュール可能なTimerなどがある。さらにDurable Functionsでは、新たに２つの概念が導入されている。

• オーケストレータ関数 – これはクラウドベースのコルーチンの一種であると解釈できる。別の言い方をすれば、オーケストレータ関数では、実行中にチェックポイントを設定して、他の関数が実行を完了するまでの間は停止し、完了するとその停止位置から再開する、ということが可能である。

• アクティビティ関数 – オーケストレータ内部で使用可能な関数で、オーケストレータからの呼び出しのみに応答することができる。アクティビティ関数は、同じオーケストレータインスタンス内で1回だけ実行されて、その結果がキャッシュされる。下記に示すのは、GitHubリポジトリの全リストを取り出すアクティビティ関数の例である。

  [FunctionName("GetAllRepositoriesForOrganization")]
  public static async Task<List> GetAllRepositoriesForOrganization([ActivityTrigger] DurableActivityContext context)
  {
  // retrieves the organization name from the Orchestrator function
  var organizationName = context.GetInput();
  // invoke the API to retrieve the list of repositories of a specific organization
  var repositories = (await github.Repository.GetAllForOrg(organizationName)).Select(x => (x.Id, x.Name)).ToList();
  return repositories;
  }
  

ActivityTrigger属性で、パラメータとしてDurableActivityContextを使用している点に注目してほしい。これは、関数をアクティビティ関数とするためのものだ。

この点に注意すれば、次の例のように、組織内のすべてのリポジトリを取得して、リポジトリ毎の未解決のイシューの数を計数し、結果をどこかに格納するようなオーケストレータ関数の定義が可能になる。

[FunctionName("Orchestrator")]
public static async Task RunOrchestrator(
    [OrchestrationTrigger] DurableOrchestrationContext context)
{
    // retrieves the organization name from the Orchestrator_HttpStart function
    var organizationName = context.GetInput();
    // retrieves the list of repositories for an organization by invoking a separate Activity Function.
    var repositories = await context.CallActivityAsync<List>("GetAllRepositoriesForOrganization", organizationName);

    // Creates an array of task to store the result of each functions
    var tasks = new Task[repositories.Count];
    for (int i = 0; i < repositories.Count; i++)
    {
        // Starting a `GetOpenedIssues` activity WITHOUT `async`
        // This will starts Activity Functions in parallel instead of sequentially.
        tasks[i] = context.CallActivityAsync("GetOpenedIssues", (repositories[i]));
    }

    // Wait for all Activity Functions to complete execution
    await Task.WhenAll(tasks);

    // Retrieve the result of each Activity Function and return them in a list
    var openedIssues = tasks.Select(x => x.Result).ToList();

    // Send the list to an Activity Function to save them to Blob Storage.
    await context.CallActivityAsync("SaveRepositories", openedIssues);

    return context.InstanceId;
}

上の例では、OrchestrationTrigger属性をDurableOrchestrationContextをパラメータとして使用することで、RunOrchestratorをオーケストレータ関数として修飾している点に注目してほしい。

このコードで興味を引くのは、contextメソッドの各呼び出しそのものがAzure Functionsであることだ。これはつまり、ワークフローの実行において、Azure Functionsのスケーラビリティと信頼性のメリットを活用できるということである。

上記のワークフローはGitHubで公開されている。詳細については、Microsoftのチュートリアルで必ず確認してほしい。