Skip to content

Instantly share code, notes, and snippets.

@jasonnerothin

jasonnerothin/https_tracer.py

Created November 6, 2025 18:27
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save jasonnerothin/e5f4a9844067ed9717131b7d97c6924e to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save jasonnerothin/e5f4a9844067ed9717131b7d97c6924e to your computer and use it in GitHub Desktop.
Download ZIP
stack agnostic http trace provider
Raw
https_tracer.py
"""
HTTPS Tracer - Distributed tracing for HTTPS client/server communication
Client-side: Uses standard requests library (no framework dependency)
Server-side: Provides generic patterns adaptable to any HTTP framework
"""
from typing import Optional, Dict, Any, Callable
import requests
from opentelemetry import trace
from opentelemetry.trace import SpanKind
from opentelemetry.propagate import inject, extract
from common.threadsafe_otlp_traces import (
get_tracer,
set_http_client_attributes,
set_http_server_attributes,
record_event,
record_exception,
set_span_error
)
# ============================================================================
# CLIENT SIDE - HTTP Requests with Trace Context
# ============================================================================
def make_traced_request(
tracer: trace.Tracer,
method: str,
url: str,
span_name: Optional[str] = None,
peer_service: Optional[str] = None,
json: Optional[Dict[str, Any]] = None,
data: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
timeout: int = 30,
**kwargs
) -> requests.Response:
"""
Make an HTTP request with trace context propagation.
Creates a CLIENT span and injects trace context into HTTP headers.
Args:
tracer: OpenTelemetry tracer instance
method: HTTP method (GET, POST, etc.)
url: Target URL
span_name: Optional span name (default: "{method} {url}")
peer_service: Target service name (for service map)
json: JSON payload
data: Request body data
headers: Additional HTTP headers
timeout: Request timeout in seconds
**kwargs: Additional arguments passed to requests.request()
Returns:
Response object
Raises:
requests.RequestException: If request fails
Example:
from common.https_tracer import make_traced_request
response = make_traced_request(
tracer=tracer,
method="POST",
url="https://ares:8443/pray",
peer_service="ares",
json={"status": "ready"},
span_name="send_prayer"
)
"""
span_name = span_name or f"{method} {url}"
with tracer.start_as_current_span(
span_name,
kind=SpanKind.CLIENT
) as span:
try:
# Set HTTP client attributes
set_http_client_attributes(
span,
method=method,
url=url,
peer_service=peer_service
)
# Prepare headers with trace context
request_headers = headers.copy() if headers else {}
inject(request_headers)
record_event(span, "request_started", {
"method": method,
"url": url
})
# Make the request
response = requests.request(
method=method,
url=url,
json=json,
data=data,
headers=request_headers,
timeout=timeout,
**kwargs
)
# Set response status code
span.set_attribute("http.status_code", response.status_code)
span.set_attribute("http.response.status_code", response.status_code)
# Mark as error if status code >= 400
if response.status_code >= 400:
set_span_error(span, f"HTTP {response.status_code}")
record_event(span, "request_completed", {
"status_code": response.status_code,
"response_size": len(response.content)
})
return response
except Exception as e:
record_exception(span, e)
raise
class TracedHTTPClient:
"""
HTTP client wrapper with automatic tracing.
Provides a simple interface for making traced HTTP requests.
"""
def __init__(
self,
tracer: trace.Tracer,
service_name: str,
default_timeout: int = 30
):
"""
Initialize traced HTTP client.
Args:
tracer: OpenTelemetry tracer instance
service_name: Name of the service making requests
default_timeout: Default request timeout
"""
self.tracer = tracer
self.service_name = service_name
self.default_timeout = default_timeout
def get(self, url: str, peer_service: Optional[str] = None, **kwargs) -> requests.Response:
"""Make a traced GET request."""
return make_traced_request(
tracer=self.tracer,
method="GET",
url=url,
peer_service=peer_service,
timeout=kwargs.pop('timeout', self.default_timeout),
**kwargs
)
def post(
self,
url: str,
peer_service: Optional[str] = None,
json: Optional[Dict[str, Any]] = None,
**kwargs
) -> requests.Response:
"""Make a traced POST request."""
return make_traced_request(
tracer=self.tracer,
method="POST",
url=url,
peer_service=peer_service,
json=json,
timeout=kwargs.pop('timeout', self.default_timeout),
**kwargs
)
def put(
self,
url: str,
peer_service: Optional[str] = None,
json: Optional[Dict[str, Any]] = None,
**kwargs
) -> requests.Response:
"""Make a traced PUT request."""
return make_traced_request(
tracer=self.tracer,
method="PUT",
url=url,
peer_service=peer_service,
json=json,
timeout=kwargs.pop('timeout', self.default_timeout),
**kwargs
)
def delete(self, url: str, peer_service: Optional[str] = None, **kwargs) -> requests.Response:
"""Make a traced DELETE request."""
return make_traced_request(
tracer=self.tracer,
method="DELETE",
url=url,
peer_service=peer_service,
timeout=kwargs.pop('timeout', self.default_timeout),
**kwargs
)
# ============================================================================
# SERVER SIDE - Generic Trace Context Extraction
# ============================================================================
def extract_trace_context_from_headers(headers: Dict[str, str]) -> Optional[trace.Context]:
"""
Extract trace context from HTTP headers (framework-agnostic).
Args:
headers: HTTP request headers as dict
Returns:
OpenTelemetry Context object, or None if extraction fails
Example:
# With any HTTP framework, extract headers as dict:
headers = {
'traceparent': '00-abc123...',
'content-type': 'application/json',
# ... other headers
}
ctx = extract_trace_context_from_headers(headers)
with tracer.start_as_current_span("handler", context=ctx, kind=SpanKind.SERVER):
# Handle request
pass
"""
try:
# OpenTelemetry extract works with dict-like objects
ctx = extract(headers)
return ctx
except Exception:
return None
def create_server_span(
tracer: trace.Tracer,
span_name: str,
method: str,
path: str,
headers: Dict[str, str],
extract_context: bool = True
):
"""
Create a SERVER span with trace context extraction (framework-agnostic).
Use this as a context manager in your HTTP handler.
Args:
tracer: OpenTelemetry tracer instance
span_name: Name for the span
method: HTTP method (GET, POST, etc.)
path: Request path
headers: Request headers as dict
extract_context: If True, extract parent context from headers
Returns:
Context manager yielding the span
Example:
# In your HTTP handler:
def handle_request(method, path, headers, body):
with create_server_span(
tracer,
"handle_pray",
method,
path,
headers
) as span:
# Process request
result = process_pray(body)
# Set response status
span.set_attribute("http.status_code", 200)
return result
"""
# Extract context if requested
context = None
if extract_context:
context = extract_trace_context_from_headers(headers)
# Return context manager
return tracer.start_as_current_span(
span_name,
context=context,
kind=SpanKind.SERVER
)
class ServerSpanHelper:
"""
Helper class for managing server spans with proper attributes.
Use this in your HTTP server handler to create properly configured spans.
"""
def __init__(
self,
tracer: trace.Tracer,
span_name: str,
method: str,
path: str,
headers: Dict[str, str]
):
"""
Initialize server span helper.
Args:
tracer: OpenTelemetry tracer instance
span_name: Name for the span
method: HTTP method
path: Request path
headers: Request headers as dict
"""
self.tracer = tracer
self.span_name = span_name
self.method = method
self.path = path
self.headers = headers
self.span = None
def __enter__(self):
"""Start the span."""
# Extract context
context = extract_trace_context_from_headers(self.headers)
# Create span
self.span = self.tracer.start_span(
self.span_name,
context=context,
kind=SpanKind.SERVER
)
# Set initial attributes
set_http_server_attributes(
self.span,
method=self.method,
route=self.path,
status_code=200 # Default, update later
)
record_event(self.span, "request_received", {
"method": self.method,
"path": self.path
})
return self.span
def __exit__(self, exc_type, exc_val, exc_tb):
"""End the span."""
if self.span:
if exc_val:
record_exception(self.span, exc_val)
self.span.set_attribute("http.status_code", 500)
record_event(self.span, "request_completed")
self.span.end()
# ============================================================================
# USAGE EXAMPLES FOR DIFFERENT SCENARIOS
# ============================================================================
"""
EXAMPLE 1: Simple HTTP Server Handler (Generic)
------------------------------------------------
def handle_pray_request(method, path, headers, body):
'''Handle incoming /pray request'''
with create_server_span(
tracer,
"handle_prayer",
method,
path,
headers
) as span:
try:
# Parse request
data = json.loads(body)
# Process
result = process_prayer(data)
# Set success status
span.set_attribute("http.status_code", 200)
return 200, json.dumps(result)
except Exception as e:
record_exception(span, e)
span.set_attribute("http.status_code", 500)
return 500, json.dumps({"error": str(e)})
EXAMPLE 2: With http.server (Python standard library)
------------------------------------------------------
from http.server import BaseHTTPRequestHandler, HTTPServer
from common.https_tracer import extract_trace_context_from_headers
class TracedHTTPHandler(BaseHTTPRequestHandler):
def do_POST(self):
# Extract headers
headers = dict(self.headers)
# Create server span
with create_server_span(
tracer,
f"POST {self.path}",
"POST",
self.path,
headers
) as span:
try:
# Read body
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length)
# Process request
result = handle_request(body)
# Send response
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(result).encode())
span.set_attribute("http.status_code", 200)
except Exception as e:
record_exception(span, e)
self.send_error(500, str(e))
EXAMPLE 3: With Flask (Optional - if you add Flask later)
----------------------------------------------------------
# Save this as common/flask_tracer.py if you add Flask later
from flask import Flask, request
from functools import wraps
def traced_route(tracer, span_name=None):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
headers = dict(request.headers)
with create_server_span(
tracer,
span_name or func.__name__,
request.method,
request.path,
headers
) as span:
result = func(*args, **kwargs)
# Extract status code if tuple returned
status_code = 200
if isinstance(result, tuple) and len(result) >= 2:
status_code = result[1]
span.set_attribute("http.status_code", status_code)
return result
return wrapper
return decorator
# Usage:
# @app.route('/pray', methods=['POST'])
# @traced_route(tracer, "handle_prayer")
# def pray():
# return {"status": "ok"}, 200
EXAMPLE 4: With aiohttp (Async)
--------------------------------
from aiohttp import web
from common.https_tracer import extract_trace_context_from_headers
async def handle_pray(request):
# Extract headers
headers = dict(request.headers)
# Create span
with create_server_span(
tracer,
"handle_prayer",
request.method,
request.path,
headers
) as span:
try:
data = await request.json()
result = await process_prayer(data)
span.set_attribute("http.status_code", 200)
return web.json_response(result)
except Exception as e:
record_exception(span, e)
span.set_attribute("http.status_code", 500)
return web.json_response({"error": str(e)}, status=500)
EXAMPLE 5: Manual Span Management (Full Control)
-------------------------------------------------
def handle_request(method, path, headers, body):
# Extract context
ctx = extract_trace_context_from_headers(headers)
# Start span manually
span = tracer.start_span(
"handle_request",
context=ctx,
kind=SpanKind.SERVER
)
try:
# Set attributes
set_http_server_attributes(span, method, path, 200)
# Process
result = process_request(body)
# Success
record_event(span, "request_completed")
return 200, result
except Exception as e:
record_exception(span, e)
span.set_attribute("http.status_code", 500)
return 500, {"error": str(e)}
finally:
span.end()
"""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment