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

[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}'.");
}

別途指定されていない限り、このページでは 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
}

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
}

__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
}

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));

[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.
}

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
        }
    }
}

[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.
}

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.";
}

[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));
    }
}

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.";
}

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

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);
}

[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);
}

[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);
}

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);
}

// 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);
    }
}

[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 };
}

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

[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);
}

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 構成については「インスタンスを開始する」を参照してください。

[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();
    }
}

[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
        });
}

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

function.json

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

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()));
}