azure-functionsfastapi

FastAPI logging in routes in Azure Functions

I noticed that any logging.info('stuff happened in a route') when in a route, it wasn't coming through to my Application Insights log. The instances of logging.info('stuff happened in entrypoint') were. I found this bit of advice but the solution is a bit vague and seemingly specific to running uvicorn and I don't know how to apply it to Azure.

For reference here are snippets from my __init__.py

import logging
import azure.functions as func
from fastapi import FastAPI
from .routes import route1

app = FastAPI()
app.include_router(route1.router)

@app.get("/api/test"):
def test():
    logging.info("entry test")
    return "test"

async def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return await func.AsgiMiddleware(app).handle_async(req, context)

Those entries make it into my logs

Then in route1 I have

from fastapi import APIRouter
import logging

router=APIRouter(prefix="/api/route1")

@router.get("/test")
def test():
    logging.info("resp route test")
    return "route test"

The route itself works, that is to say if I navigate to 'api/route1/test' I get "resp route test" in my browser but then when I check the logs, I don't have the "route test" entry.

I tried adding logging.config.dictConfig({"disable_existing_loggers": False, "version": 1}) to the top of my route1.py file but it didn't help.

Solution

  • I just solved this problem and discovered a few quirks with logging on Azure Function Apps / Application Insights

    In order to fix this you need to first pass 1) the Azure context object and 2) the parent thread's invocation_id down to your endpoint, and then, from the created thread / endpoint, you need to set thread_local_storage.invocation_id to the parent's invocation_id.

    1. Prepare the endpoint to receive the context

    There is an implicit parameter for each endpoint called request. If you just define it as a parameter, you can use it in your endpoint and inspect it in a debugger.

    from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    logging.info("hello world!") # remember print() doesn't work

    Inside that request object, there is a dictionary called scope.

    2a. AsgiFunctionApp / AsgiMiddleware

    If you're using azure.functions.AsgiFunctionApp or azure.function.AsgiMiddleware (of a certain version, not exactly sure, but more recent than 1.8 I think), the context's thread_local_storage and the parent thread's invocation_id are actually included in the scope for you.

    asgimiddleware

    In that case, you would do the following in your synchronous endpoint (remember, no need to do this in an async endpoint)

    from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    thread_local_storage = request.scope.get("azure_functions.thread_local_storage")
    if thread_local_storage is not None:
        parent_invocation_id = request.scope.get("azure_functions.invocation_id")
        thread_local_storage.invocation_id = parent_invocation_id
    logging.info("hello world!") # remember print() doesn't work

    2b. Custom middleware

    If you're using custom middleware, like this example here, you'll need to add the parent invocation_id and the context (or the thread_local_storage object) to the scope.

    Something like this: custom scope

    Then you would set it in your endpoint like so:

    from fastapi import Request

@app.get("/api/test")
def hello_world(request: Request):
    context_obj = request.scope.get("azure_context", {}).get("context_obj")
    if context_obj and hasattr(context_obj, "thread_local_storage"):
        invocation_id = request.scope.get("azure_context", {}).get("invocation_id")
        if invocation_id:
            context_obj.thread_local_storage.invocation_id = request.scope.get("azure_context", {}
            ).get("invocation_id")
    logging.info("hello world!") # remember print() doesn't work

    3. (Optional) Wrapper function

    It's annoying to include that code at the beginning of all your functions, so you can create a wrapper function for it (h/t Eugene Nikolaiev)

    from fastapi import Request
from functools import wraps

def logging_context_wrapper(func):
    """Passes invocation_id to the thread local context to enable proper Azure Functions logging.

    Can be applied to a sync handler which has request: Request parameter - to get the context from.
    """

    request_param = next(
        (param for (param, annotation) in func.__annotations__.items() if annotation is Request),
        None,
    )
    if not request_param:
        raise Exception("Function must have a request parameter of type Request.")

    @wraps(func)
    def wrapper(*args, **kwargs):
        request = kwargs[request_param]
        # if you were 2b from above, sub whatever code you used
        thread_local_storage = request.scope.get("azure_functions.thread_local_storage")
        if thread_local_storage is not None:
            thread_local_storage.invocation_id = request.scope.get("azure_functions.invocation_id")
        return func(*args, **kwargs)

    return wrapper



@app.get("/api/test")
@logging_context_wrapper
def hello_world(request: Request):
    logging.info("hello world!") # remember print() doesn't work

    Summary

    Things I would for someone to explain