What Happened Teams deploying large language models (LLMs) routinely discover that common ad hoc load tests — curl loops, asyncio scripts, or single-process generators — give misleading latency and throughput…

AIPerf 소개: 측정 도구가 측정 대상을 가려 버리는 문제 모델을 서버에 올리고 프롬프트를 던지면 응답이 돌아옵니다. 그다음에 반드시 나오는 질문은 "이거 빠른 건가요?" 입니다. 이 질문에 답하려고 개발자들이 가장 먼저 하는 일은 대개 curl을 몇 번 날려 보거나, asyncio 기반의 짧은 부하 스크립트를 직접 짜거나, 일회용 부하 생성기를 하나 더 만드는 것입니다. NVIDIA 기술 블로그가 지적하는 문제는 이 세 갈래가 결국 같은 곳에서 무너진다는 점입니다. 단일 프로세스의 성능 한계에 걸리거나, 파이썬의 전역 인터프리터 잠금(Global Interpreter Lock, GIL) 이 동시성을 제한하거나, 애초에 자기가 만든 기준선에 대고 측정한 숫자가 나옵니다. 어느 쪽이든 완전히 신뢰하기 어려운 결과가, 요구사항이 바뀌는 순간 다시 짜야 하는 도구에...

Asyncio NATS client for Python

Stop grepping through unorganized log streams. Here is how to implement structured distributed tracing, context propagation, and custom span metrics in FastAPI and Python backend services. ### The Limits of logging.info() When backend services run locally, debugging is simple: throw in a few print() statements or use standard Python logging to follow execution flow. However, once your backend scales into asynchronous tasks (asyncio), concurrent background workers (Celery/ARQ), and distributed microservices, traditional stdout logs hit a wall: * **Interleaved Log Streams:** Concurrent requests interleave log statements across threads, making it impossible to reconstruct a single user’s request path. * **Silent Bottlenecks:** A query takes 2.4 seconds, but standard logs can’t pinpoint whether the delay occurred in DB connection pooling, HTTP serialization, or external API calls. * **Context Loss:** When an HTTP request triggers an async worker, correlation IDs are lost across thread boundaries. To solve this, modern production systems use **OpenTelemetry (OTel)** the vendor-agnostic CNCF standard for collecting traces, metrics, and logs. This hands-on guide walks through implementing production-grade OpenTelemetry tracing in Python and FastAPI, handling asynchronous context propagation, and defining custom spans for silent performance bottlenecks. ### The Core Architecture of OpenTelemetry Before writing code, it is vital to understand how telemetry signals flow from your application to an observability backend (like Jaeger, Grafana Tempo, Datadog, or Honeycomb): * **TracerProvider:** The central factory object that holds resource attributes (e.g., service name, environment) and global configuration. * **Tracer:** The object used within your code to start and end execution units. * **Span:** A single timed block of work (e.g., a database query, an outbound HTTP fetch, or a execution function). A collection of nested spans forms a **Trace**. * **BatchSpanProcessor:** An in-memory queue that batches spans asynchronously before sending them to prevent blocking application execution. ### Setting Up Automatic Instrumentation in FastAPI Let’s start by installing the required OpenTelemetry packages: pip install opentelemetry-api \ opentelemetry-sdk \ opentelemetry-exporter-otlp \ opentelemetry-instrumentation-fastapi \ opentelemetry-instrumentation-httpx ### Initializing the OpenTelemetry SDK Here is how to construct a robust initialization module (telemetry.py) that handles tracer configuration and configures automatic span batching: # telemetry.py import os from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource def setup_telemetry(service_name: str = "order-processing-service") -> trace.Tracer: # 1. Define Resource Metadata (Metadata attached to every trace) resource = Resource.create( attributes={ "service.name": service_name, "deployment.environment": os.getenv("ENV", "production"), } ) # 2. Instantiate global TracerProvider provider = TracerProvider(resource=resource) # 3. Configure OTLP gRPC Exporter (pointing to collector or Jaeger) otlp_exporter = OTLPSpanExporter( endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"), insecure=True, ) # 4. Wrap with BatchSpanProcessor to avoid blocking the main event loop processor = BatchSpanProcessor(otlp_exporter) provider.add_span_processor(processor) # 5. Register global tracer provider trace.set_tracer_provider(provider) return trace.get_tracer(service_name) ### Instrumenting FastAPI Endpoints & Asynchronous Operations Once the provider is registered, instrument your FastAPI application and add custom manual instrumentation for deep internal functions using context managers. # main.py import asyncio import httpx from fastapi import FastAPI, HTTPException from opentelemetry import trace from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from telemetry import setup_telemetry # Initialize global telemetry setup tracer = setup_telemetry("payment-api") app = FastAPI(title="Order API") # Automatically instrument incoming FastAPI HTTP routes FastAPIInstrumentor.instrument_app(app) # Automatically propagate context over outgoing HTTPX client calls HTTPXClientInstrumentor().instrument() async def query_fraud_detection_service(user_id: str) -> bool: """Simulates an internal asynchronous database or microservice call.""" # Create a explicit custom child span with tracer.start_as_current_span("fraud_check_db_query") as span: # Attach high-value metadata attributes to the span span.set_attribute("user.id", user_id) span.set_attribute("db.system", "postgresql") await asyncio.sleep(0.15) # Simulate DB latency # Record events for specific milestones within a span span.add_event("fraud_score_evaluated", {"risk_score": 0.02}) return True @app.post("/checkout/{order_id}") async def process_checkout(order_id: str, user_id: str): # Obtain current active span created automatically by FastAPIInstrumentor current_span = trace.get_current_span() current_span.set_attribute("order.id", order_id) # Execute custom child function is_safe = await query_fraud_detection_service(user_id) if not is_safe: current_span.set_status(trace.Status(trace.StatusCode.ERROR, "Fraud detected")) raise HTTPException(status_code=400, detail="Transaction flagged") # Outbound HTTP calls will automatically propagate w3c traceparent headers async with httpx.AsyncClient() as client: with tracer.start_as_current_span("external_payment_gateway_call"): # The HTTPX instrumentor automatically attaches trace headers here response = await client.get("https://httpbin.org/delay/1") return {"status": "success", "order_id": order_id} ### Context Propagation Across Async Boundaries One of the most common pitfalls in Python backend observability occurs when passing context to background workers (such as ARQ, Celery, or bare asyncio.create_task). Without explicit context propagation, the trace context breaks, and the background execution appears in your observability UI as an unattached, rootless trace. ### Injecting & Extracting Context Manually When enqueuing a background job, inject the W3C traceparent headers into the task payload: from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator # 1. INJECT CONTEXT (Before enqueuing background task) def enqueue_background_job(payload: dict): carrier = {} # Extract current active context into carrier dict TraceContextTextMapPropagator().inject(carrier) # Store carrier trace headers alongside worker payload payload["_trace_context"] = carrier background_worker_queue.send(payload) # 2. EXTRACT CONTEXT (Inside Worker Process) def process_background_job(payload: dict): carrier = payload.get("_trace_context", {}) # Extract parent context from dictionary extracted_context = TraceContextTextMapPropagator().extract(carrier) # Start worker span attached directly to the original parent trace context with tracer.start_as_current_span("worker_process_task", context=extracted_context): print(f"Processing background task for order: {payload.get('order_id')}") ### Best Practices Checklist Shifting from passive logging to active OpenTelemetry tracing changes how production bottlenecks are identified and solved. ### Observability Best Practices for Python Developers: 1. **Never Block the Event Loop:** Always wrap your OTLP exporters in a BatchSpanProcessor to avoid adding network overhead to application threads. 2. **Instrument System Boundaries:** Ensure outbound HTTP clients (httpx, requests) and database drivers (SQLAlchemy, psycopg3) are instrumented so trace boundaries cross network hops cleanly. 3. **Control Attribute Cardinality:** Do not attach raw passwords, personally identifiable information (PII), or high-cardinality unique IDs (e.g., thousands of raw raw UUID strings) as span names. Store high-cardinality variables inside span attributes. 4. **Leverage Status Codes & Exceptions:** Call span.record_exception(e) inside try...except blocks to surface full exception stack traces directly inside flamegraph UI visualizations. ### Need High-Impact Technical Content for Your Team? I help engineering-focused companies, developer-tooling startups, and SaaS platforms explain complex infrastructure, backend architecture, and developer tooling through publication-grade articles. Whether you need deep-dive technical essays, developer guides, or architecture counter-narratives, feel free to reach out: * 📩 **Email:** abhishekninja2018@gmail.com * 💼 **LinkedIn:** linkedin.com/in/abhishekninja * 🛠️ **Capabilities:** Long-form Technical Essays | Hands-On Developer Tutorials | System Architecture Breakdowns | Benchmarks & Product Comparisons

Welcome to issue #228 of What’s up in the Python and tech environment?

Asyncio-based client for SpamAssassin's SPAMD service