Gateway Analytics — Monitoring Traffic Patterns and Metrics
In this tutorial, you'll learn about Gateway Analytics. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
An API gateway is the ideal vantage point for collecting analytics because every request passes through it before reaching any backend service.
What You'll Learn
By the end of this lesson, you will implement request analytics tracking, latency distribution collection, error rate monitoring, and traffic pattern visualization at the gateway level.
Why It Matters
Centralized analytics at the gateway gives you a complete picture of your API ecosystem without instrumenting every backend service individually.
Real-World Use
Durga Antivirus Pro collects gateway analytics to track which scanning endpoints are most used, detect traffic anomalies indicating abuse, and measure API response times across all services.
Gateway Analytics Pipeline
flowchart LR
Request-->Gateway
Gateway-->Counter[Metrics Counter]
Gateway-->Latency[Latency Tracker]
Gateway-->Logger[Analytics Logger]
Counter-->DB[(Time-Series DB)]
Latency-->DB
Logger-->DB
DB-->Dashboard[Dashboard]
Request Metrics Collector
The gateway records every request with method, path, status, and duration.
import time
from dataclasses import dataclass, field
from typing import Dict, List
from collections import defaultdict
import threading
@dataclass
class RequestMetric:
method: str
path: str
status_code: int
duration_ms: float
client_id: str
timestamp: float
class MetricsCollector:
def __init__(self):
self.metrics: List[RequestMetric] = []
self.lock = threading.Lock()
self.counts: Dict[str, int] = defaultdict(int)
self.latencies: Dict[str, List[float]] = defaultdict(list)
def record(self, metric: RequestMetric):
with self.lock:
self.metrics.append(metric)
key = f"{metric.method} {metric.path}"
self.counts[key] += 1
self.latencies[key].append(metric.duration_ms)
def get_request_count(self, method: str, path: str) -> int:
return self.counts.get(f"{method} {path}", 0)
def get_avg_latency(self, method: str, path: str) -> float:
key = f"{method} {path}"
latencies = self.latencies.get(key, [])
if not latencies:
return 0.0
return sum(latencies) / len(latencies)
def get_percentile(self, method: str, path: str,
percentile: float = 0.95) -> float:
key = f"{method} {path}"
latencies = sorted(self.latencies.get(key, []))
if not latencies:
return 0.0
index = int(len(latencies) * percentile)
return latencies[min(index, len(latencies) - 1)]
collector = MetricsCollector()
collector.record(RequestMetric(
"GET", "/api/scan", 200, 45.2, "client-1", time.time()
))
collector.record(RequestMetric(
"GET", "/api/scan", 200, 152.1, "client-2", time.time()
))
print(f"Avg latency: {collector.get_avg_latency('GET', '/api/scan'):.1f}ms")
print(f"P95 latency: {collector.get_percentile('GET', '/api/scan', 0.95):.1f}ms")
Traffic Pattern Analyzer
Analyze traffic patterns to detect anomalies and usage trends.
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
class TrafficAnalyzer:
def __init__(self, window_minutes: int = 5):
self.window_minutes = window_minutes
self.hourly_counts: Dict[str, int] = defaultdict(int)
self.endpoint_counts: Dict[str, int] = defaultdict(int)
self.client_counts: Dict[str, int] = defaultdict(int)
def analyze_request(self, method: str, path: str,
client_id: str, timestamp: float):
hour_key = datetime.fromtimestamp(
timestamp
).strftime("%Y-%m-%d %H:00")
endpoint_key = f"{method} {path}"
self.hourly_counts[hour_key] += 1
self.endpoint_counts[endpoint_key] += 1
self.client_counts[client_id] += 1
def get_top_endpoints(self, n: int = 10
) -> List[Tuple[str, int]]:
return sorted(
self.endpoint_counts.items(),
key=lambda x: x[1],
reverse=True
)[:n]
def get_top_clients(self, n: int = 10
) -> List[Tuple[str, int]]:
return sorted(
self.client_counts.items(),
key=lambda x: x[1],
reverse=True
)[:n]
def get_hourly_trend(self) -> List[Tuple[str, int]]:
return sorted(self.hourly_counts.items())
analyzer = TrafficAnalyzer()
analyzer.analyze_request("GET", "/api/scan", "client-1", time.time())
analyzer.analyze_request("POST", "/api/scan", "client-1", time.time())
analyzer.analyze_request("GET", "/api/report", "client-2", time.time())
print("Top endpoints:", analyzer.get_top_endpoints(5))
print("Top clients:", analyzer.get_top_clients(5))
Error Rate Tracking
Track error rates by endpoint and client to quickly identify problem areas.
from collections import defaultdict
from typing import Dict
class ErrorRateTracker:
def __init__(self):
self.total: Dict[str, int] = defaultdict(int)
self.errors: Dict[str, int] = defaultdict(int)
def record(self, endpoint: str, status_code: int):
self.total[endpoint] += 1
if status_code >= 400:
self.errors[endpoint] += 1
def error_rate(self, endpoint: str) -> float:
t = self.total.get(endpoint, 0)
if t == 0:
return 0.0
return self.errors.get(endpoint, 0) / t * 100
def high_error_endpoints(self, threshold: float = 5.0
) -> List[Tuple[str, float]]:
result = []
for endpoint in self.total:
rate = self.error_rate(endpoint)
if rate > threshold:
result.append((endpoint, rate))
return sorted(result, key=lambda x: x[1], reverse=True)
tracker = ErrorRateTracker()
tracker.record("/api/scan", 200)
tracker.record("/api/scan", 500)
tracker.record("/api/scan", 503)
tracker.record("/api/report", 200)
tracker.record("/api/report", 200)
print(f"Scan error rate: {tracker.error_rate('/api/scan'):.1f}%")
print("High error endpoints:", tracker.high_error_endpoints(5.0))
Common Mistakes
Mistake 1: Sampling Bias
Sampling only successful requests gives a skewed picture. Include errors, timeouts, and rejected requests.
Mistake 2: Not Tagging by Service Version
Without version tags, you cannot correlate analytics with deployments that may have introduced regressions.
Mistake 3: Ignoring Client-Side Metrics
Gateway analytics miss client-side latency from network issues. Combine with real user monitoring for the full picture.
Mistake 4: Storing Raw Logs Without Aggregation
Raw request logs at scale are expensive to query. Always pre-aggregate into time-series metrics.
Mistake 5: No Alerting on Anomalies
Collecting analytics without alerting on anomalies is just data hoarding. Set thresholds and notify on deviations.
Practice Questions
- What metrics should every API gateway analytics system collect as a minimum?
- Why is the P95 latency metric more useful than average latency?
- How can gateway analytics help detect a DDoS attack?
- What is the difference between request-level and session-level analytics?
- How do you handle analytics for streaming endpoints like WebSocket?
Challenge
Build a gateway analytics module that tracks request counts per endpoint per minute, calculates error rates, and emits an alert when any endpoint exceeds 10% error rate in a 5-minute window.
FAQ
Mini Project
Build a gateway analytics service that records every request with method, path, status, and duration, exposes a /metrics endpoint for Prometheus scraping, and provides top-N endpoint and client reports.
What's Next
Learn about Gateway Monitoring with Prometheus and Grafana, or explore Gateway Alerting for automated Incident Response.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro