tracing

Module

Distributed tracing middleware for Starlette-based applications. This middleware automatically extracts distributed tracing headers from incoming HTTP requests and makes them available to the Splunk AO logger within request handlers. Works with any ASGI framework built on Starlette:

FastAPI
Starlette
Any other Starlette-based framework

Example usage with FastAPI:

from fastapi import FastAPI
from splunk_ao.middleware import TracingMiddleware, get_request_logger

app = FastAPI()
app.add_middleware(TracingMiddleware)

@app.post("/process")
async def process_request(data: dict):
    # Logger automatically continues the distributed trace
    logger = get_request_logger()
    logger.add_workflow_span(input=str(data), name="process_workflow")
    # ... process request ...
    logger.conclude(output="done")
    return {"status": "success"}

Example usage with Starlette:

from starlette.applications import Starlette
from starlette.routing import Route
from splunk_ao.middleware import TracingMiddleware, get_request_logger

async def homepage(request):
    logger = get_request_logger()
    logger.add_workflow_span(input="homepage", name="homepage_handler")
    logger.conclude(output="success")
    return {"status": "ok"}

app = Starlette(
    routes=[Route("/", homepage)],
    middleware=[TracingMiddleware]
)

TracingMiddleware

Middleware that extracts distributed tracing headers from incoming requests. This middleware looks for the following headers in incoming HTTP requests:

Splunk-AO-Trace-ID: The root trace ID
Splunk-AO-Parent-ID: The parent span/trace ID to attach to

These values are stored in context variables, making them available to request handlers via the get_request_logger() function. The middleware is compatible with FastAPI and any Starlette-based framework. Note: Project and log_stream are configured per service via environment variables (SPLUNK_AO_PROJECT and SPLUNK_AO_AGENT_STREAM). They are not propagated via headers, following standard distributed tracing patterns.

dispatch

async def dispatch(self,
                   request: Request,
                   call_next: RequestResponseEndpoint) -> Response

Process the request and extract tracing headers. Arguments

request (Request): The incoming HTTP request
call_next (RequestResponseEndpoint): The next middleware or route handler

Returns

Response: The HTTP response

get_request_logger

def get_request_logger() -> SplunkAOLogger

Get a request-scoped SplunkAOLogger configured for distributed mode. Distributed mode enables the legacy trace_id/span_id continuation parameters. Completed spans use the same scheduled OTLP batch export as batch mode. This function should be called within a request handler after the TracingMiddleware has been registered. It creates a new SplunkAOLogger instance per request that automatically continues the distributed trace from the upstream service. The logger is configured using trace context extracted by the middleware:

Splunk-AO-Trace-ID: Root trace ID
Splunk-AO-Parent-ID: Parent span/trace ID to attach to

Project and log_stream are configured per service via environment variables (SPLUNK_AO_PROJECT and SPLUNK_AO_AGENT_STREAM), not propagated via headers, following standard distributed tracing patterns. If no tracing headers were present in the request, a regular logger is returned (using SPLUNK_AO_PROJECT and SPLUNK_AO_AGENT_STREAM env vars). Note: This creates a new logger per request, unlike the decorator’s get_logger_instance() which uses a singleton pattern. Returns

SplunkAOLogger: A logger instance configured for the current request’s trace context

Examples

@app.post("/process")
async def process_request(data: dict):
    logger = get_request_logger()

    # This span will be attached to the distributed trace
    logger.add_workflow_span(input=str(data), name="process_workflow")
    result = await process(data)
    logger.conclude(output=str(result))

    logger.flush()
    logger.terminate()

    return {"result": result}

@app.post("/retrieve")
async def retrieve_endpoint(query: str):
    # Get logger with trace context from upstream service
    logger = get_request_logger()

    # If trace context exists, this creates a workflow span
    # Otherwise, it starts a new trace
    if logger.trace_id:
        logger.add_workflow_span(input=query, name="retrieval_service")
    else:
        logger.start_trace(input=query, name="retrieval_service")

    results = retrieve(query, logger)

    logger.conclude(output=str(results))
    logger.flush()
    logger.terminate()

    return {"results": results}

Overview

Python SDK

API

Module

TracingMiddleware

dispatch

get_request_logger

​Module

​TracingMiddleware

​dispatch

​get_request_logger

Module

TracingMiddleware

dispatch

get_request_logger