Gateway Monitoring and Observability — Complete Guide
In this tutorial, you'll learn about Gateway Monitoring. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Monitoring the API gateway provides visibility into all API traffic, enabling performance optimization, error detection, capacity planning, and security auditing.
What You'll Learn
By the end of this lesson, you will implement metrics collection, structured request logging, distributed tracing, health check endpoints, and alerting for gateway monitoring.
Why It Matters
The gateway processes every API request, making it the ideal place to collect metrics, logs, and traces. Without monitoring, you are blind to performance issues and errors.
Real-World Use
A gateway exposes a /health endpoint, logs every request with duration and status code, emits Prometheus metrics for request rate, latency, and error rate, and sends distributed trace spans to Jaeger.
Monitoring Architecture
flowchart LR
Gateway -->|Metrics| Prometheus[Prometheus]
Gateway -->|Logs| ELK[Elasticsearch/Logstash]
Gateway -->|Traces| Jaeger[Jaeger]
Gateway -->|Health| HealthCheck[Health Endpoint]
Prometheus --> Grafana[Grafana Dashboard]
ELK --> Kibana[Kibana]
Metrics Collection
# metrics_collection.py
import time
from collections import defaultdict
from typing import Dict, List, Optional
class GatewayMetrics:
def __init__(self):
self.request_count = 0
self.status_counts: Dict[int, int] = defaultdict(int)
self.latencies: List[float] = []
self.active_connections = 0
self.error_count = 0
def record_request(self, status: int, latency_ms: float):
self.request_count += 1
self.status_counts[status] += 1
self.latencies.append(latency_ms)
if status >= 500:
self.error_count += 1
def record_connection(self, active: int):
self.active_connections = active
def average_latency(self) -> float:
if not self.latencies:
return 0.0
return sum(self.latencies[-100:]) / min(100, len(self.latencies))
def error_rate(self) -> float:
if self.request_count == 0:
return 0.0
return (self.error_count / self.request_count) * 100
def snapshot(self) -> Dict:
return {
"total_requests": self.request_count,
"status_codes": dict(self.status_counts),
"avg_latency_ms": round(self.average_latency(), 2),
"error_rate_pct": round(self.error_rate(), 2),
"active_connections": self.active_connections,
}
def export_prometheus(self) -> str:
lines = [
f"# HELP gateway_requests_total Total API requests",
f"# TYPE gateway_requests_total counter",
f"gateway_requests_total {self.request_count}",
"",
f"# HELP gateway_request_latency_ms Request latency in ms",
f"# TYPE gateway_request_latency_ms gauge",
f"gateway_request_latency_ms {self.average_latency()}",
"",
f"# HELP gateway_errors_total Total API errors",
f"# TYPE gateway_errors_total counter",
f"gateway_errors_total {self.error_count}",
"",
f"# HELP gateway_active_connections Active connections",
f"# TYPE gateway_active_connections gauge",
f"gateway_active_connections {self.active_connections}",
]
return "\n".join(lines)
metrics = GatewayMetrics()
metrics.record_request(200, 12.5)
metrics.record_request(200, 8.3)
metrics.record_request(500, 150.0)
metrics.record_request(404, 5.1)
print("Metrics snapshot:")
print(metrics.snapshot())
print("\nPrometheus format:")
print(metrics.export_prometheus())
Expected output:
Metrics snapshot:
{'total_requests': 4, 'status_codes': {200: 2, 500: 1, 404: 1}, 'avg_latency_ms': 43.97, 'error_rate_pct': 25.0, 'active_connections': 0}
Prometheus format:
# HELP gateway_requests_total Total API requests
# TYPE gateway_requests_total counter
gateway_requests_total 4
...
Structured Request Logging
# request_logging.py
import json
import time
from typing import Any, Dict, Optional
class RequestLogger:
def __init__(self):
self.logs: list = []
def log_request(self, method: str, path: str, status: int,
latency_ms: float, client_ip: str,
request_id: str, user_id: Optional[str] = None,
error: Optional[str] = None):
entry = {
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"request_id": request_id,
"method": method,
"path": path,
"status": status,
"latency_ms": round(latency_ms, 2),
"client_ip": client_ip,
"user_id": user_id,
"error": error,
}
self.logs.append(entry)
return entry
def export_json(self) -> str:
return json.dumps(self.logs, indent=2)
def query_error_logs(self, min_status: int = 500) -> list:
return [log for log in self.logs if log["status"] >= min_status]
def query_slow_logs(self, threshold_ms: float = 1000) -> list:
return [log for log in self.logs if log["latency_ms"] > threshold_ms]
logger = RequestLogger()
logger.log_request("GET", "/api/products", 200, 45.2, "10.0.0.1", "req_001", "user_42")
logger.log_request("POST", "/api/orders", 500, 2500.0, "10.0.0.2", "req_002", "user_43", "Database timeout")
logger.log_request("GET", "/api/users", 200, 12.1, "10.0.0.3", "req_003")
print(f"Total log entries: {len(logger.logs)}")
print(f"Error logs: {len(logger.query_error_logs())}")
print(f"Slow logs (>1000ms): {len(logger.query_slow_logs(1000))}")
print("\nSample log entry:")
print(json.dumps(logger.logs[1], indent=2))
Expected output:
Total log entries: 3
Error logs: 1
Slow logs (>1000ms): 1
Sample log entry:
{
"timestamp": "2026-06-28T00:00:00Z",
"request_id": "req_002",
"method": "POST",
"path": "/api/orders",
"status": 500,
"latency_ms": 2500.0,
"client_ip": "10.0.0.2",
"user_id": "user_43",
"error": "Database timeout"
}
Health Check Endpoint
# health_check.py
import time
from typing import Any, Dict, List, Optional
class HealthCheck:
def __init__(self):
self.checks: Dict[str, dict] = {}
def register(self, name: str, check_fn):
self.checks[name] = {
"fn": check_fn,
"last_check": 0,
"last_result": None,
}
def run_all(self) -> Dict:
results = {"status": "healthy", "checks": [], "timestamp": time.time()}
for name, check in self.checks.items():
try:
ok, message = check["fn"]()
status = "pass" if ok else "fail"
if status == "fail":
results["status"] = "degraded"
results["checks"].append({
"name": name,
"status": status,
"message": message,
})
except Exception as e:
results["status"] = "degraded"
results["checks"].append({
"name": name,
"status": "fail",
"message": str(e),
})
return results
hc = HealthCheck()
def check_redis():
return True, "Redis connected"
def check_db():
return True, "Database pool OK"
def check_backend():
return False, "Payment service timeout"
hc.register("redis", check_redis)
hc.register("database", check_db)
hc.register("payment-backend", check_backend)
result = hc.run_all()
print(f"Overall status: {result['status']}")
for check in result["checks"]:
print(f" {check['name']}: {check['status']} - {check['message']}")
Expected output:
Overall status: degraded
redis: pass - Redis connected
database: pass - Database pool OK
payment-backend: fail - Payment service timeout
Common Mistakes
1. Not Logging Request IDs
Without unique request IDs, correlating logs across services is impossible. Generate a request ID at the gateway and propagate it.
2. Sampling Too Aggressively
Sampling 1% of traffic misses rare errors. Sample 100% of errors and 10% of successful requests.
3. No Custom Metrics for Business Events
Request metrics are not enough. Track business metrics like order placement rate, user signup rate, and product view count.
4. Health Check Without Dependencies
A health check that always returns healthy is useless. Check connectivity to Redis, database, and critical backends.
5. Ignoring P99 Latency
Average latency hides outliers. Track P50, P95, and P99 latency to understand the true user experience.
Practice Questions
1. What metrics should every gateway expose?
Request count by status code, latency percentiles, error rate, active connections, and upstream latency per service.
2. Why is structured logging important?
Structured logs (JSON) are machine-parseable and can be indexed, searched, and analyzed by log management systems.
3. What is distributed tracing and why use it?
Tracing tracks a request across multiple services, showing where time is spent. Essential for debugging performance in Microservices.
4. How does a health check endpoint work?
The gateway exposes a /health endpoint that runs diagnostic checks on dependencies and returns the overall system status (healthy/degraded/unhealthy).
Challenge
Build a monitoring system for a gateway that exposes Prometheus metrics, produces structured JSON logs, implements distributed tracing with request IDs, and provides a health check endpoint checking three critical dependencies.
FAQ
Mini Project: Monitoring Dashboard
# monitoring.py
import time
import random
from typing import Any, Dict, List, Optional
class MonitoringDashboard:
def __init__(self):
self.metrics: Dict[str, list] = {
"requests_per_sec": [], "error_rate": [], "latency_p99": [],
}
def collect_sample(self):
self.metrics["requests_per_sec"].append(random.randint(800, 1200))
self.metrics["error_rate"].append(round(random.uniform(0.1, 2.0), 2))
self.metrics["latency_p99"].append(round(random.uniform(100, 500), 1))
def current_status(self) -> str:
if not self.metrics["error_rate"]:
return "healthy"
latest_error = self.metrics["error_rate"][-1]
if latest_error > 5.0:
return "critical"
elif latest_error > 1.0:
return "degraded"
return "healthy"
db = MonitoringDashboard()
for _ in range(10):
db.collect_sample()
print(f"Status: {db.current_status()}")
print(f"Avg RPS: {sum(db.metrics['requests_per_sec'])/len(db.metrics['requests_per_sec']):.0f}")
print(f"Latest error rate: {db.metrics['error_rate'][-1]}%")
print(f"Latest P99 latency: {db.metrics['latency_p99'][-1]}ms")
Expected output:
Status: degraded
Avg RPS: 1000
Latest error rate: 1.2%
Latest P99 latency: 350ms
What's Next
You understand gateway monitoring. Next, build the gateway project to apply everything you have learned.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro