Azure における Durable Functions でのインスタンスの管理

[アーティクル]
02/13/2024

Durable Functions のオーケストレーションは、組み込みの管理 API を使用して開始、クエリ実行、中断、再開、終了できる、長時間実行されるステートフル関数です。 Durable Functions のオーケストレーションクライアントバインディングによって、インスタンスへの外部イベントの送信や、インスタンス履歴の消去など、他のいくつかのインスタンス管理 API が公開されています。この記事では、サポートされているすべてのインスタンス管理操作について詳しく説明します。

インスタンスを開始する

オーケストレーションクライアントバインディング上の start-new (または schedule-new) メソッドにより、新しいオーケストレーションインスタンスが開始されます。内部的には、このメソッドは Durable Functions 記憶域プロバイダーを介してメッセージを書き込んだ後に戻ります。このメッセージにより、指定された名前を使用してオーケストレーション関数の開始が非同期にトリガーされます。

新しいオーケストレーションインスタンスを開始するためのパラメーターは次のとおりです。

Name:スケジュールするオーケストレーター関数の名前。
入力:オーケストレーター関数に入力として渡す必要のある JSON でシリアル化できる任意のデータ。
InstanceId: (省略可能) インスタンスの一意の ID。このパラメーターを指定しない場合、メソッドではランダムな ID が使用されます。

ヒント

可能な限り、インスタンス ID にはランダムな識別子を使用します。ランダムなインスタンス ID を使用すると、複数の VM にわたってオーケストレーター関数をスケーリングする場合に、負荷が均等に分散されるようになります。ランダムではないインスタンス ID を使用するのに適しているのは、外部ソースから ID を取得する必要があるとき、またはシングルトンオーケストレーターパターンを実装するときです。

次のコードは、新しいオーケストレーションインスタンスを開始する関数の例です。

[FunctionName("HelloWorldQueueTrigger")]
public static async Task Run(
    [QueueTrigger("start-queue")] string input,
    [DurableClient] IDurableOrchestrationClient starter,
    ILogger log)
{
    string instanceId = await starter.StartNewAsync("HelloWorld", input);
    log.LogInformation($"Started orchestration with ID = '{instanceId}'.");
}

Note

前記の C# コードは Durable Functions 2.x 用です。 Durable Functions 1.x では、DurableClient 属性の代わりに OrchestrationClient 属性を使用する必要があります。また、IDurableOrchestrationClient ではなく DurableOrchestrationClient パラメーター型を使用する必要があります。バージョン間の相違点の詳細については、Durable Functions のバージョンに関する記事を参照してください。

別途指定されていない限り、このページでは HTTP トリガーと次の function.json が使用されます。

function.json

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

Note

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

index.js

const df = require("durable-functions");

module.exports = async function(context, input) {
    const client = df.getClient(context);

    const instanceId = await client.startNew("HelloWorld", undefined, input);
    context.log(`Started orchestration with ID = ${instanceId}.`);
};

別途指定されていない限り、このページでは HTTP トリガーと次の function.json が使用されます。

function.json

{
  "scriptFile": "__init__.py",
  "bindings": [    
    {
      "name": "req",
      "type": "httpTrigger",
      "direction": "in",
      "methods": ["post"]
    },
    {
      "name": "$return",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Note

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

__init__.py

import logging
import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    
    instance_id = await client.start_new('HelloWorld', None, None)
    logging.log(f"Started orchestration with ID = ${instance_id}.")

別途指定されていない限り、このページでは HTTP トリガーと次の function.json が使用されます。

function.json

{
  "bindings": [    
    {
      "name": "Request",
      "type": "httpTrigger",
      "direction": "in",
      "methods": ["post"]
    },
    {
      "name": "Response",
      "type": "http",
      "direction": "out"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Note

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

run.ps1

param($Request, $TriggerMetadata)

$InstanceId = Start-DurableOrchestration -FunctionName 'HelloWorld'
Write-Host "Started orchestration with ID = '$InstanceId'"

@FunctionName("HelloWorldQueueTrigger")
public void helloWorldQueueTrigger(
        @QueueTrigger(name = "input", queueName = "start-queue", connection = "Storage") String input,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        final ExecutionContext context) {
    DurableTaskClient client = durableContext.getClient();        
    String instanceID = client.scheduleNewOrchestrationInstance("HelloWorld");
    context.getLogger().info("Scheduled orchestration with ID = " + instanceID);
}

オーケストレーターが開始されるのを待ってから関数に戻る場合は、waitForInstanceStart() メソッドを使用することもできます。

// wait up to 30 seconds for the scheduled orchestration to enter the "Running" state
client.waitForInstanceStart(instanceID, Duration.ofSeconds(30));

Azure Functions Core Tools

Core Tools で func durable start-new コマンドを使用することで、インスタンスを直接起動することもできます。その場合、次のパラメーターを受け取ります。

function-name (必須) : 開始する関数の名前。
input (省略可能) : 関数への入力 (インラインまたは JSON ファイル経由のどちらか)。ファイルの場合は、@ でファイルへのパスにプレフィックスを追加します (例: @path/to/file.json)。
id (省略可能) : オーケストレーションインスタンスの ID。このパラメーターを指定しないと、コマンドではランダムな GUID が使用されます。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、AzureWebJobsStorage です。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、DurableFunctionsHub です。 host.json で durableTask:HubName を使用してこれを設定することもできます。

Note

Core Tools のコマンドでは、関数アプリのルートディレクトリから実行されることが想定されています。 connection-string-setting および task-hub-name パラメーターを明示的に指定した場合、任意のディレクトリからコマンドを実行できます。関数アプリのホストが実行されていなくても、これらのコマンドを実行できますが、ホストが実行されていない場合、一部の効果を観察できないことがあります。たとえば、start-new コマンドではターゲットのタスクハブに開始メッセージがエンキューされますが、メッセージを処理できる関数アプリのホストプロセスが実行されていない限り、オーケストレーションは実際に実行されません。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

次のコマンドでは、HelloWorld という名前の関数が開始されて、counter-data.json ファイルの内容がその関数に渡されます。

func durable start-new --function-name HelloWorld --input @counter-data.json --task-hub-name TestTaskHub

インスタンスのクエリを実行する

新しいオーケストレーションインスタンスを開始した後、ほとんどの場合、ランタイム状態に対してクエリを実行して、それらが実行中であるか、完了したか、または失敗したかを調べることが必要になります。

オーケストレーションクライアントのバインド上の get-status メソッドは、オーケストレーションインスタンスの状態をクエリします。

これは、パラメーターとして instanceId (必須)、showHistory (省略可能)、showHistoryOutput (省略可能)、showInput (省略可能) を使用します。

showHistory :true に設定すると、応答に実行履歴が含まれます。
showHistoryOutput :true に設定すると、実行履歴にアクティビティ出力が含まれます。
showInput :false にすると、応答に関数の入力が含まれます。既定値は true です。

このメソッドは次のプロパティを持つオブジェクトを返します。

Name:オーケストレーター関数の名前。
InstanceId: オーケストレーションのインスタンス ID (instanceId 入力と同じにする必要があります)。
CreatedTime: オーケストレーター関数が実行を開始した時刻。
LastUpdatedTime: オーケストレーションが最後にチェックポイントされた時刻。
入力:JSON 値としての関数の入力。 showInput が false の場合、このフィールドは設定されません。
CustomStatus: JSON 形式でのカスタムオーケストレーションの状態。
出力:JSON 値としての関数の出力 (関数が完了している場合)。オーケストレーター関数が失敗した場合、このプロパティには、エラーの詳細が含まれます。オーケストレーター関数が中断または終了された場合、このプロパティには中断または終了の理由が含まれます (存在する場合)。
RuntimeStatus: 次のいずれかの値を指定できます。
- Pending: インスタンスはスケジュールされたが、まだ実行を開始していません。
- [実行中] : インスタンスが実行を開始しました。
- Completed: インスタンスが正常に完了しました。
- ContinuedAsNew: インスタンスが新しい履歴で自身を再開しました。これは一時的な状態です。
- [失敗] : インスタンスがエラーで失敗しました。
- [中止] : インスタンスが突然停止されました。
- [中断]: インスタンスは中断されました。後で再開することができます。
履歴: オーケストレーションの実行履歴。このフィールドにデータが設定されるのは、showHistory を true に設定した場合のみです。

Note

すべてのスケジュールされたタスクが完了し、かつオーケストレーターが返されるまで、オーケストレーターは Completed とマークされません。つまり、オーケストレーターが Completed とマークされるためには、return ステートメントに到達するだけでは不十分です。これは、WhenAny が使用されている場合に特に関係があります。これらのオーケストレーターは、スケジュールされたすべてのタスクが実行される前にreturnすることがよくあります。

このメソッドでは、インスタンスが存在しない場合 null (.NET と Java)、undefined (JavaScript)、または None (Python) を返します。

[FunctionName("GetStatus")]
public static async Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("check-status-queue")] string instanceId)
{
    DurableOrchestrationStatus status = await client.GetStatusAsync(instanceId);
    // do something based on the current status.
}

Note

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);

    const status = await client.getStatus(instanceId);
    // do something based on the current status.
    // example: if status.runtimeStatus === df.OrchestrationRuntimeStatus.Running: ...
}

function.json 構成については「インスタンスを開始する」を参照してください。

import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    status = await client.get_status(instance_id)
    # do something based on the current status
    # example: if (existing_instance.runtime_status is df.OrchestrationRuntimeStatus.Running) { ...

param($Request, $TriggerMetadata)

# Get instanceid from body
$InstanceId = $Request.Body.InstanceId

$Status = Get-DurableStatus -InstanceId $InstanceId -ShowHistory -ShowHistoryOutput -ShowInput
Write-Host "Status: $($Status | ConvertTo-Json)"

# Do something based on status
# example: if ($Status.runtimeStatus -eq 'Running') { ... }

@FunctionName("GetStatus")
public void getStatus(
        @QueueTrigger(name = "instanceID", queueName = "check-status-queue", connection = "Storage") String instanceID,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        final ExecutionContext context) {
    DurableTaskClient client = durableContext.getClient();        
    OrchestrationMetadata metadata = client.getInstanceMetadata(instanceID, false);
    if (metadata != null) {
        OrchestrationRuntimeStatus status = metadata.getRuntimeStatus();
        switch (status) {
            // do something based on the current status
        }
    }
}

Azure Functions Core Tools

Core Tools で func durable get-runtime-status コマンドを使用し、オーケストレーションインスタンスの状態を取得することもできます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable get-runtime-status コマンドは、次のパラメーターを受け取ります。

id (必須) : オーケストレーションインスタンスの ID。
show-input (省略可能) : true に設定すると、応答に関数の入力が含まれます。既定値は false です。
show-output (省略可能) : true に設定すると、応答に関数の出力が含まれます。既定値は false です。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

次のコマンドでは、0ab8c55a66644d68a3a8b220b12d209c というオーケストレーションインスタンス ID を持つインスタンスの状態 (入力と出力を含む) が取得されます。関数アプリのルートディレクトリから func コマンドを実行することが想定されています。

func durable get-runtime-status --id 0ab8c55a66644d68a3a8b220b12d209c --show-input true --show-output true

durable get-history コマンドを使用して、オーケストレーションインスタンスの履歴を取得できます。使用できるパラメーターは次のとおりです。

id (必須) : オーケストレーションインスタンスの ID。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

func durable get-history --id 0ab8c55a66644d68a3a8b220b12d209c

すべてのインスタンスのクエリを実行する

言語 SDK の API を使用して、タスクハブ内のすべてのオーケストレーションインスタンスの状態に対してクエリを実行できます。この "list-instances" または "get-status" API は、クエリパラメーターに一致するオーケストレーションインスタンスを表すオブジェクトの一覧を返します。

[FunctionName("GetAllStatus")]
public static async Task Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient client,
    ILogger log)
{
    var noFilter = new OrchestrationStatusQueryCondition();
    OrchestrationStatusQueryResult result = await client.ListInstancesAsync(
        noFilter,
        CancellationToken.None);
    foreach (DurableOrchestrationStatus instance in result.DurableOrchestrationState)
    {
        log.LogInformation(JsonConvert.SerializeObject(instance));
    }
    
    // Note: ListInstancesAsync only returns the first page of results.
    // To request additional pages provide the result.ContinuationToken
    // to the OrchestrationStatusQueryCondition's ContinuationToken property.
}

Note

const df = require("durable-functions");

module.exports = async function(context, req) {
    const client = df.getClient(context);

    const instances = await client.getStatusAll();
    instances.forEach((instance) => {
        context.log(JSON.stringify(instance));
    });
};

function.json 構成については「インスタンスを開始する」を参照してください。

import logging
import json
import azure.functions as func
import azure.durable_functions as df


async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    instances = await client.get_status_all()

    for instance in instances:
        logging.log(json.dumps(instance))

function.json 構成については「インスタンスを開始する」を参照してください。

@FunctionName("GetAllStatus")
public String getAllStatus(
        @HttpTrigger(name = "req", methods = {HttpMethod.GET}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    DurableTaskClient client = durableContext.getClient();
    OrchestrationStatusQuery noFilter = new OrchestrationStatusQuery();
    OrchestrationStatusQueryResult result = client.queryInstances(noFilter);
    return "Found " + result.getOrchestrationState().size() + " orchestrations.";
}

Azure Functions Core Tools

Core Tools で func durable get-instances コマンドを使用することでインスタンスにクエリを直接実行することもできます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable get-instances コマンドは、次のパラメーターを受け取ります。

top (省略可能) : このコマンドは、ページングをサポートします。このパラメーターは、要求ごとに取得されるインスタンス数に対応します。既定値は 10 です。
continuation-token (省略可能) : 取得するインスタンスのページまたはセクションを示すトークン。 get-instances を実行するたびに、次の一連のインスタンスにトークンが返されます。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

func durable get-instances

フィルターを使用してインスタンスのクエリを実行する

標準のインスタンスクエリで提供できるすべての情報が本当は必要ないときはどうしますか。たとえば、オーケストレーションの作成時刻やオーケストレーションの実行時の状態だけを調べている場合です。フィルターを適用することで、クエリを絞り込むことができます。

[FunctionName("QueryStatus")]
public static async Task Run(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestMessage req,
    [DurableClient] IDurableOrchestrationClient client,
    ILogger log)
{
    // Get the first 100 running or pending instances that were created between 7 and 1 day(s) ago
    var queryFilter = new OrchestrationStatusQueryCondition
    {
        RuntimeStatus = new[]
        {
            OrchestrationRuntimeStatus.Pending,
            OrchestrationRuntimeStatus.Running,
        },
        CreatedTimeFrom = DateTime.UtcNow.Subtract(TimeSpan.FromDays(7)),
        CreatedTimeTo = DateTime.UtcNow.Subtract(TimeSpan.FromDays(1)),
        PageSize = 100,
    };
    
    OrchestrationStatusQueryResult result = await client.ListInstancesAsync(
        queryFilter,
        CancellationToken.None);
    foreach (DurableOrchestrationStatus instance in result.DurableOrchestrationState)
    {
        log.LogInformation(JsonConvert.SerializeObject(instance));
    }
}

Note

const df = require("durable-functions");

module.exports = async function(context, req) {
    const client = df.getClient(context);

    const runtimeStatus = [
        df.OrchestrationRuntimeStatus.Completed,
        df.OrchestrationRuntimeStatus.Running,
    ];
    const instances = await client.getStatusBy(
        new Date(2021, 3, 10, 10, 1, 0),
        new Date(2021, 3, 10, 10, 23, 59),
        runtimeStatus
    );
    instances.forEach((instance) => {
        context.log(JSON.stringify(instance));
    });
};

function.json 構成については「インスタンスを開始する」を参照してください。

import logging
from datetime import datetime
import json
import azure.functions as func
import azure.durable_functions as df
from azure.durable_functions.models.OrchestrationRuntimeStatus import OrchestrationRuntimeStatus

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    runtime_status = [OrchestrationRuntimeStatus.Completed, OrchestrationRuntimeStatus.Running]

    instances = await client.get_status_by(
        datetime(2021, 3, 10, 10, 1, 0),
        datetime(2021, 3, 10, 10, 23, 59),
        runtime_status
    )

    for instance in instances:
        logging.log(json.dumps(instance))

@FunctionName("GetRunningInstances")
public String getRunningInstances(
        @HttpTrigger(name = "req", methods = {HttpMethod.GET}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    DurableTaskClient client = durableContext.getClient();
    OrchestrationStatusQuery filter = new OrchestrationStatusQuery()
        .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.PENDING, OrchestrationRuntimeStatus.RUNNING))
        .setCreatedTimeFrom(Instant.now().minus(Duration.ofDays(7)))
        .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(1)));
    OrchestrationStatusQueryResult result = client.queryInstances(filter);
    return "Found " + result.getOrchestrationState().size() + " orchestrations.";
}

Azure Functions Core Tools

Azure Functions Core Tools では、durable get-instances コマンドでフィルターを使用することもできます。前述の top、continuation-token、connection-string-setting、task-hub-name パラメーターに加えて、3 つのフィルターパラメーター (created-after、created-before、runtime-status) を使用できます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable get-instances コマンドのパラメーターを次に示します。

created-after (省略可能) : この日付/時刻 (UTC) の後に作成されたインスタンスを取得します。 ISO 8601 形式の日時が受け入れられます。
created-before (省略可能) : この日付/時刻 (UTC) の前に作成されたインスタンスを取得します。 ISO 8601 形式の日時が受け入れられます。
runtime-status (省略可能) : 特定の状態 (たとえば、実行中または完了) のインスタンスを取得します。複数の (スペースで区切られた) 状態を指定できます。
top (省略可能) : 要求ごとに取得されるインスタンスの数。既定値は 10 です。
continuation-token (省略可能) : 取得するインスタンスのページまたはセクションを示すトークン。 get-instances を実行するたびに、次の一連のインスタンスにトークンが返されます。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

フィルター (created-after、created-before、runtime-status) を何も指定しないと、コマンドでは、実行状態や作成時刻に関係なく、単に top インスタンスが取得されます。

func durable get-instances --created-after 2021-03-10T13:57:31Z --created-before  2021-03-10T23:59Z --top 15

インスタンスを終了する

実行に時間がかかっているオーケストレーションインスタンスがある場合、または単に何らかの理由で完了する前にインスタンスを停止する必要がある場合、終了できます。

終了 API の 2 つのパラメーターは "インスタンス ID" と "理由" 文字列であり、これらはログとインスタンスの状態に書き込まれます。

[FunctionName("TerminateInstance")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("terminate-queue")] string instanceId)
{
    string reason = "Found a bug";
    return client.TerminateAsync(instanceId, reason);
}

Note

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);

    const reason = "Found a bug";
    return client.terminate(instanceId, reason);
};

function.json 構成については「インスタンスを開始する」を参照してください。

import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    reason = "Found a bug"
    return await client.terminate(instance_id, reason)

param($Request, $TriggerMetadata)

# Get instance id from body
$InstanceId = $Request.Body.InstanceId
$Reason = 'Found a bug'

Stop-DurableOrchestration -InstanceId $InstanceId -Reason $Reason

@FunctionName("TerminateInstance")
public void terminateInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    String reason = "Found a bug";
    durableContext.getClient().terminate(instanceID, reason);
}

終了したインスタンスは、最終的に Terminated 状態に切り替えられます。ただし、この切り替えはすぐには行われません。代わりに、終了操作は、そのインスタンスに対する他の操作と共に、タスクハブでキューに登録されます。インスタンスクエリ API を使用して、終了したインスタンスが実際にTerminated状態になったことを確認できます。

Note

インスタンスの終了は、現在、伝達されません。アクティビティ関数およびサブオーケストレーションは、これらを呼び出したオーケストレーションインスタンスが終了しているかどうかに関係なく、完了するまで実行されます。

インスタンスの中断と再開

オーケストレーションを中断すると、実行中のオーケストレーションを停止できます。終了とは異なり、中断したオーケストレーターは後から再開することができます。

中断 API の 2 つのパラメーターはインスタンス ID と理由文字列であり、これらはログとインスタンスの状態に書き込まれます。

[FunctionName("SuspendResumeInstance")]
public static async Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("suspend-resume-queue")] string instanceId)
{
    string suspendReason = "Need to pause workflow";
    await client.SuspendAsync(instanceId, suspendReason);
    
    // Wait for 30 seconds to ensure that the orchestrator state is updated to suspended. 
    DateTime dueTime = context.CurrentUtcDateTime.AddSeconds(30);
    await context.CreateTimer(dueTime, CancellationToken.None);
    
    string resumeReason = "Continue workflow";
    await client.ResumeAsync(instanceId, resumeReason);
}

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);

    const suspendReason = "Need to pause workflow";
    await client.suspend(instanceId, suspendReason);

    // Wait for 30 seconds to ensure that the orchestrator state is updated to suspended.  
    const deadline = DateTime.fromJSDate(context.df.currentUtcDateTime, {zone: 'utc'}).plus({ seconds: 30 });
    yield context.df.createTimer(deadline.toJSDate());

    const resumeReason = "Continue workflow";
    await client.resume(instanceId, resumeReason);
};

import azure.functions as func
import azure.durable_functions as df
from datetime import timedelta

async def main(req: func.HttpRequest, starter: str, instance_id: str):
    client = df.DurableOrchestrationClient(starter)

    suspend_reason = "Need to pause workflow"
    await client.suspend(instance_id, suspend_reason)

    # Wait for 30 seconds to ensure that the orchestrator state is updated to suspended. 
    due_time = context.current_utc_datetime + timedelta(seconds=30)
    yield context.create_timer(due_time)

    resume_reason = "Continue workflow"
    await client.resume(instance_id, resume_reason)

param($Request, $TriggerMetadata)

# Get instance id from body
$InstanceId = $Request.Body.InstanceId
$SuspendReason = 'Need to pause workflow'

Suspend-DurableOrchestration -InstanceId $InstanceId -Reason $SuspendReason

# Wait for 30 seconds to ensure that the orchestrator state is updated to suspended.
$duration = New-TimeSpan -Seconds 30
Start-DurableTimer -Duration $duration

$ResumeReason = 'Continue workflow'
Resume-DurableOrchestration -InstanceId $InstanceId -Reason $ResumeReason

@FunctionName("SuspendResumeInstance")
public void suspendResumeInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    DurableTaskClient client = durableContext.getClient();  
    String suspendReason = "Need to pause workflow";
    client.suspendInstance(instanceID, suspendReason);

    // Wait for 30 seconds to ensure that the orchestrator state is updated to suspended. 
    ctx.createTimer(Duration.ofSeconds(30)).await();

    String resumeReason = "Continue workflow";
    client.getClient().resumeInstance(instanceID, resumeReason);
}

中断したインスタンスは、最終的に Suspended 状態に切り替えられます。ただし、この切り替えはすぐには行われません。代わりに、中断操作は、そのインスタンスに対する他の操作と共に、タスクハブでキューに登録されます。インスタンスクエリ API を使用して、実行中のインスタンスが実際に中断状態になったことを確認できます。

中断されたオーケストレーターは、再開されるとその状態が Running に戻ります。

Azure Functions Core Tools

Core Tools で func durable terminate コマンドを使用することで、オーケストレーションインスタンスを直接終了することもできます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable terminate コマンドは、次のパラメーターを受け取ります。

id (必須) : 終了するオーケストレーションインスタンスの ID。
reason (省略可能) : 終了の理由。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

次のコマンドでは、ID が 0ab8c55a66644d68a3a8b220b12d209c のオーケストレーションインスタンスを終了します。

func durable terminate --id 0ab8c55a66644d68a3a8b220b12d209c --reason "Found a bug"

インスタンスにイベントを送信する

一部のシナリオでは、オーケストレーター関数が待機して外部イベントをリッスンする必要があります。このことが役立つシナリオの例として、監視と人による操作のシナリオが挙げられます。

オーケストレーションクライアントの "イベント発生" API を使用して、実行中のインスタンスにイベント通知を送信できます。オーケストレーションは、"外部イベントの待機" に対するオーケストレーター API を使用して、これらのイベントをリッスンして応答できます。

"イベント発生" パラメーターは次のとおりです。

"インスタンス ID": インスタンスの一意の ID。
"イベント名": 送信するイベントの名前。
"イベントデータ": インスタンスに送信する JSON でシリアル化できるペイロード。

[FunctionName("RaiseEvent")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("event-queue")] string instanceId)
{
    int[] eventData = new int[] { 1, 2, 3 };
    return client.RaiseEventAsync(instanceId, "MyEvent", eventData);
}

Note

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);

    const eventData = [ 1, 2, 3 ];
    return client.raiseEvent(instanceId, "MyEvent", eventData);
};

function.json 構成については「インスタンスを開始する」を参照してください。

import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    event_data = [1, 2 ,3]
    return await client.raise_event(instance_id, 'MyEvent', event_data)

param($Request, $TriggerMetadata)

# Get instance id from body
$InstanceId = $Request.Body.InstanceId
$EventName = 'MyEvent'
$EventData = @(1,2,3)

Send-DurableExternalEvent -InstanceId $InstanceId -EventName $EventName -EventData $EventData

@FunctionName("RaiseEvent")
public void raiseEvent(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}) HttpRequestMessage<String> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext) {
    String instanceID = req.getBody();
    String eventName = "MyEvent";
    int[] eventData = { 1, 2, 3 };
    durableContext.getClient().raiseEvent(instanceID, eventName, eventData);
}

Note

指定したインスタンス ID のオーケストレーションインスタンスが存在しない場合、イベントメッセージは破棄されます。インスタンスが存在していても、まだイベントを待機していない場合、受信して処理する準備が整うまでイベントはインスタンスの状態で格納されます。

Azure Functions Core Tools

Core Tools で func durable raise-event コマンドを使用することで、オーケストレーションインスタンスに直接、イベントを発生させることもできます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable raise-event コマンドは、次のパラメーターを受け取ります。

id (必須) : オーケストレーションインスタンスの ID。
event-name :発生させるイベントの名前。
event-data (省略可能) : オーケストレーションインスタンスに送信するデータ。これは JSON ファイルへのパスにするか、コマンドラインでデータを直接指定することができます。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、 DurableFunctionsHubです。これは、host.json で durableTask:HubName を使用して設定することもできます。

func durable raise-event --id 0ab8c55a66644d68a3a8b220b12d209c --event-name MyEvent --event-data @eventdata.json

func durable raise-event --id 1234567 --event-name MyOtherEvent --event-data 3

オーケストレーションの完了の待機

実行時間の長いオーケストレーションでは、オーケストレーションの結果を待機して取得することが必要な場合があります。このような場合は、オーケストレーションに対してタイムアウト期間も定義できると便利です。タイムアウトを過ぎた場合は、結果ではなく、オーケストレーションの状態が返されます。

"完了まで待機またはチェック状態応答の作成" API を使用して、オーケストレーションインスタンスから実際の出力を同期的に取得できます。既定では、このメソッドの既定のタイムアウトは 10 秒で、ポーリング間隔は 1 秒です。

この API の使用方法を示す HTTP トリガー関数の例を次に示します。

// Copyright (c) .NET Foundation. All rights reserved.
// Licensed under the MIT License. See LICENSE in the project root for license information.

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

namespace VSSample
{
    public static class HttpSyncStart
    {
        private const string Timeout = "timeout";
        private const string RetryInterval = "retryInterval";

        [FunctionName("HttpSyncStart")]
        public static async Task<HttpResponseMessage> Run(
            [HttpTrigger(AuthorizationLevel.Function, methods: "post", Route = "orchestrators/{functionName}/wait")]
            HttpRequestMessage req,
            [DurableClient] 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}'.");

            TimeSpan timeout = GetTimeSpan(req, Timeout) ?? TimeSpan.FromSeconds(30);
            TimeSpan retryInterval = GetTimeSpan(req, RetryInterval) ?? TimeSpan.FromSeconds(1);
            
            return await starter.WaitForCompletionOrCreateCheckStatusResponseAsync(
                req,
                instanceId,
                timeout,
                retryInterval);
        }

        private static TimeSpan? GetTimeSpan(HttpRequestMessage request, string queryParameterName)
        {
            string queryParameterStringValue = request.RequestUri.ParseQueryString()[queryParameterName];
            if (string.IsNullOrEmpty(queryParameterStringValue))
            {
                return null;
            }

            return TimeSpan.FromSeconds(double.Parse(queryParameterStringValue));
        }
    }
}

const df = require("durable-functions");

const timeout = "timeout";
const retryInterval = "retryInterval";

module.exports = async function (context, req) {
    const client = df.getClient(context);
    const instanceId = await client.startNew(req.params.functionName, undefined, req.body);

    context.log(`Started orchestration with ID = '${instanceId}'.`);

    const timeoutInMilliseconds = getTimeInSeconds(req, timeout) || 30000;
    const retryIntervalInMilliseconds = getTimeInSeconds(req, retryInterval) || 1000;

    const response = client.waitForCompletionOrCreateCheckStatusResponse(
        context.bindingData.req,
        instanceId,
        timeoutInMilliseconds,
        retryIntervalInMilliseconds
    );
    return response;
};

function getTimeInSeconds(req, queryParameterName) {
    const queryValue = req.query[queryParameterName];
    return queryValue
        ? queryValue * 1000 // expected to be in seconds
        : undefined;
}

function.json 構成については「インスタンスを開始する」を参照してください。

import logging
import azure.functions as func
import azure.durable_functions as df

timeout = "timeout"
retry_interval = "retryInterval"

async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    instance_id = await client.start_new(req.route_params['functionName'], None, req.get_body())
    logging.log(f"Started orchestration with ID = '${instance_id}'.")

    timeout_in_milliseconds = get_time_in_seconds(req, timeout)
    timeout_in_milliseconds = timeout_in_milliseconds if timeout_in_milliseconds != None else 30000
    retry_interval_in_milliseconds = get_time_in_seconds(req, retry_interval)
    retry_interval_in_milliseconds = retry_interval_in_milliseconds if retry_interval_in_milliseconds != None else 1000

    return await client.wait_for_completion_or_create_check_status_response(
        req,
        instance_id,
        timeout_in_milliseconds,
        retry_interval_in_milliseconds
    )

def get_time_in_seconds(req: func.HttpRequest, query_parameter_name: str):
    query_value = req.params.get(query_parameter_name)
    return query_value if query_value != None else 1000

現在、Java には、このシナリオ向けの単一のメソッドはありません。ただし、いくつかの追加のコード行を使用して実装できます。

@FunctionName("HttpStartAndWait")
public HttpResponseMessage httpStartAndWait(
        @HttpTrigger(name = "req", route = "orchestrators/{functionName}/wait", methods = {HttpMethod.POST}) HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") 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);
    try {
        String timeoutString = req.getQueryParameters().get("timeout");
        Integer timeoutInSeconds = Integer.parseInt(timeoutString);
        OrchestrationMetadata orchestration = client.waitForInstanceCompletion(
                instanceId,
                Duration.ofSeconds(timeoutInSeconds),
                true /* getInputsAndOutputs */);
        return req.createResponseBuilder(HttpStatus.OK)
                .body(orchestration.getSerializedOutput())
                .header("Content-Type", "application/json")
                .build();
    } catch (TimeoutException timeoutEx) {
        // timeout expired - return a 202 response
        return durableContext.createCheckStatusResponse(req, instanceId);
    }
}

次の行で関数を呼び出します。タイムアウトには 2 秒、再試行間隔には 0.5 秒を使用します。

curl -X POST "http://localhost:7071/orchestrators/E1_HelloSequence/wait?timeout=2&retryInterval=0.5"

Note

上記の cURL コマンドは、プロジェクトに E1_HelloSequence という名前のオーケストレーター関数が含まれていることを前提としています。 HTTP トリガー関数の記述方法により、プロジェクト内の任意のオーケストレーター関数の名前に置き換えることができます。

オーケストレーションインスタンスからの応答を取得するために必要な時間に応じて 2 つのケースがあります。

オーケストレーションインスタンスが、定義されたタイムアウト (このケースでは 2 秒) 内に完了すると、応答は、同期して提供される実際のオーケストレーションインスタンス出力です。

HTTP/1.1 200 OK
Content-Type: application/json; charset=utf-8
Date: Thu, 14 Dec 2021 06:14:29 GMT
Transfer-Encoding: chunked

[
    "Hello Tokyo!",
    "Hello Seattle!",
    "Hello London!"
]

オーケストレーションインスタンスが、定義されたタイムアウト内に完了できない場合、応答は「HTTP API URL の検出」で説明されている既定のものになります。

HTTP/1.1 202 Accepted
Content-Type: application/json; charset=utf-8
Date: Thu, 14 Dec 2021 06:13:51 GMT
Location: http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177?taskHub={taskHub}&connection={connection}&code={systemKey}
Retry-After: 10
Transfer-Encoding: chunked

{
    "id": "d3b72dddefce4e758d92f4d411567177",
    "sendEventPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/raiseEvent/{eventName}?taskHub={taskHub}&connection={connection}&code={systemKey}",
    "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177?taskHub={taskHub}&connection={connection}&code={systemKey}",
    "terminatePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/terminate?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}",
    "suspendPostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/suspend?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}",
    "resumePostUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/d3b72dddefce4e758d92f4d411567177/resume?reason={text}&taskHub={taskHub}&connection={connection}&code={systemKey}"
}

Note

Webhook URL の形式は、実行している Azure Functions ホストのバージョンによって異なる場合があります。上記の例は、Azure Functions 3.0 ホストが対象です。

HTTP 管理 Webhook URL を取得する

外部システムを使用して、オーケストレーションに対するイベントを監視したり、発生させたりできます。外部システムは、HTTP API URL の検出に関するページで説明されている既定の応答の一部である Webhook URL を介して、Durable Functions と通信できます。 Webhook URL は、オーケストレーションクライアントのバインドを使用してプログラムからアクセスすることもできます。具体的には、"HTTP 管理ペイロードの作成" API を使用して、これらの Webhook URL を含むシリアル化可能なオブジェクトを取得できます。

"HTTP 管理ペイロードの作成" API には、次の 1 つのパラメーターがあります。

"インスタンス ID": インスタンスの一意の ID。

このメソッドは次の文字列プロパティを持つオブジェクトを返します。

Id:オーケストレーションのインスタンス ID (InstanceId 入力と同じにする必要があります)。
StatusQueryGetUri: オーケストレーションインスタンスの状態の URL。
SendEventPostUri: オーケストレーションインスタンスの "イベント発生" URL。
TerminatePostUri: オーケストレーションインスタンスの "終了" URL。
PurgeHistoryDeleteUri:オーケストレーションインスタンスの "消去履歴" URL。
suspendPostUri: オーケストレーションインスタンスの "中断" URL。
resumePostUri: オーケストレーションインスタンスの "再開" URL。

次の例に示すように、関数はこれらのオブジェクトのインスタンスを外部システムに送信して、対応するオーケストレーションでイベントを監視または発生させることができます。

[FunctionName("SendInstanceInfo")]
public static void SendInstanceInfo(
    [ActivityTrigger] IDurableActivityContext ctx,
    [DurableClient] IDurableOrchestrationClient client,
    [CosmosDB(
        databaseName: "MonitorDB",
        containerName: "HttpManagementPayloads",
        Connection = "CosmosDBConnectionSetting")]out dynamic document)
{
    HttpManagementPayload payload = client.CreateHttpManagementPayload(ctx.InstanceId);

    // send the payload to Azure Cosmos DB
    document = new { Payload = payload, id = ctx.InstanceId };
}

Note

前記の C# コードは Durable Functions 2.x 用です。 Durable Functions 1.x では、IDurableActivityContext ではなく DurableActivityContext を使用する必要があり、DurableClient 属性の代わりに OrchestrationClient 属性を使用する必要があり、IDurableOrchestrationClient ではなく DurableOrchestrationClient パラメーター型を使用する必要があります。バージョン間の相違点の詳細については、Durable Functions のバージョンに関する記事を参照してください。

const df = require("durable-functions");

modules.exports = async function(context, ctx) {
    const client = df.getClient(context);

    const payload = client.createHttpManagementPayload(ctx.instanceId);

    // send the payload to Azure Cosmos DB
    context.bindings.document = JSON.stringify({
        id: ctx.instanceId,
        payload,
    });
};

function.json 構成については「インスタンスを開始する」を参照してください。

import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.cosmosdb.cdb.Document:
    client = df.DurableOrchestrationClient(starter)

    payload = client.create_check_status_response(req, instance_id).get_body().decode()

    return func.cosmosdb.CosmosDBConverter.encode({
        id: instance_id,
        payload: payload
    })

using namespace System.Net

param($Request, $TriggerMetadata)

$InstanceId = $Request.Body.InstanceId
$Response = New-DurableOrchestrationCheckStatusResponse -Request $Request -InstanceId $InstanceId
Push-OutputBinding -Name Response -Value $Response

インスタンスを巻き戻す (プレビュー)

予期しない理由でオーケストレーションエラーが発生した場合は、その目的のために作成されている API を使用して、以前の正常な状態にインスタンスを "巻き戻す" ことができます。

Note

この API は、適切なエラー処理や再試行ポリシーの代わりとなるものではありません。予期しない理由でオーケストレーションインスタンスが失敗する場合にのみ使用するものです。 Failed 以外の状態 (Running、Pending、Terminated、Completed など) でのオーケストレーションは、"巻き戻し" できません。エラー処理と再試行ポリシーについて詳しくは、エラー処理に関する記事をご覧ください。

オーケストレーションクライアントのバインド上の RewindAsync (.NET) または rewind (JavaScript) メソッドを使用して、オーケストレーションを実行中の状態に戻します。またこのメソッドは、オーケストレーションエラーの原因となったアクティビティやサブオーケストレーションの実行失敗も再実行します。

たとえば、一連の人による承認が含まれるワークフローがあるものとします。承認が必要であることをユーザーに通知し、リアルタイムの応答を待機する一連のアクティビティ関数があるとします。すべての承認アクティビティが応答を受信するかタイムアウトになった後、アプリケーションの構成ミス (無効なデータベース接続文字列など) により別のアクティビティが失敗します。結果として、ワークフローの深い部分でオーケストレーションが失敗します。 RewindAsync (.NET) または rewind (JavaScript) API を使用すると、アプリケーション管理者は構成エラーを修正し、失敗したオーケストレーションを失敗の直前の状態に巻き戻すことができます。人間の対話手順はいずれも再承認が不要で、オーケストレーションは正常に完了できるようになります。

Note

"巻き戻し" 機能では、永続タイマーを使用したオーケストレーションインスタンスの巻き戻しはサポートされません。

[FunctionName("RewindInstance")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("rewind-queue")] string instanceId)
{
    string reason = "Orchestrator failed and needs to be revived.";
    return client.RewindAsync(instanceId, reason);
}

Note

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);

    const reason = "Orchestrator failed and needs to be revived.";
    return client.rewind(instanceId, reason);
};

function.json 構成については「インスタンスを開始する」を参照してください。

Azure Functions Core Tools

Core Tools で func durable rewind コマンドを使用することで、オーケストレーションインスタンスを直接巻き戻すこともできます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable rewind コマンドは、次のパラメーターを受け取ります。

id (必須) : オーケストレーションインスタンスの ID。
reason (省略可能) : オーケストレーションインスタンスを巻き戻す理由。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、host.json ファイル内のタスクハブ名が使用されます。

func durable rewind --id 0ab8c55a66644d68a3a8b220b12d209c --reason "Orchestrator failed and needs to be revived."

インスタンスの履歴を消去する

オーケストレーションに関連付けられているすべてのデータを削除するには、インスタンスの履歴を消去できます。たとえば、完了したインスタンスに関連付けられているストレージリソースを削除したいことがあります。そのためには、オーケストレーションクライアントによって定義された "消去インスタンス" API を使用します。

この最初の例では、1 つのオーケストレーションインスタンスを消去する方法を示します。

[FunctionName("PurgeInstanceHistory")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [QueueTrigger("purge-queue")] string instanceId)
{
    return client.PurgeInstanceHistoryAsync(instanceId);
}

const df = require("durable-functions");

module.exports = async function(context, instanceId) {
    const client = df.getClient(context);
    return client.purgeInstanceHistory(instanceId);
};

function.json 構成については「インスタンスを開始する」を参照してください。

import azure.functions as func
import azure.durable_functions as df

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)

    return await client.purge_instance_history(instance_id)

@FunctionName("PurgeInstance")
public HttpResponseMessage purgeInstance(
        @HttpTrigger(name = "req", methods = {HttpMethod.POST}, route = "purge/{instanceID}") HttpRequestMessage<?> req,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        @BindingName("instanceID") String instanceID) {
    PurgeResult result = durableContext.getClient().purgeInstance(instanceID);
    if (result.getDeletedInstanceCount() == 0) {
        return req.createResponseBuilder(HttpStatus.NOT_FOUND)
                .body("No completed instance with ID '" + instanceID + "' was found!")
                .build();
    } else {
        return req.createResponseBuilder(HttpStatus.OK)
                .body("Successfully purged data for " + instanceID)
                .build();
    }
}

次の例では、指定した時間間隔後に完了したすべてのオーケストレーションインスタンスの履歴を消去する、タイマーによってトリガーされる関数を示します。この場合は、30 日以上前に完了したすべてのインスタンスのデータが削除されます。この関数の例は、UTC の午後 12 時 00 分に 1 日に 1 回実行するようにスケジュールされています。

[FunctionName("PurgeInstanceHistory")]
public static Task Run(
    [DurableClient] IDurableOrchestrationClient client,
    [TimerTrigger("0 0 12 * * *")] TimerInfo myTimer)
{
    return client.PurgeInstanceHistoryAsync(
        DateTime.MinValue,
        DateTime.UtcNow.AddDays(-30),  
        new List<OrchestrationStatus>
        {
            OrchestrationStatus.Completed
        });
}

Note

purgeInstanceHistoryBy メソッドを使用し、複数のインスタンスのインスタンス履歴を条件付きで消去することができます。

function.json

{
  "bindings": [
    {
      "schedule": "0 0 12 * * *",
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in"
    },
    {
      "name": "starter",
      "type": "durableClient",
      "direction": "in"
    }
  ],
  "disabled": false
}

Note

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

index.js

const df = require("durable-functions");

module.exports = async function (context, myTimer) {
    const client = df.getClient(context);
    const createdTimeFrom = new Date(0);
    const createdTimeTo = new Date().setDate(today.getDate() - 30);
    const runtimeStatuses = [ df.OrchestrationRuntimeStatus.Completed ];
    return client.purgeInstanceHistoryBy(createdTimeFrom, createdTimeTo, runtimeStatuses);
};

import azure.functions as func
import azure.durable_functions as df
from azure.durable_functions.models.DurableOrchestrationStatus import OrchestrationRuntimeStatus
from datetime import datetime, timedelta

async def main(req: func.HttpRequest, starter: str, instance_id: str) -> func.HttpResponse:
    client = df.DurableOrchestrationClient(starter)
    created_time_from = datetime.min
    created_time_to = datetime.today() + timedelta(days = -30)
    runtime_statuses = [OrchestrationRuntimeStatus.Completed]

    return await client.purge_instance_history_by(created_time_from, created_time_to, runtime_statuses)

@FunctionName("PurgeInstances")
public void purgeInstances(
        @TimerTrigger(name = "purgeTimer", schedule = "0 0 12 * * *") String timerInfo,
        @DurableClientInput(name = "durableContext") DurableClientContext durableContext,
        ExecutionContext context) throws TimeoutException {
    PurgeInstanceCriteria criteria = new PurgeInstanceCriteria()
            .setCreatedTimeFrom(Instant.now().minus(Duration.ofDays(60)))
            .setCreatedTimeTo(Instant.now().minus(Duration.ofDays(30)))
            .setRuntimeStatusList(List.of(OrchestrationRuntimeStatus.COMPLETED));
    PurgeResult result = durableContext.getClient().purgeInstances(criteria);
    context.getLogger().info(String.format("Purged %d instance(s)", result.getDeletedInstanceCount()));
}

Note

履歴の消去の操作を成功させるには、ターゲットインスタンスの実行時の状態が完了、終了、または失敗になっている必要があります。

Azure Functions Core Tools

Core Tools で func durable purge-history コマンドを使用することでオーケストレーションインスタンスの履歴を消去できます。前のセクションの 2 つ目の C# の例と同様に、指定した時間間隔中に作成されたすべてのオーケストレーションインスタンスの履歴が消去されます。さらに、実行状態によって消去されるインスタンスをフィルター処理できます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable purge-history コマンドにはいくつかのパラメーターがあります。

created-after (省略可能) : この日付/時刻 (UTC) の後に作成されたインスタンスの履歴を消去します。 ISO 8601 形式の日時が受け入れられます。
created-before (省略可能) : この日付/時刻 (UTC) の前に作成されたインスタンスの履歴を消去します。 ISO 8601 形式の日時が受け入れられます。
runtime-status (省略可能) : 特定の状態 (たとえば、実行中または完了) のインスタンスの履歴を消去します。複数の (スペースで区切られた) 状態を指定できます。
connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、host.json ファイル内のタスクハブ名が使用されます。

次のコマンドでは、2021 年 11 月 14 日午後 7 時 35 分 (UTC) より前に作成されたすべての失敗したインスタンスの履歴が削除されます。

func durable purge-history --created-before 2021-11-14T19:35:00.0000000Z --runtime-status failed

タスクハブを削除する

Core Tools の func durable delete-task-hub コマンドを使用して、Azure storage のテーブル、キュー、BLOB などの、特定のタスクハブに関連付けられているすべてのストレージ成果物を削除できます。

Note

Core Tools のコマンドは、現在、ランタイム状態を永続化するために既定の Azure Storage プロバイダーを使用する場合にのみサポートされます。

durable delete-task-hub コマンドには 2 つのパラメーターがあります。

connection-string-setting (省略可能) : 使用するストレージ接続文字列を含むアプリケーション設定の名前。既定では、 AzureWebJobsStorageです。
task-hub-name (省略可能) : 使用する Durable Functions タスクハブの名前。既定では、host.json ファイル内のタスクハブ名が使用されます。

次のコマンドでは、UserTest タスクハブに関連付けられている Azure ストレージデータがすべて削除されます。

func durable delete-task-hub --task-hub-name UserTest

次のステップ

バージョン管理の方法

インスタンス管理用の組み込みの HTTP API リファレンス

Azure における Durable Functions でのインスタンスの管理

インスタンスを開始する

Azure Functions Core Tools

インスタンスのクエリを実行する

Azure Functions Core Tools

すべてのインスタンスのクエリを実行する

Azure Functions Core Tools

フィルターを使用してインスタンスのクエリを実行する

Azure Functions Core Tools

インスタンスを終了する

インスタンスの中断と再開

Azure Functions Core Tools

インスタンスにイベントを送信する

Azure Functions Core Tools

オーケストレーションの完了の待機

HTTP 管理 Webhook URL を取得する

インスタンスを巻き戻す (プレビュー)

Azure Functions Core Tools

インスタンスの履歴を消去する

Azure Functions Core Tools

タスク ハブを削除する

次のステップ

その他のリソース

タスクハブを削除する