🤖

Python Function App で別スレッドでログを出力する方法

2023/12/27に公開

概要

Python の Function App を動作させていて、別スレッドからログ出力をしようとしたのにうまく出力されないというケースがあるかと思います。
少し調べてみたのですが、そのような場合に解決する方法がわかりましたので、参考迄に記載しておきます。

詳細

これが対処を施す前のコードです。
別スレッドを作成し、test001 メソッドにおいて、logging.info('Logging in another thread.') を記述していますが、このコードではログが出力されません。

変更前 (ログが出ないコード)

import logging
import threading
import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:

    print(threading.currentThread().getName())
    logging.info('Python HTTP trigger function processed a request.')
    logging.info('logging in main')

    t = threading.Thread(target=test001, args=())
    t.start()

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
    else:
        return func.HttpResponse(
             "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

def test001():
    #context.thread_local_storage.invocation_id = context.invocation_id
    print(threading.currentThread().getName())
    logging.info('Logging in another thread.')

出力結果は以下の通りです。Logging in another thread.はないです。

以下のように、関数のシグネチャに context を含めて、それを test001 メソッドにも引き渡してみます。また、test001 メソッドにて context.thread_local_storage.invocation_id = context.invocation_idを追加します。

変更後 (ログ出力に成功するコード)

import logging
import threading
import azure.functions as func

def main(req: func.HttpRequest,context) -> func.HttpResponse:

    print(threading.currentThread().getName())
    logging.info('Python HTTP trigger function processed a request.')
    logging.info('logging in main')

    t = threading.Thread(target=test001, args=(context,))
    t.start()

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.")
    else:
        return func.HttpResponse(
             "This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

def test001(context):
    context.thread_local_storage.invocation_id = context.invocation_id
    print(threading.currentThread().getName())
    logging.info('Logging in another thread.')

すると以下の通り、'Logging in another thread.' も出力されるようになりました。

まとめ

他の要因によって別スレッドのログが出力されない場合もあるかと思いますが、上記を実施していない場合には今回記載した内容が要因である可能性が高いと思いますので、一旦試してみるといいかと思います。

参考資料:

概要

詳細

まとめ

Discussion