Durable Functions におけるタスクハブ (Azure Functions)

[アーティクル]
05/11/2023

Durable Functions の "タスクハブ" は、保留中のすべての作業を含め、ストレージ内にあるアプリケーションの現在の状態を表すものです。関数アプリの実行中、オーケストレーション、アクティビティ、エンティティ関数の進行状況は、タスクハブに継続的に格納されます。これにより、アプリケーションが何らかの理由で一時的に停止または中断した後、再起動が必要になった場合でも、中断したところから処理を再開することができます。また、関数アプリを使ってコンピューティング worker を動的にスケーリングすることもできます。

Diagram showing concept of function app and task hub concept.

概念的には、タスクハブには次の情報が格納されます。

すべてのオーケストレーションとエンティティインスタンスのインスタンス状態。
以下を含む、処理されるメッセージ
- 実行を待機しているアクティビティを表すアクティビティメッセージ。
- インスタンスへの配信を待機している任意のインスタンスメッセージ。

アクティビティメッセージとインスタンスメッセージの違いは、アクティビティメッセージはステートレスなので、どこでも処理できるのに対し、インスタンスメッセージはインスタンス ID で識別される特定のステートフルインスタンス (オーケストレーションまたはエンティティ) に配信する必要がある点です。

内部的には、各ストレージプロバイダーは、インスタンスの状態とメッセージを表現するために異なる組織を使うことがあります。たとえば、Azure Storage プロバイダーの場合は Azure Storage Queues に、MSSQL プロバイダーの場合はリレーショナルテーブルにメッセージを格納します。これらの違いは、アプリケーションの設計に関する限り重要ではありませんが、パフォーマンス特性に影響するものもあります。この点については、後述する「ストレージでの表現」で説明します。

作業項目

タスクハブのアクティビティメッセージとインスタンスメッセージは、関数アプリが処理する必要がある作業を表しています。関数アプリは、実行中に、タスクハブから "作業項目" を継続的にフェッチします。各作業項目は 1 つ以上のメッセージを処理しています。 2 種類の作業項目があります。

アクティビティ作業項目: アクティビティ関数を実行してアクティビティメッセージを処理します。
オーケストレーター作業項目: オーケストレーターまたはエンティティ関数を実行して 1 つ以上のインスタンスメッセージを処理します。

worker は、構成された worker ごとのコンカレンシー制限に従って、複数の作業項目を同時に処理できます。

worker は作業項目を完了すると、その効果をタスクハブにコミットします。これらの効果は、実行された関数の種類によって異なります。

アクティビティ関数が完了すると、結果を含むインスタンスメッセージが作成され、親オーケストレーターインスタンスに送信されます。
オーケストレーター関数を使って、オーケストレーションの状態と履歴を更新します。新しいメッセージを作成することもできます。
完了したエンティティ関数を使って、エンティティの状態を更新します。新しいインスタンスメッセージを作成することもできます。

オーケストレーションの場合、各作業項目はそのオーケストレーションの実行の 1 つのエピソードを表します。オーケストレーターが処理する新しいメッセージが生成されると、エピソードが開始されます。このようなメッセージは、オーケストレーションを開始する必要があることを示す場合、アクティビティ、エンティティ呼び出し、タイマー、またはサブオーケストレーションが完了したことを示す場合、外部イベントを表す場合があります。このメッセージによって、オーケストレーターが結果を処理し、次のエピソードを続行する作業項目がトリガーされます。そのエピソードは、オーケストレーターが完了するか、新しいメッセージを待機する必要があるポイントに到達した時点で終了します。

実行例

ファンアウトファンインオーケストレーションについて考えてみましょう。2 つのアクティビティを並行して開始し、その両方が完了するまで待機するというものです。

[FunctionName("Example")]
public static async Task Run([OrchestrationTrigger] IDurableOrchestrationContext context)
{
    Task t1 = context.CallActivityAsync<int>("MyActivity", 1);
    Task t2 = context.CallActivityAsync<int>("MyActivity", 2);
    await Task.WhenAll(t1, t2);
}

module.exports = df.orchestrator(function*(context){
    const tasks = [];
    tasks.push(context.df.callActivity("MyActivity", 1));
    tasks.push(context.df.callActivity("MyActivity", 2));
    yield context.df.Task.all(tasks);
});

def orchestrator_function(context: df.DurableOrchestrationContext):
    tasks = []
    tasks.append(context.call_activity("MyActivity", 1))
    tasks.append(context.call_activity("MyActivity", 2))
    yield context.task_all(tasks)

param($Context)

$Tasks = @()

$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 1 -NoWait
$Tasks += Invoke-DurableActivity -FunctionName 'MyActivity' -Input 2 -NoWait

Wait-DurableTask -Task $Tasks

@FunctionName("Example")
public void exampleOrchestrator(
        @DurableOrchestrationTrigger(name = "ctx") TaskOrchestrationContext ctx) {
    Task<Void> t1 = ctx.callActivity("MyActivity", 1);
    Task<Void> t2 = ctx.callActivity("MyActivity", 2);
    ctx.allOf(List.of(t1, t2)).await();
}

このオーケストレーションは、クライアントによって開始された後、関数アプリによって一連の作業項目として処理されます。完了した各作業項目によって、コミット時にタスクハブの状態が更新されます。手順は次のとおりです。

あるクライアントが、インスタンス ID "123" の新しいオーケストレーションを開始するよう要求します。クライアントがこの要求を完了すると、タスクハブにはオーケストレーション状態のプレースホルダーとインスタンスメッセージが格納されます。

ラベル ExecutionStarted は、オーケストレーションの履歴に参加するさまざまなメッセージとイベントを識別する、多数ある履歴イベント種類の 1 つです。
worker は "オーケストレーター作業項目" を実行して ExecutionStarted メッセージを処理します。オーケストレーションコードの実行を開始するオーケストレーター関数を呼び出します。このコードを使って、2 つのアクティビティをスケジュールし、結果を待機している間は実行を停止します。 worker がこの作業項目をコミットすると、タスクハブのコンテンツは次のようになります

ランタイム状態は Running になり、2 つの新しい TaskScheduled メッセージが追加され、履歴には 5 つのイベント OrchestratorStarted、ExecutionStarted、TaskScheduled、TaskScheduled、OrchestratorCompleted が含まれるようになりました。これらのイベントは、このオーケストレーションの実行の最初のエピソードを表しています。
worker は "アクティビティ作業項目" を実行し、TaskScheduled メッセージの 1 つを処理します。入力 "2" を指定してアクティビティ関数を呼び出します。アクティビティ関数が完了すると、その結果を含む TaskCompleted メッセージが作成されます。 worker がこの作業項目をコミットすると、タスクハブのコンテンツは次のようになります
worker は "オーケストレーター作業項目" を実行して TaskCompleted メッセージを処理します。オーケストレーションがまだメモリにキャッシュされている場合は、そのまま実行を再開できます。そうでなければ、worker はまず履歴を再生してオーケストレーションの現在の状態を復旧します。次にオーケストレーションを継続し、アクティビティの結果を配信します。この結果を受け取った後も、オーケストレーションはまだ他のアクティビティの結果を待機しているので、もう一度実行を停止します。 worker がこの作業項目をコミットすると、タスクハブのコンテンツは次のようになります

オーケストレーションの履歴には、さらに 3 つのイベント OrchestratorStarted、TaskCompleted、OrchestratorCompleted が含まれています。これらのイベントは、このオーケストレーションの実行の 2 つ目のエピソードを表しています。
worker は "アクティビティ作業項目" を実行し、残りの TaskScheduled メッセージを処理します。入力 "1" を指定してアクティビティ関数を呼び出します。 worker がこの作業項目をコミットすると、タスクハブのコンテンツは次のようになります
worker はもう 1 つの "オーケストレーター作業項目" を実行して TaskCompleted メッセージを処理します。この 2 つ目の結果を受け取った後、オーケストレーションは完了します。 worker がこの作業項目をコミットすると、タスクハブのコンテンツは次のようになります

ランタイム状態は Completed になり、オーケストレーションの履歴にはさらに 4 つのイベント OrchestratorStarted、TaskCompleted、ExecutionCompleted、OrchestratorCompleted が含まれています。これらのイベントは、このオーケストレーションの実行の 3 つ目かつ最後のエピソードを表しています。

このオーケストレーションの実行の最終履歴には、12 個のイベント OrchestratorStarted、ExecutionStarted、TaskScheduled、TaskScheduled、OrchestratorCompleted、OrchestratorStarted、TaskCompleted、OrchestratorCompleted、OrchestratorStarted、TaskCompleted、ExecutionCompleted、OrchestratorCompleted が含まれます。

注意

表示されるスケジュールは 1 つだけではありません。わずかに異なるスケジュールが多数あります。たとえば、2 つ目のアクティビティが先に完了した場合、TaskCompleted インスタンスメッセージは両方とも 1 つの作業項目によって処理される可能性があります。その場合、エピソードは 2 つしかないので、実行履歴は少し短くなり、次の 10 個のイベントが含まれます: OrchestratorStarted、ExecutionStarted、TaskScheduled、TaskScheduled、OrchestratorCompleted、OrchestratorStarted、TaskCompleted、TaskCompleted、ExecutionCompleted、OrchestratorCompleted。

タスクハブの管理

次は、タスクハブの作成または削除、複数の関数アプリを実行する場合のタスクハブの正しい使い方、タスクハブのコンテンツを検査する方法について詳しく見てみましょう。

作成と削除

関数アプリの初回起動時には、必要なリソースをすべて含む空のタスクハブがストレージに自動的に作成されます。

既定の Azure Storage プロバイダーを使う場合、追加の構成は必要ありません。それ以外の場合は、ストレージプロバイダーがタスクハブに必要なストレージリソースを適切にプロビジョニングしてアクセスできるように、ストレージプロバイダーの構成手順に関する記事に従ってください。

注意

タスクハブは、関数アプリを停止または削除しても自動的には削除され "ません"。そのデータを保持する必要がなくなった場合は、タスクハブ、そのコンテンツ、または含まれているストレージアカウントを手動で削除する必要があります。

ヒント

開発シナリオでは、クリーンな状態からの再起動が必要になることがよくあります。これをすばやく行うには、構成したタスクハブの名前を変更するだけ済みます。こうすることで、アプリケーションの再起動時に新しい空のタスクハブが強制的に作成されます。この場合、古いデータは削除されないことに注意してください。

複数の関数アプリ

複数の関数アプリがストレージアカウントを共有している場合、各関数アプリには個別のタスクハブ名を構成する "必要があります"。この要件はステージングスロットにも適用されます。各ステージングスロットは、一意のタスクハブ名で構成する必要があります。単一のストレージアカウントには、複数のタスクハブを含めることができます。通常、この制限は他のストレージプロバイダーにも適用されます。

次の図は、共有ストレージアカウントと専用 Azure Storage アカウントの各関数アプリにタスクハブが 1 つあることを示しています。

Diagram showing shared and dedicated storage accounts.

Note

タスクハブ共有ルールの例外は、リージョンのディザスターリカバリー用にアプリを構成しているかどうかという点です。詳細については、ディザスターリカバリーと地理的分散に関する記事を参照してください。

コンテンツ検査

タスクハブのコンテンツを検査するには、一般的な方法がいくつかあります。

関数アプリ内では、インスタンスストアのクエリを実行するメソッドがクライアントオブジェクトに用意されています。サポートされているクエリの種類の詳細については、インスタンス管理に関する記事を参照してください。
同様に、HTTP API には、オーケストレーションとエンティティの状態のクエリを実行する REST 要求が用意されています。詳細については、HTTP の API リファレンスに関する記事を参照してください。
Durable Functions Monitor ツールを使うと、タスクハブを検査できます。また、視覚的な表示のためのさまざまなオプションが用意されています。

一部のストレージプロバイダーでは、基となるストレージに直接アクセスしてタスクハブを検査することもできます。

Azure Storage プロバイダーを使っている場合、インスタンス状態はインスタンステーブルと履歴テーブルに格納され、Azure Storage Explorer などのツールを使って検査することができます。
MSSQL ストレージプロバイダーを使っている場合、SQL クエリとツールを使って、データベース内のタスクハブのコンテンツを検査することができます。

ストレージでの表現

各ストレージプロバイダーは、ストレージ内のタスクハブを表現するために異なる内部組織を使います。この組織を理解することは、必須ではありませんが、関数アプリのトラブルシューティング時や、パフォーマンス、スケーラビリティ、コスト目標を確保しようとするときに役立ちます。そのため、ストレージプロバイダーごとに、データがストレージ内でどのように編成されるかについて簡単に説明します。さまざまなストレージプロバイダーのオプションとその比較については、Durable Functions ストレージプロバイダーに関する記事を参照してください。

Azure Storage プロバイダー

Azure Storage プロバイダーは、次のコンポーネントを使用してストレージ内のタスクハブを表します。

2 つの Azure テーブルにインスタンスの状態が格納されます。
1 つの Azure キューにアクティビティメッセージが格納されます。
1 つ以上の Azure キューにインスタンスメッセージが格納されます。これらのいわゆる "コントロールキュー" はそれぞれ、インスタンス ID のハッシュに基づいて、すべてのインスタンスメッセージのサブセットが割り当てられるパーティションを表します。
BLOB やサイズの大きいメッセージのリースに使用される追加の BLOB コンテナーがいくつかあります。

たとえば、PartitionCount = 4 を持つ xyz という名前が付いたタスクハブには、次のキューとテーブルが含まれています。

Diagram showing Azure Storage provider storage storage organization for 4 control queues.

次に、これらのコンポーネントとそれらが果たす役割について詳しく説明します。

Azure Storage プロバイダーでタスクハブがどのように表現されているかについては、Azure Storage プロバイダーのドキュメントを参照してください。

Netherite ストレージプロバイダー

Netherite は、タスクハブのすべての状態を、指定した数のパーティションに分割します。ストレージでは、次のリソースが使われます。

1 つの Azure Storage BLOB コンテナー。パーティションごとにグループ化されたすべての BLOB が格納されています。
1 つの Azure テーブル。パーティションに関する発行されたメトリックが格納されています。
パーティション間でメッセージを配信するための Azure Event Hubs 名前空間。

たとえば、PartitionCount = 32 を持つ mytaskhub というタスクハブは、ストレージ内で次のように表されます。

Diagram showing Netherite storage organization for 32 partitions.

Note

タスクハブの状態はすべて x-storage BLOB コンテナー内に格納されます。 DurableTaskPartitions テーブルと EventHubs 名前空間には冗長データが含まれています。そのコンテンツが失われた場合、自動的に復旧することができます。そのため、既定の有効期限を過ぎたメッセージを保持するために、Azure Event Hubs 名前空間を構成する必要はありません。

Netherite は、パーティションの現在の状態を表すために、ログとチェックポイントに基づいたイベントソーシングメカニズムを使います。ブロック BLOB とページ BLOB の両方が使われます。この形式をストレージから直接読み取ることはできないため、インスタンスストアに対してクエリを実行するときには関数アプリが実行中である必要があります。

Netherite ストレージプロバイダーのタスクハブの詳細については、Netherite ストレージプロバイダーのタスクハブ情報に関する記事を参照してください。

MSSQL ストレージプロバイダー

すべてのタスクハブのデータは、いくつかのテーブルを使って、1 つのリレーショナルデータベースに格納されます。

dt.Instances と dt.History の各テーブルにはインスタンスの状態が格納されます。
dt.NewEvents テーブルにはインスタンスメッセージが格納されます。
dt.NewTasks テーブルにはアクティビティメッセージが格納されます。

Diagram showing MSSQL storage organization.

複数のタスクハブが同じデータベース内に独立して共存できるように、各テーブルには主キーの一部として TaskHub 列が含まれています。他の 2 つのプロバイダーとは異なり、MSSQL プロバイダーにはパーティションの概念がありません。

MSSQL ストレージプロバイダーのタスクハブの詳細については、Microsoft SQL (MSSQL) ストレージプロバイダーのタスクハブ情報に関する記事を参照してください。

タスクハブ名

タスクハブは、これらの規則に準拠する必要がある名前で識別されます。

英数字のみを含む
文字で始まる
最小長が 3 文字で、最大長が 45 文字である

次の例に示すように、タスクハブ名は host.json ファイルで宣言されます。

host.json (Functions 2.0)

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "MyTaskHub"
    }
  }
}

host.json (Functions 1.x)

{
  "durableTask": {
    "hubName": "MyTaskHub"
  }
}

次の host.json ファイルの例に示すように、タスクハブはアプリの設定を使用して構成することもできます。

host.json (Functions 1.0)

{
  "durableTask": {
    "hubName": "%MyTaskHub%"
  }
}

host.json (Functions 2.0)

{
  "version": "2.0",
  "extensions": {
    "durableTask": {
      "hubName": "%MyTaskHub%"
    }
  }
}

タスクハブ名は、MyTaskHub アプリ設定の値に設定されます。次の local.settings.json は、MyTaskHub 設定を samplehubname として定義する方法を示しています。

{
  "IsEncrypted": false,
  "Values": {
    "MyTaskHub" : "samplehubname"
  }
}

Note

デプロイスロットを使用する場合は、アプリ設定を使用してタスクハブ名を構成することをお勧めします。特定のスロットが常に特定のタスクハブを使用するようにする場合は、"スロット固定" アプリ設定を使用します。

タスクハブ名は、host.json に加えて、オーケストレーションクライアントバインディングメタデータでも設定できます。これは、別の関数アプリに存在するオーケストレーションまたはエンティティにアクセスする必要がある場合に便利です。次のコードでは、アプリ設定として構成されているタスクハブを操作するためにオーケストレーションクライアントバインドを使用する関数の記述方法を示しています。

[FunctionName("HttpStart")]
public static async Task<HttpResponseMessage> Run(
    [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}")] HttpRequestMessage req,
    [DurableClient(TaskHub = "%MyTaskHub%")] IDurableOrchestrationClient starter,
    string functionName,
    ILogger log)
{
    // Function input comes from the request content.
    object eventData = await req.Content.ReadAsAsync<object>();
    string instanceId = await starter.StartNewAsync(functionName, eventData);

    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");

    return starter.CreateCheckStatusResponse(req, instanceId);
}

Note

前記の例は Durable Functions 2.x 用です。 Durable Functions 1.x の場合、IDurableOrchestrationContext の代わりに DurableOrchestrationContext を使用する必要があります。バージョン間の相違点の詳細については、Durable Functions のバージョンに関する記事を参照してください。

function.json ファイルのタスクハブプロパティは、アプリ設定を通じて設定されます。

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

Note

この例では、Durable Functions 2.x をターゲットにしています。バージョン 1.x の場合、durableClient ではなく orchestrationClient を使用します。

function.json ファイルのタスクハブプロパティは、アプリ設定を通じて設定されます。

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

Note

この例では、Durable Functions 2.x をターゲットにしています。バージョン 1.x の場合、durableClient ではなく orchestrationClient を使用します。

function.json ファイルのタスクハブプロパティは、アプリ設定を通じて設定されます。

{
    "name": "input",
    "taskHub": "%MyTaskHub%",
    "type": "durableClient",
    "direction": "in"
}

Note

この例では、Durable Functions 2.x をターゲットにしています。バージョン 1.x の場合、durableClient ではなく orchestrationClient を使用します。

@FunctionName("HttpStart")
public HttpResponseMessage httpStart(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext", taskHub = "%MyTaskHub%") DurableClientContext durableContext,
        @BindingName("functionName") String functionName,
        final ExecutionContext context) {

    DurableTaskClient client = durableContext.getClient();
    String instanceId = client.scheduleNewOrchestrationInstance(functionName);
    context.getLogger().info("Created new Java orchestration with instance ID = " + instanceId);
    return durableContext.createCheckStatusResponse(req, instanceId);
}

Note

クライアントバインドメタデータでタスクハブ名を構成する必要があるのは、ある関数アプリを使用して別の関数アプリのオーケストレーションとエンティティにアクセスする場合のみです。クライアント関数がオーケストレーションやエンティティと同じ関数アプリで定義されている場合は、バインドメタデータでタスクハブ名を指定するのは避ける必要があります。既定では、すべてのクライアントバインドでは、タスクハブのメタデータは host.json の設定から取得されます。

タスクハブの名前は、先頭文字をアルファベットとする必要があります。また、使用できるのはアルファベットと数値だけです。指定しない場合、次の表に示すように、既定のタスクハブ名が使用されます。

Durable の拡張機能のバージョン	既定のタスクハブ名
2.x	Azure にデプロイする場合、タスクハブ名は関数アプリの名前から派生します。 Azure の外部で実行する場合、既定のタスクハブ名は `TestHubName` です。
1.x	すべての環境で、既定のタスクハブ名は `DurableFunctionsHub` です。

拡張機能のバージョン間の相違点の詳細については、Durable Functions のバージョンに関する記事を参照してください。

Note

この名前は共有ストレージアカウント内に複数のタスクハブがある場合に、それぞれのタスクハブを区別するものです。共有ストレージアカウントを共有する関数アプリが複数ある場合、host.json ファイルでタスクハブごとに異なる名前を明示的に構成する必要があります。このようにしないと、複数の関数アプリがメッセージをめぐって互いに競合し、その結果、オーケストレーションが Pending または Running 状態で予期せず "スタック" するなど、未定義の動作が発生する可能性があります。

次のステップ

オーケストレーションのバージョン管理を処理する方法の詳細

次の方法で共有

Durable Functions におけるタスクハブ (Azure Functions)

作業項目

実行例

タスクハブの管理

作成と削除

複数の関数アプリ

コンテンツ検査

ストレージでの表現

Azure Storage プロバイダー

Netherite ストレージプロバイダー

MSSQL ストレージプロバイダー

タスクハブ名

host.json (Functions 2.0)

host.json (Functions 1.x)

host.json (Functions 1.0)

host.json (Functions 2.0)

次のステップ

フィードバック

その他のリソース

次の方法で共有

Durable Functions におけるタスク ハブ (Azure Functions)

作業項目

実行例

タスク ハブの管理

作成と削除

複数の関数アプリ

コンテンツ検査

ストレージでの表現

Azure Storage プロバイダー

Netherite ストレージ プロバイダー

MSSQL ストレージ プロバイダー

タスク ハブ名

host.json (Functions 2.0)

host.json (Functions 1.x)

host.json (Functions 1.0)

host.json (Functions 2.0)

次のステップ

フィードバック

その他のリソース

Durable Functions におけるタスクハブ (Azure Functions)

タスクハブの管理

Netherite ストレージプロバイダー

MSSQL ストレージプロバイダー

タスクハブ名