Backend Logging Patterns and Structured Logging — Complete Implementation Guide
In this tutorial, you'll learn about Backend Logging Patterns and Structured Logging. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Backend logging is the practice of recording application events in a structured, searchable format that enables debugging, monitoring, and Observability across distributed services.
What You'll Learn
By the end of this tutorial, you will implement structured JSON logging in Python and Node.js, configure log levels and context enrichment, integrate with ELK Stack and Loki for centralized log aggregation, and design logging strategies for debugging and auditing.
Why It Matters
Without structured logging, debugging production issues requires SSH-ing into servers and grepping flat text files. With structured logging, every log entry contains searchable key-value pairs that can be aggregated across hundreds of services. Doda Browser uses structured logging to correlate user requests across its browser, sync, and scan services.
Real-World Use
A user reports that their file upload failed. With structured logging, the operations team searches for the user's request_id across all services. They find the API Gateway log, the upload handler log, the virus scanner log, and the storage service log -- all correlated by the same trace_id. The error is traced to a timeout in the virus scanner, and the team resolves it in minutes.
Logging Architecture
Graph TD subgraph "Application" S1[Service 1] -->|JSON logs| F1[File / stdout] S2[Service 2] -->|JSON logs| F2[File / stdout] S3[Service 3] -->|JSON logs| F3[File / stdout] end subgraph "Log Aggregation" F1 --> A[Log Shipper - Filebeat / Promtail] F2 --> A F3 --> A A --> B[Central Storage - Elasticsearch / Loki] end subgraph "Visualization" B --> C[Kibana / Grafana] B --> D[Alert Manager] end style B fill:#f90,color:#fff
Applications emit structured JSON logs to stdout. Log shippers (Filebeat, Promtail) forward logs to centralized storage. Engineers query logs using Kibana or Grafana and set alerts on error patterns.
Python: Structured Logging
# structured_logger.py
# Structured JSON logging with context enrichment
import JSON
import logging
import uuid
import socket
from datetime import datetime, timezone
from functools import wraps
from typing import Optional, Dict, Any
class JSONFormatter(logging.Formatter):
"""Custom formatter that outputs structured JSON logs."""
def __init__(self, service_name: str = "myapp", **kwargs):
super().__init__()
self.service_name = service_name
self.hostname = socket.gethostname()
def format(self, record: logging.LogRecord) -> str:
log_entry = {
'timestamp': datetime.fromtimestamp(
record.created, tz=timezone.utc
).isoformat(),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno,
'service': self.service_name,
'hostname': self.hostname,
}
# Add exception info if present
if record.exc_info and record.exc_info[0]:
log_entry['exception'] = {
'type': record.exc_info[0].__name__,
'message': str(record.exc_info[1]),
'traceback': self.formatException(record.exc_info),
}
# Add extra context fields
for key, value in getattr(record, 'context', {}).items():
log_entry[key] = value
return JSON.dumps(log_entry, default=str)
class StructuredLogger:
"""Logger wrapper that adds context enrichment."""
def __init__(self, name: str, service_name: str = "myapp", level=logging.INFO):
self.logger = logging.getLogger(name)
self.logger.setLevel(level)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter(service_name=service_name))
self.logger.handlers = []
self.logger.addHandler(handler)
def _with_context(self, **kwargs) -> logging.Logger:
"""Create a log record with extra context."""
old_Factory = logging.getLogRecordFactory()
def record_Factory(*args, **kwargs_Factory):
record = old_Factory(*args, **kwargs_Factory)
record.context = kwargs
return record
logging.setLogRecordFactory(record_Factory)
return self.logger
def info(self, message: str, **context):
"""Log with INFO level and context."""
self._with_context(**context).info(message)
def error(self, message: str, **context):
"""Log with ERROR level and context."""
self._with_context(**context).error(message)
def warning(self, message: str, **context):
"""Log with WARNING level and context."""
self._with_context(**context).warning(message)
def debug(self, message: str, **context):
"""Log with DEBUG level and context."""
self._with_context(**context).debug(message)
# Create logger
log = StructuredLogger('myapp.API', service_name='dodatech-API')
# Usage
log.info('User login successful',
user_id='usr_12345',
ip_address='192.168.1.1',
auth_method='oauth2',
session_duration_ms=245,
)
try:
raise ValueError('Invalid payment amount')
except Exception as e:
log.error('Payment processing failed',
order_id='ORD-789',
amount=9999,
currency='USD',
error=str(e),
)
Expected log output:
{"timestamp": "2026-06-22T10:00:00+00:00", "level": "INFO", "logger": "myapp.API", "message": "User login successful", "module": "myapp", "function": "<module>", "line": 42, "service": "dodatech-API", "hostname": "web-01", "user_id": "usr_12345", "ip_address": "192.168.1.1", "auth_method": "oauth2"}
Node.js: Structured Logging with Pino
// logger.js
// Structured JSON logging with Pino
const pino = require('pino');
const { v4: uuidv4 } = require('uuid');
// Create logger with custom serializers
const logger = pino({
level: Process.env.LOG_LEVEL || 'info',
timestamp: pino.stdTimeFunctions.isoTime,
formatters: {
level(label) {
return { level: label };
},
},
serializers: {
req: pino.stdSerializers.req,
res: pino.stdSerializers.res,
err: pino.stdSerializers.err,
},
redact: {
paths: ['req.headers.authorization', 'req.headers.cookie', 'body.password', 'body.token'],
censor: '[REDACTED]',
},
});
// Correlation middleware for Express
function correlationMiddleware(req, res, next) {
const correlationId = req.headers['x-correlation-id'] || uuidv4();
req.correlationId = correlationId;
res.set('X-Correlation-ID', correlationId);
const childLogger = logger.child({
correlationId,
method: req.method,
URL: req.URL,
ip: req.ip,
userAgent: req.get('User-Agent'),
});
req.log = childLogger;
const start = Date.now();
res.on('finish', () => {
const duration = Date.now() - start;
childLogger.info({
msg: 'request completed',
statusCode: res.statusCode,
durationMs: duration,
contentLength: res.get('Content-Length'),
});
});
next();
}
// Usage
const Express = require('Express');
const app = Express();
app.use(correlationMiddleware);
app.get('/users/:id', async (req, res) => {
req.log.info({ userId: req.params.id }, 'Fetching user');
try {
const user = { id: req.params.id, name: 'Alice' };
req.log.info({ userId: req.params.id, found: true }, 'User found');
res.JSON(user);
} catch (error) {
req.log.error({ err: error, userId: req.params.id }, 'Failed to fetch user');
res.status(500).JSON({ error: 'Internal error' });
}
});
// Database query logging wrapper
function queryLogger(dbClient) {
return {
async query(text, params) {
const start = Date.now();
try {
const result = await dbClient.query(text, params);
const duration = Date.now() - start;
logger.info({
msg: 'database query',
query: text.substring(0, 100),
durationMs: duration,
rowCount: result.rowCount,
});
return result;
} catch (error) {
const duration = Date.now() - start;
logger.error({
msg: 'database query failed',
query: text.substring(0, 100),
durationMs: duration,
err: error,
});
throw error;
}
},
};
}
module.exports = { logger, correlationMiddleware, queryLogger };
Log Levels and Configuration
# log-config.YAML
# Centralized logging configuration
logging:
default_level: INFO
format: JSON
# Per-module log levels for granular control
modules:
myapp.API: INFO
myapp.db: WARN
myapp.cache: DEBUG
myapp.HTTP_client: ERROR
# Sensitive fields to redact
redact:
- password
- secret
- token
- authorization
- cookie
- ssn
- credit_card
- API_key
# Context fields always included
always_include:
- service_name
- environment
- hostname
- correlation_id
# Sampling configuration for high-volume logs
sampling:
DEBUG: 0.1 # Log 10% of debug messages
INFO: 1.0 # Log all info messages
WARN: 1.0
ERROR: 1.0
# Output destinations
outputs:
- type: stdout
format: JSON
- type: file
path: /var/log/myapp/app.log
max_size_mb: 100
max_files: 10
format: JSON
# Filebeat configuration for shipping to Elasticsearch
filebeat:
inputs:
- type: log
paths:
- /var/log/myapp/*.log
JSON.keys_under_root: true
JSON.add_error_key: true
JSON.message_key: message
output.elasticsearch:
hosts: ["HTTP://elasticsearch:9200"]
index: "myapp-logs-%{+yyyy.MM.dd}"
Common Errors
1. Logging Sensitive Data
Logging passwords, credit card numbers, API keys, or personal identifiable information (PII) creates a Compliance violation and security risk. Use a redaction list to automatically censor sensitive fields. Audit logs regularly for accidental exposure.
2. Using Unstructured Text Logs
"User 12345 logged in at 10:00" is hard to search and parse. Use structured JSON logs with field names: {"user_id": "12345", "event": "login", "timestamp": "2026-06-22T10:00:00Z"}. Structured logs enable automated analysis, filtering, and alerting.
3. Not Including Correlation IDs
In a Microservices Architecture, a single request spans multiple services. Without a correlation ID, you cannot connect logs across services. Generate a unique trace_id at the entry point and propagate it through all downstream calls via HTTP headers.
4. Logging Too Much at INFO Level
Logging every database query, cache hit, and function call at INFO level generates gigabytes of noise, making it hard to find real issues. Use DEBUG for detailed diagnostics and reserve INFO for meaningful business events (user signup, order placed, payment processed).
5. Not Managing Log Volume
High-traffic applications generate terabytes of logs daily. Implement sampling for debug logs, set retention policies, and use log levels to filter in production. Store critical logs (ERROR, WARN) longer and rotate verbose logs frequently.
6. Blocking the Application with Synchronous Logging
Synchronous disk writes in the request path add latency. Use async logging (Pino defaults to async, Python's QueueHandler) or log to stdout and let the container runtime handle log shipping. Never log inside performance-critical hot paths.
Practice Questions
1. Why is structured JSON logging better than plain text logging?
JSON logs are machine-readable, searchable, and parseable by log aggregation tools (Elasticsearch, Loki). Each field is queryable independently. Plain text logs require regex parsing and are prone to format drift across services.
2. What is a correlation ID and why is it important?
A correlation ID (or trace ID) is a unique identifier attached to a request at the entry point and propagated to all downstream services. It allows operators to search for a single ID and see every log entry across all services involved in handling that request.
3. How do log levels help in production debugging?
Log levels control verbosity. INFO shows normal operation. WARN shows potential issues. ERROR shows failures. DEBUG shows detailed diagnostics. In production, run at WARN or INFO normally. When debugging a specific issue, temporarily enable DEBUG for the relevant module.
4. What is log sampling and when should you use it?
Log sampling logs only a fraction of events (e.g., 10% of DEBUG messages). It reduces log volume and storage costs. Use sampling for high-volume, low-value logs (health checks, debug traces). Never sample ERROR or WARN logs.
5. Challenge: Design a centralized logging strategy for a Microservices application with 15 services. Each service must: (1) emit structured JSON logs with a correlation ID propagated from the API Gateway (2) redact sensitive fields (passwords, tokens, PII) (3) use appropriate log levels (INFO for business events, DEBUG for diagnostics, ERROR for failures) (4) include request duration, status code, and user ID for every request. The logs must be shipped to Elasticsearch via Filebeat and visualized in Kibana. Configure a 30-day retention for ERROR logs and 7-day for INFO logs. Set up alerts for ERROR rate exceeding 1% per service.
Mini Project: Log Query CLI
# log_query.py
# Command-line log query tool for Elasticsearch
import JSON
from datetime import datetime, timedelta
from typing import Optional, List
class LogQuery:
"""Query and display logs from Elasticsearch."""
def __init__(self, es_host: str = "HTTP://localhost:9200"):
self.es_host = es_host
def query(
self,
query_string: str,
level: Optional[str] = None,
service: Optional[str] = None,
hours_back: int = 1,
limit: int = 50,
) -> List[dict]:
"""Query logs with filters."""
import requests
must_conditions = [
{"query_string": {"query": query_string}}
]
if level:
must_conditions.append({"term": {"level": level}})
if service:
must_conditions.append({"term": {"service": service}})
es_query = {
"query": {
"bool": {
"must": must_conditions,
"filter": [
{
"range": {
"@timestamp": {
"gte": f"now-{hours_back}h",
"lt": "now",
}
}
}
],
}
},
"sort": [{"@timestamp": "desc"}],
"size": limit,
}
response = requests.post(
f"{self.es_host}/_search",
JSON=es_query,
headers={"Content-Type": "application/JSON"},
)
response.raise_for_status()
data = response.JSON()
logs = []
for hit in data.get('hits', {}).get('hits', []):
source = hit['_source']
logs.append({
'timestamp': source.get('@timestamp', ''),
'level': source.get('level', ''),
'service': source.get('service', ''),
'message': source.get('message', ''),
'correlation_id': source.get('correlation_id', ''),
'duration_ms': source.get('duration_ms', ''),
'status_code': source.get('status_code', ''),
})
return logs
def display(self, logs: List[dict]):
"""Format and print log entries."""
print(f"{'Timestamp':<30} {'Level':<8} {'Service':<20} {'Message'}")
print("=" * 100)
for log in logs:
ts = log['timestamp'][:19] if log['timestamp'] else ''
level = log['level']
service = log['service'][:20]
message = log['message'][:50]
print(f"{ts:<30} {level:<8} {service:<20} {message}")
def error_summary(self, hours_back: int = 24):
"""Show error count by service for the last N hours."""
import requests
es_query = {
"query": {
"bool": {
"filter": [
{"term": {"level": "ERROR"}},
{"range": {"@timestamp": {"gte": f"now-{hours_back}h"}},
]
}
},
"aggs": {
"by_service": {
"terms": {"field": "service.keyword", "size": 20},
"aggs": {
"by_error": {
"terms": {"field": "message.keyword", "size": 5}
}
}
}
},
"size": 0,
}
response = requests.post(
f"{self.es_host}/_search", JSON=es_query,
headers={"Content-Type": "application/JSON"},
)
data = response.JSON()
print(f"=== Error Summary (last {hours_back}h) ===")
for bucket in data['aggregations']['by_service']['buckets']:
print(f"\n{bucket['key']}: {bucket['doc_count']} errors")
for error in bucket['by_error']['buckets']:
print(f" - {error['key']}: {error['doc_count']} times")
# Usage
lq = LogQuery()
logs = lq.query('payment', level='ERROR', service='payment-API', hours_back=24)
lq.display(logs)
FAQ
Related Concepts
What's Next
You now understand backend logging patterns. Next, learn about health check endpoints for monitoring service health, then explore environment configuration for managing log levels per environment.
- Practice daily -- Convert your application's logging from print/console.log to structured JSON logging
- Build a project -- Build a centralized log query tool that searches across multiple services and displays correlated log entries
- Explore related topics -- Check out OpenTelemetry for distributed tracing and the ELK Stack for log aggregation
Built by the developers of Doda Browser, DodaZIP, and Durga Antivirus Pro.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro