Python Decorator for Azure App Insights

Image by Andrea Piacquadio from pexels.com

In enterprise solutions, we must create the log (traces) and performance tracking entries. This can be a complicated task, especially in a multi-service cloud environment. In this blog, we will show you how to do this with Python decorator writing traces to Azure Application Insights.

Azure Application Insights is the Azure de facto resource for telemetry, monitoring, observability. Traces, exceptions, event logs, etc. are captured in App Insights. And, it provides a very comprehensive query language, Kusto Query Language (KQL).

Python is the primary programming language for Azure Machine Learning (AML). There are a lot of samples and documents on Python SDK for AML.  Even Guido van Rossum came out of retirement to join Microsoft. (Creator of Python joins Microsoft, prompting speculation)

With these (Python & Azure App Insights) said, let's get to the core of this blog.

Here we have a very tiny Azure Function written in Python. And, we shall use it to illustrate how we use Python decorator to add traces in Azure App Insights.

import azure.functions as func

def func1(val: str):
    return val.upper()


def func2():
    print("does something")


def main(req: func.HttpRequest) -> func.HttpResponse:
    val = func1("hello")
    func2()
    return func.HttpResponse(
        f"{val}. This HTTP triggered function executed successfully.")

I want to add traces for func1 and func2 without having to inject any code into them. And, Python decorator is good for handling this situation. The next code snippet illustrates this.

import azure.functions as func
from ..logger import observe

@observe({"data_processing": "step1"})
def func1(val: str):
    return val.upper()

@observe({"data_processing": "step2"})
def func2():
    print("does something")


def main(req: func.HttpRequest) -> func.HttpResponse:
    val = func1("hello")
    func2()
    return func.HttpResponse(
        f"{val}. This HTTP triggered function executed successfully.")

Before we get into the code (from ..logger import observe) for the decorator. {"data_processing": "step1"} and {"data_processing": "step2"} tell the decorator to include more information in the trace. On top of this, I want to add more statistics to the traces. These statistics can only be collected after running the function. The next set of code snippets shows how we handle this.

import azure.functions as func
from ..logger import observe

@observe({"data_processing": "step1"})
def func1(val: str):
    return {
        "logging_dimensions": {
            "data_size": 1000,
            "complete": True,
        },
        "result": val.upper(),
    }

@observe({"data_processing": "step2"})
def func2():
    print("does something")
    # decorator is ok with no return result.

def main(req: func.HttpRequest) -> func.HttpResponse:
    val = func1("hello")
    func2()
    return func.HttpResponse(
        f"{val}. This HTTP triggered function executed successfully.")

We modified func1 to return a dictionary that contains the statistics (logging_dimensions to be added to the traces).

To summary what we intend to do.

1. decorator adds a trace to Azure App Insights before the decorated function is called. Having the following information
1. function name and the list of parameters passed into it.
2. extra information from the decorator.
2. decorator adds a trace to Azure App Insights after the decorated function is called. Having the following information
1. function name and the list of parameters passed into it.
2. extra information from the decorator.
3. the duration to execute the function.
4. the statistics (logging_dimensions) returned by the function (if any)
3. decorator adds an error lo g to Azure App Insights if the decorated function raises an exception. Having the following information
1. function name and the list of parameters passed into it.
2. extra information from the decorator.
3. exception message

Below is the logger.py implementation. It implements all the above requirements (see def observe). 

The init function needs to be called once to add App Insights log handler to the Python logging system. The APP_INSIGHT_CONNECTION_STR environment parameter value needs to be set. In Azure Function App, this is configured in the Function Configuration.

import logging
import os
import time

from opencensus.ext.azure.log_exporter import AzureLogHandler

initialized = False


def init():
    """Initialize logger for Azure Application Insights.

    This initialization steps will only occur once.
    """
    global initialized

    if not initialized:
        initialized = True

        conn_str = os.getenv("APP_INSIGHT_CONNECTION_STR")
        handler = AzureLogHandler(connection_string=conn_str)

        formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
        handler.setFormatter(formatter)
        root = logging.getLogger()
        root.setLevel(os.environ.get("LOG_LEVEL", "INFO"))  # LOG_LEVEL is externalized
        root.addHandler(handler)


def observe(dimensions: dict = {}):
    """Observe function call.

    Args:
        dimensions (dict, optional): Extra dimensions to be added to log.
        Defaults to {}.

    Returns:
        function: Decorated function
    """
    init()

    def _func(func):
        def _wrapper(*args, **kwargs):
            start_time = time.time()

            # add log entry before calling the function
            custom_dims = {"parameters": str({"args": args, "kwargs": kwargs})}
            custom_dims.update(dimensions)
            logging.info(
                f"calling function, {func.__name__}",
                extra={"custom_dimensions": custom_dims},
            )

            # call the function
            try:
                result = func(*args, **kwargs)
                duration = time.strftime(
                    "%H:%M:%S",
                    time.gmtime(time.time() - start_time),
                )

                # if the result of function is a dictionary and contains
                # a logging_dimensions, add it to log dimension, and
                # set the result to the value of "result" key of the dictionary
                if result and isinstance(result, dict):
                    logging_dimensions = result.get("logging_dimensions")
                    if logging_dimensions:
                        custom_dims.update(
                            {
                                "duration": duration,
                            }
                        )
                        custom_dims.update(logging_dimensions)
                        result = result.get("result")

                # add log entry after calling the function
                logging.info(
                    f"completed function call, {func.__name__}",
                    extra={"custom_dimensions": custom_dims},
                )

                return result
            except Exception as e:
                custom_dims.update({"error": str(e)})
                logging.error(
                    f"error in function call, {func.__name__}",
                    extra={"custom_dimensions": custom_dims},
                )
                raise e

        return _wrapper

    return _func

We have a dependency in our implementation which is opencensus-ext-azure.

We have shown that we are able to use a Python decorator to simplify creation of traces in App Insights. Most importantly (which is also our main objective), the decorated function remains almost unchanged.