Azure Event Grid の 30 秒タイムアウト制限を Azure Durable Functions で回避する

import json
import logging
import os
import time
from datetime import datetime, timezone, timedelta

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


app = df.DFApp()

# 既存の WAITE_SECONDS も互換で読む。今後は WAIT_SECONDS 推奨。
WAIT_SECONDS = int(os.getenv("WAIT_SECONDS", os.getenv("WAITE_SECONDS", "50")))
ACTIVE_RUNTIME_STATUSES = {"running", "pending", "continuedasnew"}


def _runtime_status_text(runtime_status):
    if runtime_status is None:
        return None

    runtime_status_value = (
        getattr(runtime_status, "value", None)
        or getattr(runtime_status, "name", None)
        or runtime_status
    )
    return str(runtime_status_value)


def _is_active_runtime_status(runtime_status):
    if runtime_status is None:
        return False

    normalized = runtime_status.rsplit(".", 1)[-1].lower()
    return normalized in ACTIVE_RUNTIME_STATUSES


def _existing_status_for_log(existing_status):
    if isinstance(existing_status, dict):
        return existing_status

    for method_name in ("to_json", "to_dict"):
        method = getattr(existing_status, method_name, None)
        if callable(method):
            try:
                return method()
            except Exception as exc:
                return {
                    "repr": repr(existing_status),
                    "serialization_error": f"{method_name}: {exc}",
                }

    status_dict = getattr(existing_status, "__dict__", None)
    if status_dict is not None:
        return status_dict

    return existing_status


@app.function_name(name="BlobCreatedEventGridProbe")
@app.event_grid_trigger(arg_name="event")
@app.durable_client_input(client_name="client")
async def blob_created_event_grid_probe(
    event: func.EventGridEvent,
    client: df.DurableOrchestrationClient,
) -> None:
    """
    Event Grid から直接呼ばれる関数。

    ここでは長時間処理をしない。
    Durable Orchestration を開始して、すぐに正常終了する。
    """

    data = event.get_json()
    blob_url = data.get("url") if isinstance(data, dict) else None

    event_time = event.event_time
    if event_time is not None and event_time.tzinfo is None:
        event_time = event_time.replace(tzinfo=timezone.utc)
    elif event_time is not None:
        event_time = event_time.astimezone(timezone.utc)

    payload = {
        "probe": "event-grid-blob-created-latency",
        "utc_now": datetime.now(timezone.utc).isoformat(),
        "event_id": event.id,
        "event_type": event.event_type,
        "event_time": event_time.isoformat() if event_time else None,
        "subject": event.subject,
        "blob_url": blob_url,
        "data": data,
        "wait_seconds": WAIT_SECONDS,
    }

    # Event Grid の同一イベントが再配信されたか確認しやすくするため、
    # event.id から固定 instance_id を作る。
    instance_id = f"eg-{event.id}"

    logging.info(
        "EVENTGRID_PROBE_RECEIVED %s",
        json.dumps(
            {
                **payload,
                "instance_id": instance_id,
            },
            ensure_ascii=False,
            default=str,
        ),
    )

    # 重複配送、または Event Grid retry が発生した場合に、
    # 同じ event_id / instance_id で検知しやすくする。
    existing_status = await client.get_status(instance_id)

    if existing_status is not None:
        runtime_status = getattr(existing_status, "runtime_status", None)

        if runtime_status is None:
            runtime_status = getattr(existing_status, "runtimeStatus", None)

        if runtime_status is None and isinstance(existing_status, dict):
            runtime_status = (
                existing_status.get("runtimeStatus")
                or existing_status.get("runtime_status")
            )

        runtime_status_text = _runtime_status_text(runtime_status)

        logging.info(
            "EVENTGRID_PROBE_EXISTING_STATUS_RAW %s",
            json.dumps(
                {
                    "event_id": event.id,
                    "instance_id": instance_id,
                    "existing_status_type": str(type(existing_status)),
                    "existing_status": _existing_status_for_log(existing_status),
                    "existing_status_repr": repr(existing_status),
                    "runtime_status": runtime_status_text,
                    "utc_now": datetime.now(timezone.utc).isoformat(),
                },
                ensure_ascii=False,
                default=str,
            ),
        )

        if _is_active_runtime_status(runtime_status_text):
            logging.info(
                "EVENTGRID_PROBE_DUPLICATE_OR_ALREADY_STARTED %s",
                json.dumps(
                    {
                        "event_id": event.id,
                        "instance_id": instance_id,
                        "runtime_status": runtime_status_text,
                        "utc_now": datetime.now(timezone.utc).isoformat(),
                    },
                    ensure_ascii=False,
                    default=str,
                ),
            )
            return

        logging.info(
            "EVENTGRID_PROBE_EXISTING_STATUS_NOT_DUPLICATE %s",
            json.dumps(
                {
                    "event_id": event.id,
                    "instance_id": instance_id,
                    "runtime_status": runtime_status_text,
                    "reason": "runtime_status_is_not_active",
                    "utc_now": datetime.now(timezone.utc).isoformat(),
                },
                ensure_ascii=False,
                default=str,
            ),
        )

    started_instance_id = await client.start_new(
        "EventGridProbeOrchestrator",
        instance_id,
        payload,
    )

    logging.info(
        "EVENTGRID_PROBE_ORCHESTRATION_STARTED %s",
        json.dumps(
            {
                "event_id": event.id,
                "instance_id": started_instance_id,
                "wait_seconds": WAIT_SECONDS,
                "utc_now": datetime.now(timezone.utc).isoformat(),
            },
            ensure_ascii=False,
            default=str,
        ),
    )

    # ここで return するため、Event Grid にはすぐ成功応答される。
    return


@app.function_name(name="EventGridProbeOrchestrator")
@app.orchestration_trigger(context_name="context")
def event_grid_probe_orchestrator(context: df.DurableOrchestrationContext):
    """
    Durable Orchestrator。
    Activity 内で time.sleep する。
    """

    payload = context.get_input()
    result = yield context.call_activity("SleepAndLogEventGridProbe", payload)
    return result


@app.function_name(name="SleepAndLogEventGridProbe")
@app.activity_trigger(input_name="payload")
def sleep_and_log_event_grid_probe(payload: dict) -> dict:
    """
    Activity 内で time.sleep する。
    """

    wait_seconds = int(payload.get("wait_seconds", 50))

    logging.info(
        "EVENTGRID_PROBE_ACTIVITY_SLEEP_STARTED %s",
        json.dumps(
            {
                "utc_now": datetime.now(timezone.utc).isoformat(),
                "event_id": payload.get("event_id"),
                "wait_seconds": wait_seconds,
            },
            ensure_ascii=False,
            default=str,
        ),
    )

    time.sleep(wait_seconds)

    record = {
        "probe": "event-grid-blob-created-latency",
        "phase": "activity_sleep_completed",
        "utc_now": datetime.now(timezone.utc).isoformat(),
        "event_id": payload.get("event_id"),
        "event_type": payload.get("event_type"),
        "event_time": payload.get("event_time"),
        "subject": payload.get("subject"),
        "blob_url": payload.get("blob_url"),
        "wait_seconds": wait_seconds,
    }

    logging.info(
        "EVENTGRID_PROBE_ACTIVITY_SLEEP_COMPLETED %s",
        json.dumps(record, ensure_ascii=False, default=str),
    )

    return record