Gateway Logging — Structured Logging for API Gateways
In this tutorial, you'll learn about Gateway Logging. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
An API Gateway processes every request, making it the ideal layer for structured logging with correlation IDs that trace requests across all downstream services.
What You'll Learn
By the end of this lesson, you will implement structured gateway logging, generate and propagate correlation IDs, capture request/response metadata, and integrate with centralized logging systems.
Why It Matters
Gateway logs provide a complete picture of every request entering your system, enabling debugging, auditing, and performance analysis without instrumenting each service.
Real-World Use
Durga Antivirus Pro logs every API request at its gateway with a unique correlation ID, recording client IP, endpoint, latency, and status code for all scan and analysis requests.
Structured Logging at Gateway
flowchart LR
Client-->Gateway
Gateway-->Log[Structured Logger]
Log-->Console[stdout]
Log-->File[Log File]
Log-->ES[Elasticsearch]
Console-->Kibana
ES-->Kibana
Correlation ID Middleware
Every request gets a unique correlation ID that is passed to downstream services.
import uuid
import logging
import json
from datetime import datetime
from typing import Dict, Optional
class CorrelationLogger:
def __init__(self, service_name: str = "gateway"):
self.service_name = service_name
self.logger = logging.getLogger(service_name)
self._setup_handler()
def _setup_handler(self):
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.logger.setLevel(logging.INFO)
def generate_correlation_id(self) -> str:
return str(uuid.uuid4())
def log_request(self, correlation_id: str,
method: str, path: str,
client_ip: str, headers: Dict):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"service": self.service_name,
"correlation_id": correlation_id,
"type": "request",
"method": method,
"path": path,
"client_ip": client_ip,
"user_agent": headers.get("user-agent", ""),
"content_type": headers.get("content-type", ""),
}
self.logger.info(json.dumps(entry))
def log_response(self, correlation_id: str,
status_code: int,
duration_ms: float,
response_size: int):
entry = {
"timestamp": datetime.utcnow().isoformat(),
"service": self.service_name,
"correlation_id": correlation_id,
"type": "response",
"status_code": status_code,
"duration_ms": round(duration_ms, 2),
"response_size": response_size,
}
self.logger.info(json.dumps(entry))
logger = CorrelationLogger()
cid = logger.generate_correlation_id()
logger.log_request(cid, "POST", "/api/scan",
"203.0.113.42",
{"user-agent": "DodaBrowser/2.0"})
logger.log_response(cid, 200, 45.23, 1024)
Gateway Log Middleware
A middleware wrapper that logs every request and response automatically.
import time
from functools import wraps
from typing import Callable, Any
class GatewayLogMiddleware:
def __init__(self, logger: CorrelationLogger):
self.logger = logger
def wrap_handler(self, handler: Callable) -> Callable:
@wraps(handler)
def middleware(request: Dict) -> Dict:
correlation_id = request.get("headers", {}).get(
"x-correlation-id",
self.logger.generate_correlation_id()
)
start = time.time()
self.logger.log_request(
correlation_id,
request.get("method", "GET"),
request.get("path", "/"),
request.get("client_ip", "unknown"),
request.get("headers", {})
)
try:
response = handler(request)
duration = (time.time() - start) * 1000
self.logger.log_response(
correlation_id,
response.get("status_code", 200),
duration,
len(str(response.get("body", "")))
)
response["headers"] = response.get("headers", {})
response["headers"]["x-correlation-id"] = \
correlation_id
return response
except Exception as e:
duration = (time.time() - start) * 1000
self.logger.lotter.error(json.dumps({
"timestamp": datetime.utcnow().isoformat(),
"correlation_id": correlation_id,
"type": "error",
"error": str(e),
"duration_ms": round(duration, 2),
}))
raise
return middleware
def sample_handler(request):
return {"status_code": 200, "body": {"result": "ok"}}
middleware = GatewayLogMiddleware(logger)
wrapped = middleware.wrap_handler(sample_handler)
result = wrapped({
"method": "GET",
"path": "/api/health",
"client_ip": "10.0.0.1",
"headers": {"content-type": "application/json"}
})
print(f"Response: {result}")
Log Aggregation and Shipping
Ship gateway logs to a centralized system for analysis.
import json
from queue import Queue
from threading import Thread
from typing import List, Dict
import time
class LogShipper:
def __init__(self, batch_size: int = 100,
flush_interval: float = 5.0):
self.queue: Queue = Queue()
self.batch_size = batch_size
self.flush_interval = flush_interval
self._running = True
self._worker = Thread(target=self._flush_loop,
daemon=True)
self._worker.start()
def ship(self, log_entry: Dict):
self.queue.put(log_entry)
def _flush_loop(self):
batch: List[Dict] = []
last_flush = time.time()
while self._running:
try:
entry = self.queue.get(timeout=1.0)
batch.append(entry)
if (len(batch) >= self.batch_size
or time.time() - last_flush
>= self.flush_interval):
self._send_batch(batch)
batch = []
last_flush = time.time()
except Exception:
if batch:
self._send_batch(batch)
batch = []
last_flush = time.time()
def _send_batch(self, batch: List[Dict]):
payload = json.dumps(batch)
print(f"Shipping {len(batch)} logs: {len(payload)} bytes")
# In production: HTTP POST to Elasticsearch, Loki, etc
def stop(self):
self._running = False
shipper = LogShipper()
shipper.ship({"correlation_id": "abc", "type": "request"})
shipper.ship({"correlation_id": "abc", "type": "response"})
time.sleep(1)
shipper.stop()
Common Mistakes
Mistake 1: Logging Sensitive Data
Never log Authorization headers, API keys, or request bodies containing PII.
Mistake 2: Blocking the Request Path
Synchronous logging blocks the request. Always use async logging or a background queue.
Mistake 3: Inconsistent Correlation IDs
If downstream services regenerate correlation IDs, you lose the trace across services. Propagate the original ID.
Mistake 4: Logging at Wrong Level
Using debug for critical errors or error for normal 404s makes log analysis noisy and expensive.
Mistake 5: No Log Rotation or Retention
Gateway logs grow fast. Without rotation and retention policies, you will run out of disk space.
Practice Questions
- What is the purpose of a correlation ID in gateway logging?
- Why should gateway log entries be structured (JSON) rather than plain text?
- What fields should every gateway log entry contain?
- How do you prevent gateway logging from impacting request latency?
- What is the recommended log retention period for gateway logs?
Challenge
Build a gateway logging system that writes structured JSON logs to stdout, generates correlation IDs for requests that do not have one, and ships logs in batches of 50 to an HTTP endpoint.
FAQ
Mini Project
Build a gateway logging middleware that captures method, path, status code, client IP, duration, and correlation ID for every request, writes structured JSON to stdout, and includes a configurable sensitive-header redaction list.
What's Next
Learn about Gateway Analytics for traffic pattern analysis, or explore Gateway Monitoring with Prometheus metrics.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro