How to Mock an OTLP HTTP Endpoint in Python

This code implements a lightweight HTTP server that accepts OTLP/HTTP trace exports, stores spans by trace ID, and exposes them via a simple GET endpoint for debugging.

Easy Python 3.9+ Aug 9, 2026 Observability & SRE 13 views 0 copies

otlp http mock tracing observability server

Python code

42 lines

Python 3.9+

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from collections import defaultdict

class TraceHandler(BaseHTTPRequestHandler):
    traces = defaultdict(list)

    def do_POST(self):
        if self.path == "/v1/traces":
            length = int(self.headers.get("Content-Length", 0))
            payload = json.loads(self.rfile.read(length))
            for resource_spans in payload.get("resourceSpans", []):
                for scope_spans in resource_spans.get("scopeSpans", []):
                    for span in scope_spans.get("spans", []):
                        trace_id = span.get("traceId")
                        self.traces[trace_id].append(span)
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{}')
        else:
            self.send_response(404)
            self.end_headers()

    def do_GET(self):
        if self.path == "/traces":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(dict(self.traces), indent=2).encode())
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, format, *args):
        pass

if __name__ == "__main__":
    server = ThreadingHTTPServer(("127.0.0.1", 4318), TraceHandler)
    print("OTLP HTTP mock listening on port 4318")
    print("POST /v1/traces to export, GET /traces to view collected spans")
    server.serve_forever()

Output

stdout

OTLP HTTP mock listening on port 4318
POST /v1/traces to export, GET /traces to view collected spans

How it works

This script runs a threaded HTTP server on 127.0.0.1:4318 — the standard OTLP HTTP port for traces. TraceHandler.do_POST reads the JSON payload and recursively extracts every span from the resourceSpans and scopeSpans nesting, grouping them by traceId in a class-level defaultdict(list). The GET /traces endpoint returns the accumulated spans as JSON, which lets you inspect exported trace data without a real collector. ThreadingHTTPServer handles concurrent POST requests, and log_message is silenced to avoid noisy console output.

Common mistakes

Forgetting to handle nested `scopeSpans` — the OTLP structure is `resourceSpans` > `scopeSpans` > `spans`.
Not sending a `Content-Type` header in the POST response, which can confuse some clients.
Using a single-threaded `HTTPServer` under load, causing delays when multiple spans arrive concurrently.
Assuming the payload is always JSON — OTLP can also be protobuf-based, which this mock does not handle.

Variations

Use `socketserver.TCPServer` with `allow_reuse_address = True` to avoid 'Address already in use' restart issues.
Extend the handler to log spans to a file or forward them to a real collector for end-to-end testing.

Real-world use cases

When developing an OpenTelemetry exporter, test it against a local mock to verify payload structure without spinning up a full collector.
In CI, run this mock to inspect whether services emit the expected spans and attributes during integration tests.
For debugging trace sampling or batching logic, collect spans locally and review them via the GET endpoint.

How to Mock an OTLP HTTP Endpoint in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Keep learning

Tutorials

Quizzes