Quick-Start to Integrating OpenTelemetry with FastAPI (Part 1)

No long useless rants here. Just code. Let’s go!! (That snack saw me through some of the code below, so it deserves a place here!)

from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler 
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, SimpleLogRecordProcessor, ConsoleLogExporter
from opentelemetry.instrumentation.logging import LoggingInstrumentor
from opentelemetry._logs import (
    SeverityNumber,
    get_logger,
    get_logger_provider,
    std_to_otel,
    set_logger_provider
)

# The default Otel SDK completely ignores formatters when outputting the message being logged.
# We overcome this by creating our own LoggingHandler class which respects formatters.
class FormattedLoggingHandler(LoggingHandler):
    def emit(self, record: logging.LogRecord) -> None:
        msg = self.format(record)
        record.msg = msg
        record.args = None
        self._logger.emit(self._translate(record))

def otel_get_env_vars():
    otel_http_headers = {}
    try:
        decoded_http_headers = os.getenv("OTEL_ENDPOINT_HTTP_HEADERS", "")
        key_values = decoded_http_headers.split(",")
        for key_value in key_values:
            key, value = key_value.split("=")
            otel_http_headers[key] = value
    except Exception as e:
        print(f"Error parsing OTEL_ENDPOINT_HTTP_HEADERS: {str(e)}")
    otel_endpoint_url = os.getenv("OTEL_ENDPOINT_URL", None)
    
    return otel_endpoint_url, otel_http_headers

def otel_trace_init():
    trace.set_tracer_provider(
       TracerProvider(
           resource=Resource.create({}),
       ),
    )
    if DEBUG_LOG_OTEL_TO_PROVIDER:
        otel_endpoint_url, otel_http_headers = otel_get_env_vars()
        otlp_span_exporter = OTLPSpanExporter(endpoint=otel_endpoint_url,headers=otel_http_headers)
        trace.get_tracer_provider().add_span_processor(BatchSpanProcessor(otlp_span_exporter))
    if DEBUG_LOG_OTEL_TO_CONSOLE:
        trace.get_tracer_provider().add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))

def otel_logging_init():
    # ------------Logging
    # Set logging level
    # CRITICAL = 50
    # ERROR = 40
    # WARNING = 30
    # INFO = 20
    # DEBUG = 10
    # NOTSET = 0
    # default = WARNING
    log_level = str(os.getenv("OTEL_PYTHON_LOG_LEVEL", "INFO")).upper()
    if (log_level == "CRITICAL"):
        log_level = logging.CRITICAL
        print(f"Using log level: CRITICAL / {log_level}")
    elif (log_level == "ERROR"):
        log_level = logging.ERROR
        print(f"Using log level: ERROR / {log_level}")
    elif (log_level == "WARNING"):
        log_level = logging.WARNING
        print(f"Using log level: WARNING / {log_level}")
    elif (log_level == "INFO"):
        log_level = logging.INFO
        print(f"Using log level: INFO / {log_level}")
    elif (log_level == "DEBUG"):
        log_level = logging.DEBUG
        print(f"Using log level: DEBUG / {log_level}")
    elif (log_level == "NOTSET"):
        log_level = logging.INFO
        print(f"Using log level: NOTSET / {log_level}")

    # ------------ Opentelemetry loging initialization

    logger_provider = LoggerProvider(
        resource=Resource.create({})
    )
    set_logger_provider(logger_provider)
    if DEBUG_LOG_OTEL_TO_CONSOLE:
        console_log_exporter = ConsoleLogExporter()
        logger_provider.add_log_record_processor(SimpleLogRecordProcessor(console_log_exporter))
    if DEBUG_LOG_OTEL_TO_PROVIDER:
        otel_endpoint_url, otel_http_headers = otel_get_env_vars()
        otlp_log_exporter = OTLPLogExporter(endpoint=otel_endpoint_url,headers=otel_http_headers)
        logger_provider.add_log_record_processor(BatchLogRecordProcessor(otlp_log_exporter))

    otel_log_handler = FormattedLoggingHandler(logger_provider=logger_provider)

    # This has to be called first before logger.getLogger().addHandler() so that it can call logging.basicConfig first to set the logging format
    # based on the environment variable OTEL_PYTHON_LOG_FORMAT
    LoggingInstrumentor().instrument()
    logFormatter = logging.Formatter(os.getenv("OTEL_PYTHON_LOG_FORMAT", None))
    otel_log_handler.setFormatter(logFormatter)
    logging.getLogger().addHandler(otel_log_handler)


def fastapi_init():
    app = FastAPI(
        docs_url = None,
        redoc_url = None,
        title = "Placeholder",
        version = "Placeholder",
        description = "Placeholder"
    )
    
    @app.get("/redoc", include_in_schema=False)
    def overridden_redoc():
        return get_redoc_html(openapi_url="/openapi.json", title="Human Managed", redoc_favicon_url="https://invicta.io/favicon.ico")

    @app.get("/docs", include_in_schema=False)
    def overridden_swagger():
        return get_swagger_ui_html(openapi_url="/openapi.json", title="Human Managed", swagger_favicon_url="https://invicta.io/favicon.ico")

    # For otel compatibility. It seems HTTPException is only trapped by Otel with otel.status_description
    # set to the string "HTTPException:", which is not very informative for us. Here, we implement a custom
    # error handler for HTTPException class.
    @app.exception_handler(StarletteHTTPException)
    async def http_exception_handler(request, exc):
        current_span = trace.get_current_span()
        if (current_span is not None) and (current_span.is_recording()):
            current_span.set_attributes(
                {
                    "http.status_text": str(exc.detail),
                    "otel.status_description": f"{exc.status_code} / {str(exc.detail)}",
                    "otel.status_code": "ERROR"
                }
            )
        return PlainTextResponse(json.dumps({ "detail" : str(exc.detail) }), status_code=exc.status_code)
    FastAPIInstrumentor.instrument_app(app)        
    return app

DEBUG_LOG_OTEL_TO_CONSOLE = os.getenv("DEBUG_LOG_OTEL_TO_CONSOLE", 'False').lower() == 'true'
DEBUG_LOG_OTEL_TO_PROVIDER = os.getenv("DEBUG_LOG_OTEL_TO_PROVIDER", 'False').lower() == 'true'
otel_trace_init()
otel_logging_init()
app = fastapi_init()
app.include_router(apirouter.api_router_v1)
app.include_router(apirouter.invict_api_router_v1)
app.include_router(apirouter.api_router)
app.add_middleware(GZipMiddleware, minimum_size=1000000)

And there you have it. This is what’s in main.py in my implementation (ok, I lied, there are a few more bits of missing code removed because they aren’t relevant to the otel + FastAPI discussion :P ). Short & sweet!

Before you go jumping into this code, I recommend you read the 2nd part of this series as well. It expands on the code above.