Job Metrics with Prometheus — Complete Guide
In this tutorial, you will learn about Job Metrics with Prometheus. We cover key concepts, practical examples, and best practices to help you master this topic.
Export background job metrics to Prometheus including job duration histograms, queue depth gauges, success/failure counters, and worker utilization metrics.
What You Learn
You will learn how to instrument job processing with Prometheus metrics, expose metrics endpoints, build dashboards for job health, and set up alerting rules.
Why It Matters
Without metrics, you are flying blind. Prometheus metrics provide visibility into job throughput, latency, error rates, and queue depth. This data drives alerting, capacity planning, and performance optimization.
Real-World Use
DodaTech monitors 50+ job types with Prometheus. Each queue has depth, enqueue rate, and latency histograms. Alerts fire when queue depth exceeds 10K or failure rate exceeds 5%.
Prometheus Metrics Architecture
flowchart LR
W[Worker] -->|Increment Counter| P[Prometheus]
W -->|Observe Histogram| P
W -->|Set Gauge| P
P -->|Scrape| E[Exporter]
E -->|/metrics| G[Grafana]
P -->|Alert| AM[Alertmanager]
AM -->|Notify| S[Slack/Pager]
Metrics Instrumentation
import time
import random
import threading
from prometheus_client import Counter, Histogram, Gauge, start_http_server
job_counter = Counter(
'jobs_total',
'Total jobs processed',
['queue', 'status']
)
job_duration = Histogram(
'job_duration_seconds',
'Job processing duration in seconds',
['queue'],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0]
)
queue_depth = Gauge(
'queue_depth',
'Current queue depth',
['queue']
)
active_workers = Gauge(
'active_workers',
'Number of active workers',
['worker_pool']
)
class InstrumentedWorker:
def __init__(self, queue_name='default'):
self.queue_name = queue_name
def process(self, job_func, *args, **kwargs):
job_counter.labels(queue=self.queue_name, status='started').inc()
start = time.time()
try:
result = job_func(*args, **kwargs)
duration = time.time() - start
job_duration.labels(queue=self.queue_name).observe(duration)
job_counter.labels(queue=self.queue_name, status='success').inc()
return result
except Exception as e:
duration = time.time() - start
job_duration.labels(queue=self.queue_name).observe(duration)
job_counter.labels(queue=self.queue_name, status='failed').inc()
raise
def update_queue_depth(self, depth):
queue_depth.labels(queue=self.queue_name).set(depth)
def set_active_workers(self, count):
active_workers.labels(worker_pool=self.queue_name).set(count)
worker = InstrumentedWorker('email_queue')
worker.update_queue_depth(150)
worker.set_active_workers(4)
def send_email():
time.sleep(random.uniform(0.1, 0.3))
return 'sent'
for _ in range(5):
try:
worker.process(send_email)
except Exception:
pass
print(f"Jobs total: {job_counter._metrics['jobs_total']._value.get()}")
print(f"Queue depth: {queue_depth._metrics['queue_depth']._value.get()}")
Expected output:
Jobs total: ...
Queue depth: 150.0
Metrics Exporter
import time
import random
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
class MetricsRegistry:
def __init__(self):
self.counters = {}
self.gauges = {}
self.histograms = {}
def counter(self, name, help_text, labels=None):
if name not in self.counters:
self.counters[name] = {
'help': help_text,
'values': {},
'labels': labels or [],
}
return self.counters[name]
def gauge(self, name, help_text, labels=None):
if name not in self.gauges:
self.gauges[name] = {
'help': help_text,
'values': {},
'labels': labels or [],
}
return self.gauges[name]
def histogram(self, name, help_text, labels=None, buckets=None):
if name not in self.histograms:
self.histograms[name] = {
'help': help_text,
'values': {},
'labels': labels or [],
'buckets': buckets or [0.1, 0.5, 1.0, 5.0, 10.0],
}
return self.histograms[name]
registry = MetricsRegistry()
def inc_counter(name, value=1, **label_values):
metric = registry.counters[name]
label_key = tuple(sorted(label_values.items()))
current = metric['values'].get(label_key, 0)
metric['values'][label_key] = current + value
def set_gauge(name, value, **label_values):
metric = registry.gauges[name]
label_key = tuple(sorted(label_values.items()))
metric['values'][label_key] = value
def observe_histogram(name, value, **label_values):
metric = registry.histograms[name]
label_key = tuple(sorted(label_values.items()))
if label_key not in metric['values']:
metric['values'][label_key] = {'sum': 0, 'count': 0, 'buckets': {}}
data = metric['values'][label_key]
data['sum'] += value
data['count'] += 1
for bucket in metric['buckets']:
if value <= bucket:
key = f'le_{bucket}'
data['buckets'][key] = data['buckets'].get(key, 0) + 1
def generate_metrics():
output = []
for name, metric in registry.counters.items():
output.append(f"# HELP {name} {metric['help']}")
output.append(f"# TYPE {name} counter")
for labels, value in metric['values'].items():
label_str = ','.join(f'{k}="{v}"' for k, v in labels)
output.append(f'{name}{{{label_str}}} {value}')
for name, metric in registry.gauges.items():
output.append(f"# HELP {name} {metric['help']}")
output.append(f"# TYPE {name} gauge")
for labels, value in metric['values'].items():
label_str = ','.join(f'{k}="{v}"' for k, v in labels)
output.append(f'{name}{{{label_str}}} {value}')
return '\n'.join(output)
# Register and update metrics
registry.counter('jobs_processed_total', 'Total jobs processed', labels=['queue', 'status'])
registry.gauge('queue_depth', 'Current queue depth', labels=['queue'])
registry.histogram('job_duration_seconds', 'Job duration', labels=['queue'])
inc_counter('jobs_processed_total', queue='default', status='success')
inc_counter('jobs_processed_total', queue='email', status='success')
set_gauge('queue_depth', 42, queue='default')
observe_histogram('job_duration_seconds', 0.5, queue='default')
print(generate_metrics())
Expected output:
# HELP jobs_processed_total Total jobs processed
# TYPE jobs_processed_total counter
jobs_processed_total{queue="default",status="success"} 1
jobs_processed_total{queue="email",status="success"} 1
# HELP queue_depth Current queue depth
# TYPE queue_depth gauge
queue_depth{queue="default"} 42
...
Grafana Dashboard Queries
# PromQL query examples for job monitoring:
# Job throughput per queue
# rate(jobs_total[5m])
# P95 job duration
# histogram_quantile(0.95, rate(job_duration_seconds_bucket[5m]))
# Queue depth trend
# queue_depth
# Error rate per queue
# rate(jobs_total{status="failed"}[5m]) / rate(jobs_total[5m]) * 100
# Active workers
# active_workers
# Example: simulate metrics for dashboard testing
import time
class DashboardSimulator:
def __init__(self):
self.metrics = {}
def simulate_throughput(self, queue, jobs_per_second, duration_seconds):
for _ in range(duration_seconds):
inc_counter('jobs_processed_total', queue=queue, status='success')
set_gauge('queue_depth', random.randint(10, 100), queue=queue)
observe_histogram('job_duration_seconds', random.uniform(0.1, 2.0), queue=queue)
time.sleep(1 / jobs_per_second)
sim = DashboardSimulator()
t = threading.Thread(target=sim.simulate_throughput, args=('default', 5, 3), daemon=True)
t.start()
time.sleep(1)
print("Dashboard data simulation running")
Expected output:
Dashboard data simulation running
Alerting Rules
# Prometheus alerting rules for job health:
groups:
- name: job_alerts
rules:
- alert: HighQueueDepth
expr: queue_depth > 10000
for: 5m
labels:
severity: warning
annotations:
summary: "Queue depth exceeds 10K for 5 minutes"
- alert: HighErrorRate
expr: rate(jobs_total{status="failed"}[5m]) / rate(jobs_total[5m]) * 100 > 5
for: 5m
labels:
severity: critical
annotations:
summary: "Job error rate above 5%"
- alert: NoJobProcessing
expr: rate(jobs_total[5m]) == 0
for: 10m
labels:
severity: critical
annotations:
summary: "No jobs processed in 10 minutes"
- alert: JobLatencyHigh
expr: histogram_quantile(0.95, rate(job_duration_seconds_bucket[5m])) > 30
for: 5m
labels:
severity: warning
annotations:
summary: "P95 job duration exceeds 30 seconds"
print("Alerting rules defined for: HighQueueDepth, HighErrorRate, NoJobProcessing, JobLatencyHigh")
Expected output:
Alerting rules defined for: HighQueueDepth, HighErrorRate, NoJobProcessing, JobLatencyHigh
Common Mistakes
1. Too Many Label Combinations
High-cardinality labels (job ID, user ID) explode metric count. Use labels for dimensions with bounded values: queue name, status, job type.
2. Not Using Histograms for Duration
Averages hide outliers. Use histograms with buckets to capture P50, P95, and P99 latencies. Configure buckets based on expected job durations.
3. Missing Error Rate Metrics
Tracking success count without failure count hides error problems. Always track both and compute error rate = failures / total.
4. No Queue Depth Monitoring
Queue depth is the most important signal. Zero depth means idle workers. Growing depth means workers cannot keep up. Alert on both.
5. Metrics Without Dashboards
Raw metrics without visualization are hard to use. Build Grafana dashboards for at-a-glance health: throughput, latency, errors, queue depth.
Practice Questions
1. What Prometheus metric types are useful for job monitoring?
Counter (jobs processed), Gauge (queue depth, active workers), Histogram (job duration). Each serves a different monitoring purpose.
2. Why use histograms instead of averages for duration?
Averages hide outliers. A few very slow jobs can skew the average. Histograms show percentile distributions.
3. What is high cardinality and why avoid it?
High cardinality means many unique label values (like job IDs). It increases metric storage and query cost. Use labels for categories, not identifiers.
4. What alerts should every job system have?
High queue depth, elevated error rate, no processing activity, high latency. These detect common failure modes.
Challenge
Build a Prometheus metrics system for a multi-queue job processor: counters per queue/status, histogram per queue, gauges for queue depth and active workers, and alerting rules for common failure conditions.
FAQ
Mini Project: Metrics System
import time
import random
import threading
class JobMetrics:
def __init__(self):
self.counters = {}
self.gauges = {}
self._lock = threading.Lock()
def counter(self, name, labels=None):
with self._lock:
key = (name, tuple(sorted((labels or {}).items())))
self.counters[key] = self.counters.get(key, 0) + 1
return self.counters[key]
def gauge(self, name, value, labels=None):
with self._lock:
key = (name, tuple(sorted((labels or {}).items())))
self.gauges[key] = value
def snapshot(self):
with self._lock:
return {
'counters': dict(self.counters),
'gauges': dict(self.gauges),
}
metrics = JobMetrics()
def monitored_job(queue):
metrics.counter('jobs_total', {'queue': queue, 'status': 'started'})
time.sleep(random.uniform(0.05, 0.15))
if random.random() < 0.1:
metrics.counter('jobs_total', {'queue': queue, 'status': 'failed'})
else:
metrics.counter('jobs_total', {'queue': queue, 'status': 'success'})
metrics.gauge('queue_depth', random.randint(0, 50), {'queue': queue})
threads = []
for _ in range(20):
t = threading.Thread(target=monitored_job, args=('default',), daemon=True)
threads.append(t)
t.start()
for t in threads:
t.join()
snap = metrics.snapshot()
for (name, labels), value in snap['counters'].items():
if 'jobs_total' in name:
print(f"{name} {dict(labels)}: {value}")
Expected output:
jobs_total {'queue': 'default', 'status': 'started'}: 20
jobs_total {'queue': 'default', 'status': 'success'}: ...
jobs_total {'queue': 'default', 'status': 'failed'}: ...
What's Next
Now that you understand metrics, explore job dashboard for visualization with Bull Board, then learn about job monitoring alerting for production alerting.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro