Azure Functions の Azure Queue Storage トリガーの開発環境を作る #Azure

新年早々 Azure Functions に苦しめられている Docker おじさんです。

Azure Functions 難しくないですか？私はそろそろ心が折れそうです。「App Service で扱えない重めの非同期処理を FaaS に投げたいな〜」くらいの軽い気持ちではじめただけなんですが、まさかこんな面倒なことになるなんて。

まだ Azure Functions の全容は掴めていないですが、とりあえず Azurite の Queue と組み合わせてローカルでそれっぽい開発環境は作れたのでメモをしておきます。

作るもの

Azure Functions では HTTP トリガーだと「即時にレスポンスを返して裏で非同期処理しておく」といったことが出来ないようなので、ジョブキュー (Azure Queue Storage) を用意しておいてそれをトリガーに関数を起動するようなやつを作ります。

Functions サービスの設定

compose.yml

services:

  functions:
    build: functions
    platform: linux/x86_64
    volumes:
      - ./functions:/home/site/wwwroot
    environment:
      AZURE_QUEUE_STORAGE: DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;QueueEndpoint=http://azurite:10001/devstoreaccount1;

  azurite:
    image: mcr.microsoft.com/azure-storage/azurite
    volumes:
      - azurite:/data

volumes:
  azurite:

functions/Dockerfile

FROM --platform=linux/x86_64 mcr.microsoft.com/azure-functions/python:4-python3.11

ENV AzureWebJobsScriptRoot=/home/site/wwwroot \
    AzureFunctionsJobHost__Logging__Console__IsEnabled=true \
    AzureWebJobsFeatureFlags=EnableWorkerIndexing

RUN pip install azure-functions

COPY . /home/site/wwwroot

functions/host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

※ host.json は func init --docker --python で自動生成されたやつそのままで、手を加えていない

functions/function_app.py

from logging import getLogger

logger = getLogger(__name__)

import azure.functions as func

app = func.FunctionApp()


@app.queue_trigger(
    arg_name="payload",
    queue_name="myqueue",
    connection="AZURE_QUEUE_STORAGE",
)
def queue_example(payload: func.QueueMessage) -> None:
    message = payload.get_body().decode()
    logger.info(f"received message: {message}")

キューを投げる

compose.yml に別のサービスを生やすなどして、そこからキューを投げていきます。(※ azure-storage-queue パッケージのインストールが必要)

from base64 import b64encode
from azure.storage.queue import QueueClient
from azure.core.exceptions import ResourceExistsError

connect_str = "DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;QueueEndpoint=http://azurite:10001/devstoreaccount1;"


def get_or_create_queue(queue_name: str) -> QueueClient:
    queue = QueueClient.from_connection_string(connect_str, queue_name)
    try:
        queue.create_queue()
    except ResourceExistsError:
        pass
    return queue


def enqueue_message(queue_name: str, message: str):
    payload = b64encode(message.encode()).decode()
    get_or_create_queue(queue_name).send_message(payload)


enqueue_message("myqueue", "わいわい")

これを実行したところ、ログにメッセージが出たので無事に受け取れたっぽいです。

functions-1      | info: Function.queue_example.User[0]
functions-1      |       received message: わいわい
functions-1      | info: Function.queue_example[2]
functions-1      |       Executed 'Functions.queue_example' (Succeeded, Id=e766166d-4419-4abf-80e8-03699eef7e54, Duration=21ms)

ハマったところ

Apple シリコン Mac (ARM) だと Azure Functions 用のベースイメージが動かなくて、Rosetta を有効にして x86_64 をエミュレーションする必要があった
(未解決) Azure Functions のイメージが SIGINT で止まってくれなくて困る (init=true してもダメだった)
(未解決) local.settings.json に書いた設定をなぜか一切読み込んでくれないので、各種設定は Dockerfile や compose.yml に書いて環境変数として渡すことでなんとかした
@app.queue_trigger の引数 connection には接続文字列を直接渡すのではなく、環境変数名を渡す必要があった (とてもわかりにくいのでやめてほしい)
環境変数 AzureWebJobsFeatureFlags に EnableWorkerIndexing を設定しないと関数を読み込んでくれなかった (ここの設定まだよくわかっていない)
Python の QueueClient クラスには「キューが作成済みか否か」を確認する関数は存在しないので、 create_queue() してみて ResourceExistsError を捕まえる必要があった (多言語の SDK には exists 関数があるっぽいのに...)
Azure Functions に渡したいメッセージは Base64 エンコードしないと受け取ってくれなかった