Structured Logging for Background Jobs
In this tutorial, you will learn about Structured Logging for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement structured logging in background jobs with JSON format, correlation IDs, log levels, context enrichment, and integration with log aggregation systems.
What You Learn
You will learn how to add structured logging to job workers, include job context in log entries, use correlation IDs for tracing, and integrate with log aggregation systems.
Why It Matters
Plain text logs are hard to search and analyze. Structured logs in JSON format enable filtering, aggregation, and dashboarding. Correlation IDs connect related log entries across job executions.
Real-World Use
DodaTech's job workers log every job execution as JSON with: job_id, job_type, queue, worker_id, duration, status, and error. Logs are shipped to Elasticsearch for search and Grafana for dashboarding.
Structured Logger
import json
import time
import logging
import traceback
class StructuredLogger:
def __init__(self, name='job-worker'):
self.name = name
def _base_entry(self):
return {
'logger': self.name,
'timestamp': time.time(),
'service': 'job-worker',
}
def info(self, message, **context):
entry = self._base_entry()
entry['level'] = 'INFO'
entry['message'] = message
entry.update(context)
print(json.dumps(entry))
def warning(self, message, **context):
entry = self._base_entry()
entry['level'] = 'WARNING'
entry['message'] = message
entry.update(context)
print(json.dumps(entry))
def error(self, message, exception=None, **context):
entry = self._base_entry()
entry['level'] = 'ERROR'
entry['message'] = message
if exception:
entry['exception'] = str(exception)
entry['traceback'] = traceback.format_exc()
entry.update(context)
print(json.dumps(entry))
def job_started(self, job_id, job_type, queue, worker_id):
self.info('job_started', job_id=job_id, job_type=job_type,
queue=queue, worker_id=worker_id)
def job_completed(self, job_id, duration_ms, result=None):
self.info('job_completed', job_id=job_id, duration_ms=round(duration_ms, 2),
result=result)
def job_failed(self, job_id, error, attempt=None):
self.error('job_failed', job_id=job_id, attempt=attempt, exception=error)
log = StructuredLogger()
log.job_started('j-001', 'scan_file', 'scans', 'worker-1')
log.job_completed('j-001', 1250, {'threats': 0})
log.job_failed('j-002', TimeoutError('Connection timeout'), attempt=2)
Expected output:
{"logger": "job-worker", "timestamp": ..., "level": "INFO", "message": "job_started", "job_id": "j-001", ...}
{"logger": "job-worker", "timestamp": ..., "level": "INFO", "message": "job_completed", "job_id": "j-001", ...}
{"logger": "job-worker", "timestamp": ..., "level": "ERROR", "message": "job_failed", "job_id": "j-002", ...}
Correlation ID for Job Tracing
import uuid
import time
import json
import threading
class CorrelationContext:
_local = threading.local()
@classmethod
def get(cls):
return getattr(cls._local, 'correlation_id', None)
@classmethod
def set(cls, correlation_id):
cls._local.correlation_id = correlation_id
@classmethod
def generate(cls):
cid = str(uuid.uuid4())
cls.set(cid)
return cid
class LogWithCorrelation(StructuredLogger):
def _base_entry(self):
entry = super()._base_entry()
cid = CorrelationContext.get()
if cid:
entry['correlation_id'] = cid
return entry
def from_job(self, job):
CorrelationContext.set(job.get('correlation_id') or CorrelationContext.generate())
self.job_started(job.get('id'), job.get('type'), job.get('queue'), job.get('worker'))
corr_log = LogWithCorrelation()
CorrelationContext.generate()
corr_log.info('Processing order', order_id='ORD-123', user='alice')
print(f"Correlation ID: {CorrelationContext.get()[:8]}...")
Expected output:
{"logger": "job-worker", "timestamp": ..., "level": "INFO", "message": "Processing order", "correlation_id": "...", "order_id": "ORD-123", ...}
Correlation ID: a1b2c3d4...
Job Context Enricher
import json
import time
import os
import socket
class LogEnricher:
def __init__(self):
self.hostname = socket.gethostname()
self.pid = os.getpid()
self.environment = os.getenv('APP_ENV', 'development')
def enrich(self, log_entry, job_context=None):
log_entry['hostname'] = self.hostname
log_entry['pid'] = self.pid
log_entry['environment'] = self.environment
if job_context:
log_entry['job_id'] = job_context.get('id')
log_entry['job_type'] = job_context.get('type')
log_entry['queue'] = job_context.get('queue')
log_entry['attempt'] = job_context.get('attempt', 1)
return log_entry
def format_job_log(self, message, job_context=None, extra=None):
entry = {
'timestamp': time.time(),
'message': message,
'level': 'INFO',
}
entry = self.enrich(entry, job_context)
if extra:
entry.update(extra)
return json.dumps(entry)
enricher = LogEnricher()
job_ctx = {'id': 'j-003', 'type': 'payment', 'queue': 'payments', 'attempt': 1}
log_line = enricher.format_job_log('Processing payment', job_ctx, {'amount': 50, 'currency': 'USD'})
parsed = json.loads(log_line)
print(f"Host: {parsed['hostname']}, Env: {parsed['environment']}")
print(f"Job: {parsed['job_type']}/{parsed['job_id']}")
Expected output:
Host: ..., Env: development
Job: payment/j-003
Log Aggregation Integration
import json
import time
import threading
import queue
class LogAggregator:
def __init__(self, batch_size=10, flush_interval=5):
self.batch_size = batch_size
self.flush_interval = flush_interval
self._queue = queue.Queue()
self._running = True
self._batches_sent = 0
threading.Thread(target=self._flush_loop, daemon=True).start()
def emit(self, log_entry):
self._queue.put(log_entry)
def _flush_loop(self):
while self._running:
time.sleep(self.flush_interval)
self._flush()
def _flush(self):
batch = []
while not self._queue.empty() and len(batch) < self.batch_size:
batch.append(self._queue.get_nowait())
if batch:
payload = '\n'.join(json.dumps(e) for e in batch)
self._batches_sent += 1
print(f"[AGGREGATOR] Sent batch {self._batches_sent} ({len(batch)} entries)")
def shutdown(self):
self._running = False
self._flush()
class JobLogger:
def __init__(self, aggregator):
self.aggregator = aggregator
def log_job(self, job_id, status, duration_ms, error=None):
entry = {
'timestamp': time.time(),
'type': 'job_event',
'job_id': job_id,
'status': status,
'duration_ms': round(duration_ms, 2),
}
if error:
entry['error'] = str(error)
self.aggregator.emit(entry)
aggregator = LogAggregator(batch_size=5, flush_interval=1)
logger = JobLogger(aggregator)
for i in range(7):
logger.log_job(f'job-{i}', 'completed' if i % 3 else 'failed', i * 100)
time.sleep(2)
aggregator.shutdown()
Expected output:
[AGGREGATOR] Sent batch 1 (5 entries)
[AGGREGATOR] Sent batch 2 (2 entries)
Common Mistakes
1. Logging Sensitive Data
Job logs may contain PII, passwords, or payment data. Sanitize sensitive fields before logging.
2. No Log Rotation
Unbounded log files fill disk space. Configure log rotation by size and retention period.
3. Blocking the Worker with Log I/O
Synchronous log writes slow down job processing. Use async logging or a dedicated log queue.
4. Inconsistent Log Format
Different services log in different formats. Standardize on JSON format with required fields across all workers.
5. Missing Correlation IDs
Without correlation IDs, you cannot trace a job across log entries. Always include job_id and correlation_id.
Practice Questions
1. Why use structured logging (JSON) over plain text?
JSON logs are machine-parseable, searchable, and filterable. Tools like Elasticsearch and Grafana can index and query structured fields.
2. What is a correlation ID?
A unique identifier attached to a job that appears in all related log entries. Enables tracing the job across services and log lines.
3. How do you prevent sensitive data in logs?
Sanitize log entries before writing: mask credit cards, redact passwords, remove PII. Use a log sanitizer middleware.
4. What fields should every job log contain?
timestamp, level, message, job_id, job_type, queue, worker_id, duration_ms, status, correlation_id.
Challenge
Build a logging system for job workers: JSON structured format, correlation IDs per job, context enrichment with host and environment, batch aggregation for performance, and log sanitization for sensitive data.
FAQ
Mini Project: Structured Logger
import json
import time
import threading
class StructuredJobLogger:
def __init__(self):
self._buffer = []
self._lock = threading.Lock()
def log(self, level, message, **kwargs):
entry = {'timestamp': time.time(), 'level': level, 'message': message, **kwargs}
with self._lock:
self._buffer.append(entry)
if level in ('ERROR', 'CRITICAL'):
self.flush()
def flush(self):
with self._lock:
while self._buffer:
print(json.dumps(self._buffer.pop(0)))
def job(self, job_id, event, duration=None, error=None):
entry = {'job_id': job_id, 'event': event}
if duration:
entry['duration_ms'] = round(duration * 1000, 2)
if error:
entry['error'] = str(error)
self.log('INFO', f'job_{event}', **entry)
jlog = StructuredJobLogger()
jlog.job('j-1', 'started')
jlog.job('j-1', 'completed', duration=1.5)
jlog.job('j-2', 'failed', error=ValueError('bad data'))
jlog.flush()
Expected output:
{"timestamp": ..., "level": "INFO", "message": "job_started", "job_id": "j-1", "event": "started"}
{"timestamp": ..., "level": "INFO", "message": "job_completed", "job_id": "j-1", "event": "completed", "duration_ms": 1500.0}
{"timestamp": ..., "level": "INFO", "message": "job_failed", "job_id": "j-2", "event": "failed", "error": "bad data"}
What's Next
Now that you understand structured logging, explore job performance scaling for optimizing throughput, then learn about job cost optimization.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro