Durable functions の概要
Azureのサーバレスのサービスとして Azure Functions があり、これの拡張機能としてのサービス。Azure Functionsに比べて長時間稼働などが可能となる。一言で簡略化していえば、
状態管理・再実行により長時間稼働に対応できる Azure Functions
と言えるかと思う。
Durable Functions は、Azure Functions の拡張機能であり、オーケストレーター、アクティビティ、エンティティ関数をコードに記述することで、サーバーレス環境でステートフル ワークフローを構築できます。 Durable Functions ランタイムは、状態、チェックポイント、再試行、復旧を管理するため、ワークフローを長期間にわたって確実に実行できます。
長時間稼働をするため、各タスクの進行状況をサーバだけでなくストレージアカウントを用いて管理するような仕組みになっている。
- キューストレージ
- テーブルストレージ
Task Hubとキュー・テーブルストレージの関係
durable functions の名前を付けることができ、公式ドキュメントでは Task Hub という名前で定義されている。Task Hub はストレージ管理のための論理的な単位としてみなせる。
# host.json
{
"version": "2.0",
"extensions": {
"durableTask": {
"hubName": "xyz" # ここがTaskhub名
}
}
}この Task Hub 名を使って、キューストレージ・テーブルストレージのテーブル名が作成される。
キューストレージ・テーブルストレージのうち、以下の４つが重要。
- インスタンステーブル（左から一つ目の instances と記載のもの）
- 履歴テーブル（左から２つ目の History と記載のもの）
- 作業項目キュー（左から３つ目の Work items と記載のもの）
- コントロールキュー（左４つ目の Control-00）
これらのストレージの名前は、Task Hub の名前から始まるようになっている。（インスタンステーブルであれば、<TaskHub名>Instances ）
durable functions をトリガーした際、インスタンスIDという一意のIDが自動生成される。
instance_id = await client.start_new() # ここでインスタンスIDが生成インスタンスIDはインスタンスの識別子であり、これが キューストレージ・テーブルストレージの各種テーブルのキーとして利用される。（＝どのインスタンスがもつレコードかが判別できるようになっている）
各種ストレージの役割について
キューストレージ
作業項目キューは、実行待ちのアクティビティを管理するためのもの。以下のメソッドでアクティビティが呼び出されるとこのキューの中に格納され、オーケストレータが順次実行していく。
# pythonランタイム
result1 = yield context.call_activity(...)コントロールキューはオーケストレータの管理用のキューとなっている。作業項目キューのアクティビティが完了すると、レスポンスなどの情報がこのキューへ格納される。
テーブルストレージ
キューストレージはメッセージ配信に利用されており、永続的に管理するためにテーブルストレージが必要となる。永続的に保管できるため、durable functions のリプレイ時が可能となる。
- コントロールキューの内容は一時的な保管のため、永続的に管理するために履歴テーブルが必要になる。
- インスタンステーブルにはインスタンスIDの情報（ステータス、作成日時等）が格納されている
durable functionsの大まかな処理の流れ
HTTP トリガーされた場合を想定。公式ドキュメントのコード例を参考に、durable の処理とキュー・テーブルの関係性を概略的に整理する。
# クライアント関数
@app.route(route="StartChaining", methods=["POST"])
@app.durable_client_input(client_name="client")
async def start_chaining(req: func.HttpRequest, client) -> func.HttpResponse:
"""HTTP trigger to start the function chaining orchestration."""
instance_id = await client.start_new("chaining_orchestration")
logging.info(f"Started chaining orchestration with ID = '{instance_id}'.")
return client.create_check_status_response(req, instance_id)
# アクティビティ関数
@app.activity_trigger(input_name="city")
def say_hello(city: str) -> str:
"""Activity function that returns a greeting for a city."""
logging.info(f"Saying hello to {city}.")
return f"Hello {city}!"
# オーケストレータ関数
@app.orchestration_trigger(context_name="context")
def chaining_orchestration(context: df.DurableOrchestrationContext):
"""Function chaining orchestration: calls activities sequentially."""
result1 = yield context.call_activity("say_hello", "Tokyo")
result2 = yield context.call_activity("say_hello", "Seattle")
result3 = yield context.call_activity("say_hello", "London")
return [result1, result2, result3]
HTTP trigger
↓
クライアント関数のインスタンス生成（client.start_new())
↓
インスタンステーブルに新規追加・管理される
↓
オーケストレータ関数がアクティビティをスケジュール（context.call_activity())
↓
作業項目キューにアクティビティが追加・管理
↓
アクティビティ関数がアクティビティを処理
↓
コントロールキューへレスポンス
↓
オーケストレータがヒストリーテーブルを使ってリプレイしながら処理を進める
参考
