Debugging distributed systems with trace correlation, cross-service workflows, request replay, clock skew handling, and chaos engineering
Scope: Distributed tracing correlation, cross-service debugging, request replay, traffic shadowing, clock skew, chaos engineering, cascading failures
Lines: 440
Last Updated: 2025-10-26
Use this skill when:
Don't use for:
Unique challenges:
Layer 1: Request Tracing (follow single request)
├─ Trace ID correlation across services
├─ Span hierarchy (parent-child relationships)
└─ Timing analysis (latency breakdown)
Layer 2: Cross-Service Analysis (multi-request patterns)
├─ Request replay/shadowing
├─ Dependency mapping
└─ Circuit breaker states
Layer 3: System-Wide Behavior (emergent issues)
├─ Clock skew detection
├─ Distributed deadlock analysis
└─ Cascading failure patterns
Layer 4: Chaos Engineering (controlled failure injection)
├─ Network partition simulation
├─ Latency injection
└─ Service failure scenarios| Pattern | Symptom | Debugging Approach | |---------|---------|-------------------| | Cascading failure | One slow service causes timeouts everywhere | Trace latency, check circuit breakers | | Distributed deadlock | Services waiting on each other indefinitely | Analyze request dependencies, check for cycles | | Split brain | Multiple nodes think they're leader | Check consensus logs, clock skew | | Thundering herd | Simultaneous requests overwhelm service | Analyze request timestamps, check for jitter | | Data inconsistency | Different services see different data | Check event ordering, compare timestamps |
import uuid
from opentelemetry import trace
from opentelemetry.propagate import inject, extract
from fastapi import FastAPI, Request, HTTPException
import httpx
import structlog
app = FastAPI()
tracer = trace.get_tracer(__name__)
logger = structlog.get_logger()
# Service A: Gateway
@app.post("/api/orders")
async def create_order(request: Request, order_data: dict):
# Extract trace context from incoming request (if exists)
ctx = extract(request.headers)
with tracer.start_as_current_span("create_order", context=ctx) as span:
# Get trace ID for logging
trace_id = format(span.get_span_context().trace_id, '032x')
span_id = format(span.get_span_context().span_id, '016x')
log = logger.bind(trace_id=trace_id, span_id=span_id)
log.info("Order creation started", order_data=order_data)
# Propagate trace context to downstream services
headers = {}
inject(headers) # Injects traceparent, tracestate headers
try:
# Call Service B (Inventory)
async with httpx.AsyncClient() as client:
inventory_response = await client.post(
"http://inventory-service/api/reserve",
json={"product_id": order_data["product_id"]},
headers=headers # Propagate trace context
)
if inventory_response.status_code != 200:
log.error(
"Inventory reservation failed",
status_code=inventory_response.status_code,
response=inventory_response.text
)
raise HTTPException(
status_code=503,
detail="Inventory service unavailable"
)
# Call Service C (Payment)
payment_response = await client.post(
"http://payment-service/api/charge",
json={"amount": order_data["amount"]},
headers=headers # Same trace context
)
if payment_response.status_code != 200:
# Compensating transaction: release inventory
await client.post(
"http://inventory-service/api/release",
json={"product_id": order_data["product_id"]},
headers=headers
)
log.error("Payment failed, inventory released")
raise HTTPException(
status_code=402,
detail="Payment processing failed"
)
log.info("Order created successfully", trace_id=trace_id)
return {
"order_id": str(uuid.uuid4()),
"trace_id": trace_id,
"status": "success"
}
except httpx.RequestError as e:
log.error("HTTP request failed", error=str(e), exc_info=True)
span.record_exception(e)
raise HTTPException(status_code=503, detail="Service unavailable")
# Service B: Inventory Service
@app.post("/api/reserve")
async def reserve_inventory(request: Request, data: dict):
# Extract trace context from upstream service
ctx = extract(request.headers)
with tracer.start_as_current_span("reserve_inventory", context=ctx) as span:
trace_id = format(span.get_span_context().trace_id, '032x')
log = logger.bind(trace_id=trace_id, service="inventory")
log.info("Inventory reservation", product_id=data["product_id"])
# Simulate inventory check
if data["product_id"] == "out-of-stock":
log.error("Product out of stock", product_id=data["product_id"])
return {"status": "out_of_stock"}, 409
log.info("Inventory reserved", product_id=data["product_id"])
return {"status": "reserved"}
# Service C: Payment Service
@app.post("/api/charge")
async def charge_payment(request: Request, data: dict):
ctx = extract(request.headers)
with tracer.start_as_current_span("charge_payment", context=ctx) as span:
trace_id = format(span.get_span_context().trace_id, '032x')
log = logger.bind(trace_id=trace_id, service="payment")
log.info("Payment processing", amount=data["amount"])
# Simulate payment processing
if data["amount"] > 10000:
log.error("Payment amount exceeds limit", amount=data["amount"])
return {"status": "declined"}, 402
log.info("Payment successful", amount=data["amount"])
return {"status": "charged"}Query traces across services (Grafana Tempo):
# Find all spans for trace ID
{trace_id="a1b2c3d4e5f6g7h8"}
# Find traces with errors in any service
{status.code="ERROR"}
# Find slow traces (>1s)
{duration > 1s}
# Find traces involving specific service
{service.name="payment-service"}import json
import asyncio
from typing import Dict, Any
from datetime import datetime
import httpx
class RequestRecorder:
"""
Record requests for later replay.
"""
def __init__(self, storage_path: str = "/tmp/requests.jsonl"):
self.storage_path = storage_path
async def record_request(
self,
method: str,
url: str,
headers: Dict[str, str],
body: Any,
response_status: int,
response_body: Any
):
"""
Record request/response for replay.
"""
record = {
"timestamp": datetime.utcnow().isoformat(),
"request": {
"method": method,
"url": url,
"headers": dict(headers),
"body": body
},
"response": {
"status": response_status,
"body": response_body
}
}
with open(self.storage_path, "a") as f:
f.write(json.dumps(record) + "\n")
class RequestReplayer:
"""
Replay recorded requests against test environment.
"""
def __init__(self, storage_path: str = "/tmp/requests.jsonl"):
self.storage_path = storage_path
async def replay_all(self, target_url: str):
"""
Replay all recorded requests.
"""
with open(self.storage_path, "r") as f:
for line in f:
record = json.loads(line)
await self.replay_request(record, target_url)
async def replay_request(self, record: Dict, target_url: str):
"""
Replay single request.
"""
req = record["request"]
# Replace production URL with target URL
url = req["url"].replace("https://api.example.com", target_url)
async with httpx.AsyncClient() as client:
try:
response = await client.request(
method=req["method"],
url=url,
headers=req["headers"],
json=req["body"]
)
# Compare response with recorded response
original_status = record["response"]["status"]
if response.status_code != original_status:
print(f"MISMATCH: Expected {original_status}, got {response.status_code}")
print(f"URL: {url}")
print(f"Response: {response.text}")
except Exception as e:
print(f"ERROR replaying request: {e}")
print(f"URL: {url}")
# Middleware to record requests
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
recorder = RequestRecorder()
class RecordingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# Record request
body = await request.body()
response = await call_next(request)
# Record response
response_body = b""
async for chunk in response.body_iterator:
response_body += chunk
await recorder.record_request(
method=request.method,
url=str(request.url),
headers=request.headers,
body=body.decode() if body else None,
response_status=response.status_code,
response_body=response_body.decode()
)
return Response(
content=response_body,
status_code=response.status_code,
headers=dict(response.headers)
)
# Usage
app.add_middleware(RecordingMiddleware)
# Replay requests in test environment
# python replay.py --target http://localhost:8000# Kubernetes: Shadow traffic to debug environment
apiVersion: v1
kind: Service
metadata:
name: api-service
spec:
selector:
app: api
ports:
- port: 80
targetPort: 8080
---
# Envoy/Istio: Mirror traffic to debug instance
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
name: api-service
spec:
hosts:
- api-service
http:
- match:
- headers:
x-debug:
exact: "true"
route:
- destination:
host: api-service
subset: production
weight: 100
mirror:
host: api-service
subset: debug # Mirror to debug instance
mirrorPercentage:
value: 100 # Mirror 100% of matching requests
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
name: api-service
spec:
host: api-service
subsets:
- name: production
labels:
version: v1
- name: debug
labels:
version: debugApplication-level traffic shadowing:
from fastapi import Request, BackgroundTasks
import httpx
@app.post("/api/orders")
async def create_order(
request: Request,
order_data: dict,
background_tasks: BackgroundTasks
):
# Process request normally
result = await process_order(order_data)
# Shadow request to debug environment (async)
if should_shadow_request(request):
background_tasks.add_task(
shadow_request,
method=request.method,
url=str(request.url),
headers=dict(request.headers),
body=order_data
)
return result
async def shadow_request(method: str, url: str, headers: dict, body: dict):
"""
Send request to debug environment in background.
"""
# Replace production URL with debug URL
debug_url = url.replace("api.example.com", "debug.example.com")
async with httpx.AsyncClient() as client:
try:
response = await client.request(
method=method,
url=debug_url,
headers=headers,
json=body,
timeout=5.0 # Don't block production traffic
)
# Log differences
logger.info(
"Shadow request completed",
status_code=response.status_code,
url=debug_url
)
except Exception as e:
# Don't fail production traffic
logger.error("Shadow request failed", error=str(e))
def should_shadow_request(request: Request) -> bool:
"""
Determine if request should be shadowed.
"""
# Shadow 10% of traffic
import random
return random.random() < 0.10
# Or shadow specific users
# user_id = request.headers.get("X-User-ID")
# return user_id in ["debug_user_1", "debug_user_2"]import time
from datetime import datetime, timedelta
import structlog
logger = structlog.get_logger()
class ClockSkewDetector:
"""
Detect and handle clock skew in distributed systems.
"""
def __init__(self, max_skew_seconds: int = 300): # 5 minutes
self.max_skew_seconds = max_skew_seconds
def check_timestamp(
self,
remote_timestamp: str,
service_name: str
) -> bool:
"""
Check if remote timestamp is within acceptable range.
"""
try:
remote_time = datetime.fromisoformat(remote_timestamp)
local_time = datetime.utcnow()
skew = abs((remote_time - local_time).total_seconds())
if skew > self.max_skew_seconds:
logger.warning(
"Clock skew detected",
service=service_name,
remote_timestamp=remote_timestamp,
local_timestamp=local_time.isoformat(),
skew_seconds=skew
)
return False
return True
except Exception as e:
logger.error("Invalid timestamp format", error=str(e))
return False
def sync_with_ntp(self):
"""
Sync local clock with NTP server.
"""
import ntplib
from datetime import datetime
ntp_client = ntplib.NTPClient()
try:
response = ntp_client.request('pool.ntp.org', version=3)
local_time = time.time()
ntp_time = response.tx_time
offset = ntp_time - local_time
logger.info(
"NTP sync completed",
offset_ms=offset * 1000,
ntp_server="pool.ntp.org"
)
return offset
except Exception as e:
logger.error("NTP sync failed", error=str(e))
return None
# Middleware to check clock skew
from fastapi import Request, HTTPException
clock_detector = ClockSkewDetector()
@app.middleware("http")
async def check_clock_skew(request: Request, call_next):
# Check if request has timestamp header
remote_timestamp = request.headers.get("X-Timestamp")
if remote_timestamp:
if not clock_detector.check_timestamp(
remote_timestamp,
request.headers.get("X-Service-Name", "unknown")
):
raise HTTPException(
status_code=400,
detail="Clock skew detected. Check NTP sync."
)
response = await call_next(request)
return responseNTP sync commands:
# Check NTP status
timedatectl status
# Sync with NTP server (Linux)
sudo ntpdate pool.ntp.org
# Enable NTP sync (systemd)
sudo timedatectl set-ntp true
# Check clock offset
ntpq -pimport random
import asyncio
from fastapi import HTTPException
class ChaosMonkey:
"""
Inject controlled failures for debugging.
"""
def __init__(self, enabled: bool = False):
self.enabled = enabled
async def inject_latency(
self,
min_ms: int = 100,
max_ms: int = 1000,
probability: float = 0.1
):
"""
Inject random latency.
"""
if not self.enabled:
return
if random.random() < probability:
delay = random.randint(min_ms, max_ms) / 1000
logger.warning("CHAOS: Injecting latency", delay_seconds=delay)
await asyncio.sleep(delay)
def inject_error(
self,
probability: float = 0.05,
error_code: int = 503
):
"""
Inject random errors.
"""
if not self.enabled:
return
if random.random() < probability:
logger.warning("CHAOS: Injecting error", error_code=error_code)
raise HTTPException(
status_code=error_code,
detail="Chaos engineering: simulated failure"
)
def inject_timeout(
self,
probability: float = 0.05,
timeout_seconds: int = 30
):
"""
Inject timeout by sleeping indefinitely.
"""
if not self.enabled:
return
if random.random() < probability:
logger.warning("CHAOS: Injecting timeout", timeout_seconds=timeout_seconds)
import time
time.sleep(timeout_seconds)
# Enable chaos in specific environment
import os
chaos = ChaosMonkey(enabled=os.getenv("CHAOS_ENABLED") == "true")
@app.get("/api/users")
async def get_users():
# Inject chaos
await chaos.inject_latency(probability=0.1)
chaos.inject_error(probability=0.05)
users = await db.get_users()
return usersUsing Chaos Mesh (Kubernetes):
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
name: network-delay
spec:
action: delay
mode: one
selector:
namespaces:
- production
labelSelectors:
app: api-service
delay:
latency: "100ms"
jitter: "50ms"
duration: "30s"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
name: pod-failure
spec:
action: pod-failure
mode: one
selector:
namespaces:
- production
labelSelectors:
app: payment-service
duration: "60s"## Step 1: Identify Trace
- [ ] Extract trace ID from error logs
- [ ] Find all spans in trace (Grafana Tempo, Jaeger)
- [ ] Identify failed span and service
## Step 2: Analyze Dependencies
- [ ] Check which services were called
- [ ] Verify trace context propagation
- [ ] Look for missing spans (broken propagation)
## Step 3: Check Timing
- [ ] Compare timestamps across services
- [ ] Look for clock skew (>5min difference)
- [ ] Identify slow spans (>1s)
## Step 4: Correlate with Infrastructure
- [ ] Check CPU/memory on affected services
- [ ] Review network metrics (latency, packet loss)
- [ ] Check for recent deployments
## Step 5: Reproduce
- [ ] Replay request in staging
- [ ] Use traffic shadowing
- [ ] Inject chaos (latency, errors)# Grafana Loki + Tempo
# Find logs for trace
{app="api-service"} | json | trace_id="abc123"
# Find errors in trace
{trace_id="abc123"} | json | level="error"
# Find slow requests
{app="api-service"} | json | duration_ms > 1000 | trace_id!=""# WRONG: Breaks trace chain
requests.get("http://service-b/api/data")
# CORRECT: Propagate trace context
headers = {}
inject(headers)
requests.get("http://service-b/api/data", headers=headers)# WRONG: Clock skew causes issues
event_time = datetime.now() # Local clock!
# CORRECT: Use monotonic clock or logical timestamps
event_time = time.monotonic() # Monotonic clock
# Or: Use Lamport/Vector clocks for ordering# WRONG: Leaves system in inconsistent state
await reserve_inventory()
await charge_payment() # Fails, but inventory not released!
# CORRECT: Compensate on failure
try:
await reserve_inventory()
await charge_payment()
except:
await release_inventory() # Compensating transaction
raiseLast Updated: 2025-10-26 Format Version: 1.0 (Atomic)