Job Monitoring and Alerting — Complete Guide
In this tutorial, you will learn about Job Monitoring and Alerting. We cover key concepts, practical examples, and best practices to help you master this topic.
Set up comprehensive monitoring and alerting for background jobs with Prometheus alerts, Grafana dashboards, Slack notifications, and PagerDuty integration.
What You Learn
You will learn how to define alerting rules for job systems, set up notification channels, build monitoring dashboards, and implement on-call escalation for job failures.
Why It Matters
Without alerting, job failures go unnoticed until users complain. Monitoring and alerting detect issues early: queue backups, error spikes, worker outages, and missed schedules.
Real-World Use
DodaTech's job monitoring alerts on: queue depth > 10K for 5 minutes, error rate > 5% for 2 minutes, no jobs processed for 10 minutes, and any job in DLQ for > 1 hour.
Alert Configuration
import time
import json
import threading
class AlertRule:
def __init__(self, name, metric, condition, threshold, duration, severity):
self.name = name
self.metric = metric
self.condition = condition
self.threshold = threshold
self.duration = duration
self.severity = severity
self.breach_start = None
self.firing = False
def evaluate(self, value):
now = time.time()
breached = False
if self.condition == '>':
breached = value > self.threshold
elif self.condition == '<':
breached = value < self.threshold
elif self.condition == '==':
breached = value == self.threshold
if breached:
if self.breach_start is None:
self.breach_start = now
elif now - self.breach_start >= self.duration and not self.firing:
self.firing = True
return True
else:
self.breach_start = None
self.firing = False
return False
class AlertManager:
def __init__(self):
self.rules = []
self.notifiers = []
def add_rule(self, rule):
self.rules.append(rule)
def add_notifier(self, notifier):
self.notifiers.append(notifier)
def evaluate_all(self, metrics):
fired = []
for rule in self.rules:
value = metrics.get(rule.metric)
if value is not None and rule.evaluate(value):
fired.append(rule)
self._notify(rule, value)
return fired
def _notify(self, rule, value):
for notifier in self.notifiers:
notifier(rule, value)
def slack_notifier(rule, value):
print(f"[SLACK] {rule.severity.upper()}: {rule.name} - {rule.metric}={value} (threshold: {rule.threshold})")
def pagerduty_notifier(rule, value):
if rule.severity == 'critical':
print(f"[PAGERDUTY] Escalating: {rule.name}")
metrics_stream = {
'queue_depth': 12000,
'error_rate': 6.5,
'jobs_processed_5m': 0,
'dlq_count': 5,
}
alerts = AlertManager()
alerts.add_rule(AlertRule('High Queue Depth', 'queue_depth', '>', 10000, 5, 'warning'))
alerts.add_rule(AlertRule('High Error Rate', 'error_rate', '>', 5, 120, 'critical'))
alerts.add_rule(AlertRule('No Processing', 'jobs_processed_5m', '==', 0, 600, 'critical'))
alerts.add_notifier(slack_notifier)
alerts.add_notifier(pagerduty_notifier)
fired = alerts.evaluate_all(metrics_stream)
print(f"Fired alerts: {len(fired)}")
Expected output:
[SLACK] WARNING: High Queue Depth - queue_depth=12000 (threshold: 10000)
[SLACK] CRITICAL: High Error Rate - error_rate=6.5 (threshold: 5)
[PAGERDUTY] Escalating: High Error Rate
[SLACK] CRITICAL: No Processing - jobs_processed_5m=0 (threshold: 0)
[PAGERDUTY] Escalating: No Processing
Fired alerts: 3
Monitoring Dashboard Builder
import time
import json
class MonitorDashboard:
def __init__(self, name='Job Monitor'):
self.name = name
self.panels = []
def add_gauge(self, title, metric, min_val=0, max_val=100):
self.panels.append({
'type': 'gauge',
'title': title,
'metric': metric,
'min': min_val,
'max': max_val,
})
def add_timeseries(self, title, metrics):
self.panels.append({
'type': 'timeseries',
'title': title,
'metrics': metrics,
})
def add_stat(self, title, metric, suffix=''):
self.panels.append({
'type': 'stat',
'title': title,
'metric': metric,
'suffix': suffix,
})
def render_html(self, data):
html = f'<div class="dashboard"><h2>{self.name}</h2><div class="grid">'
for panel in self.panels:
value = data.get(panel['metric'], 'N/A')
if isinstance(value, float):
value = f"{value:.1f}"
html += f'<div class="panel"><h3>{panel["title"]}</h3><div class="value">{value}</div></div>'
html += '</div></div>'
return html
dash = MonitorDashboard('Job Processing Monitor')
dash.add_gauge('Queue Depth', 'queue_depth', 0, 20000)
dash.add_stat('Error Rate', 'error_rate', '%')
dash.add_stat('Active Workers', 'active_workers')
dash.add_timeseries('Throughput', ['jobs_completed', 'jobs_failed'])
data = {'queue_depth': 4500, 'error_rate': 2.3, 'active_workers': 8, 'jobs_completed': 1500}
print(dash.render_html(data))
Expected output:
<div class="dashboard"><h2>Job Processing Monitor</h2><div class="grid">...
Health Check Endpoint
import json
import time
class JobHealthChecker:
def __init__(self):
self.checks = {}
def register_check(self, name, check_func):
self.checks[name] = check_func
def run_all(self):
results = {}
all_healthy = True
for name, check_func in self.checks.items():
try:
healthy, message = check_func()
results[name] = {'healthy': healthy, 'message': message}
if not healthy:
all_healthy = False
except Exception as e:
results[name] = {'healthy': False, 'message': str(e)}
all_healthy = False
return {
'status': 'healthy' if all_healthy else 'unhealthy',
'timestamp': time.time(),
'checks': results,
}
def prometheus_format(self):
output = []
for name, check_func in self.checks.items():
try:
healthy, _ = check_func()
output.append(f'job_health{{check="{name}"}} {1 if healthy else 0}')
except Exception:
output.append(f'job_health{{check="{name}"}} 0')
return '\n'.join(output)
def check_redis():
return True, "Redis connected"
def check_queue_depth():
import redis
r = redis.Redis()
depth = r.llen('default')
healthy = depth < 10000
return healthy, f"Queue depth: {depth}"
def check_workers():
return True, "8 workers active"
hc = JobHealthChecker()
hc.register_check('redis', check_redis)
hc.register_check('queue_depth', lambda: (True, "Depth OK"))
hc.register_check('workers', check_workers)
result = hc.run_all()
print(f"Status: {result['status']}")
print(f"Checks: {len(result['checks'])}")
print(hc.prometheus_format())
Expected output:
Status: healthy
Checks: 3
job_health{check="redis"} 1
job_health{check="queue_depth"} 1
job_health{check="workers"} 1
Notification Channels
import time
import json
class NotificationChannel:
def send(self, title, message, severity='info'):
raise NotImplementedError
class SlackChannel(NotificationChannel):
def __init__(self, webhook_url=None):
self.webhook_url = webhook_url
def send(self, title, message, severity='info'):
emoji = {'info': ':information_source:', 'warning': ':warning:', 'critical': ':red_circle:'}
print(f"[SLACK] {emoji.get(severity, '')} *{title}*: {message}")
class EmailChannel(NotificationChannel):
def send(self, title, message, severity='info'):
print(f"[EMAIL] Subject: {title} ({severity.upper()})")
print(f"[EMAIL] Body: {message}")
class PagerDutyChannel(NotificationChannel):
def send(self, title, message, severity='info'):
if severity in ('critical', 'error'):
print(f"[PAGERDUTY] Trigger: {title} - {message}")
class AlertDispatcher:
def __init__(self):
self.channels = {}
def add_channel(self, name, channel, min_severity='info'):
self.channels[name] = {'channel': channel, 'min_severity': min_severity}
def dispatch(self, title, message, severity='info'):
sent = []
for name, config in self.channels.items():
severity_levels = {'info': 0, 'warning': 1, 'error': 2, 'critical': 3}
if severity_levels.get(severity, 0) >= severity_levels.get(config['min_severity'], 0):
config['channel'].send(title, message, severity)
sent.append(name)
return sent
dispatcher = AlertDispatcher()
dispatcher.add_channel('slack', SlackChannel(), 'info')
dispatcher.add_channel('email', EmailChannel(), 'warning')
dispatcher.add_channel('pagerduty', PagerDutyChannel(), 'critical')
dispatcher.dispatch('Queue Depth Alert', 'Backup queue depth is 15,000', 'warning')
dispatcher.dispatch('Worker Down', 'Worker worker-3 has been offline for 5 minutes', 'critical')
Expected output:
[SLACK] :warning: *Queue Depth Alert*: Backup queue depth is 15,000
[EMAIL] Subject: Queue Depth Alert (WARNING)
[EMAIL] Body: Backup queue depth is 15,000
[SLACK] :red_circle: *Worker Down*: Worker worker-3 has been offline for 5 minutes
[EMAIL] Subject: Worker Down (CRITICAL)
[EMAIL] Body: Worker worker-3 has been offline for 5 minutes
[PAGERDUTY] Trigger: Worker Down - Worker worker-3 has been offline for 5 minutes
Common Mistakes
1. Alert Fatigue
Too many alerts desensitize operators. Only alert on actionable conditions. Use severity levels and suppress noisy alerts.
2. No Duration Before Alerting
A 1-second queue spike does not warrant an alert. Require conditions to persist for a duration before firing.
3. Missing Alert on Zero Activity
No jobs processed for 10 minutes is often more critical than high queue depth. Monitor for absence of expected activity.
4. Not Testing Alerts
Alerts that never fired are not tested. Schedule periodic test alerts to verify notification channels.
5. Too Many Metrics
Monitoring 100 metrics is overwhelming. Focus on 5-10 key signals: queue depth, error rate, throughput, latency, worker count.
Practice Questions
1. What metrics should every job system monitor?
Queue depth, error rate, job throughput, P95 latency, active worker count, and dead letter queue depth.
2. Why add duration to alert conditions?
To avoid false positives from transient spikes. A condition must persist for N seconds before alerting.
3. What is alert fatigue and how to prevent it?
Operators ignoring alerts due to too many false positives. Prevent by tuning thresholds, adding duration, and using severity levels.
4. How do notification channels differ by severity?
Info: Slack. Warning: Slack + Email. Critical: Slack + Email + PagerDuty (on-call). Escalate based on severity.
Challenge
Build a monitoring system for job processing: 5 key metrics with alerting rules, Slack and email notifications, Grafana-style dashboard, health check endpoint for Prometheus, and on-call escalation for critical alerts.
FAQ
Mini Project: Monitoring System
import time
import json
class Monitor:
def __init__(self):
self.rules = []
self.alerts = []
def rule(self, name, condition, threshold, severity='info'):
self.rules.append({'name': name, 'cond': condition, 'threshold': threshold, 'severity': severity})
def check(self, metrics):
for rule in self.rules:
val = metrics.get(rule['cond'])
if val is not None and val > rule['threshold']:
alert = {'rule': rule['name'], 'value': val, 'severity': rule['severity'], 'time': time.time()}
self.alerts.append(alert)
print(f"[{rule['severity'].upper()}] {rule['name']}: {val}")
m = Monitor()
m.rule('High Queue Depth', 'queue_depth', 10000, 'warning')
m.rule('High Error Rate', 'error_rate', 5, 'critical')
m.check({'queue_depth': 15000, 'error_rate': 3, 'throughput': 100})
m.check({'queue_depth': 8000, 'error_rate': 8, 'throughput': 100})
print(f"Total alerts: {len(m.alerts)}")
Expected output:
[WARNING] High Queue Depth: 15000
[CRITICAL] High Error Rate: 8
Total alerts: 2
What's Next
Now that you understand monitoring, explore structured logging for jobs for better Observability, then learn about job performance scaling.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro