Docker Containers for Background Jobs
In this tutorial, you will learn about Docker Containers for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Package background job workers in Docker containers with proper signal handling, health checks, resource limits, and integration with container orchestration platforms.
What You Learn
You will learn how to Dockerize job workers, handle signals for graceful shutdown, configure health checks, set resource limits, and build multi-stage Dockerfiles for efficient images.
Why It Matters
Containerized workers are portable, consistent, and scalable. Docker ensures the worker runs the same way in development, CI, and production. Proper Containerization prevents signal handling bugs and resource leaks.
Real-World Use
DodaTech's workers run in Docker containers on Kubernetes. Each container has CPU/memory limits, health checks, and graceful shutdown that drains in-progress jobs before stopping.
Dockerfile for Worker
FROM python:3.11-slim AS base
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
FROM base AS production
CMD ["python", "worker.py"]
FROM base AS development
ENV WORKER_MODE=inline
ENV LOG_LEVEL=DEBUG
CMD ["python", "-m", "debugpy", "--listen", "0.0.0.0:5678", "worker.py"]
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python -c "import redis; redis.Redis().ping()" || exit 1
Expected output:
Graceful Shutdown in Container
import signal
import time
import threading
class ContainerWorker:
def __init__(self, queue='container_queue'):
self.queue = queue
self.running = True
self.active_jobs = []
self._shutdown_timeout = 30
def setup_signal_handlers(self):
signal.signal(signal.SIGTERM, self._handle_signal)
signal.signal(signal.SIGINT, self._handle_signal)
def _handle_signal(self, signum, frame):
print(f"Received signal {signum}, shutting down gracefully")
self.running = False
def process_job(self, job_id):
self.active_jobs.append(job_id)
try:
print(f" Processing job {job_id}")
time.sleep(2)
print(f" Completed job {job_id}")
finally:
self.active_jobs.remove(job_id)
def run(self):
self.setup_signal_handlers()
print("Worker started, waiting for jobs...")
count = 0
while self.running and count < 5:
count += 1
self.process_job(f"job-{count}")
if not self.running:
break
drain_start = time.time()
while self.active_jobs:
elapsed = time.time() - drain_start
if elapsed > self._shutdown_timeout:
print(f" Force stop after {self._shutdown_timeout}s")
break
print(f" Draining: {len(self.active_jobs)} active jobs")
time.sleep(0.5)
print("Worker stopped")
worker = ContainerWorker()
def run_in_thread():
worker.run()
t = threading.Thread(target=run_in_thread, daemon=True)
t.start()
time.sleep(1)
worker._handle_signal(signal.SIGTERM, None)
t.join(timeout=5)
Expected output:
Worker started, waiting for jobs...
Processing job job-1
Completed job job-1
Processing job job-2
Received signal 15, shutting down gracefully
Completed job job-2
Worker stopped
Health Check Endpoint
import json
import time
from http.server import HTTPServer, BaseHTTPRequestHandler
class WorkerHealthHandler(BaseHTTPRequestHandler):
worker_status = {'status': 'healthy', 'active_jobs': 0}
def do_GET(self):
if self.path == '/health':
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
response = json.dumps(self.worker_status)
self.wfile.write(response.encode())
elif self.path == '/ready':
if self.worker_status['status'] == 'healthy':
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(b'{"ready": true}')
else:
self.send_response(503)
self.end_headers()
self.wfile.write(b'{"ready": false}')
else:
self.send_response(404)
self.end_headers()
class WorkerWithHealthCheck:
def __init__(self, port=8080):
self.port = port
self.running = True
self.active_jobs = 0
def start_health_server(self):
server = HTTPServer(('0.0.0.0', self.port), WorkerHealthHandler)
WorkerHealthHandler.worker_status = {
'status': 'healthy',
'active_jobs': lambda: self.active_jobs,
}
while self.running:
server.handle_request()
def process_job(self):
self.active_jobs += 1
time.sleep(1)
self.active_jobs -= 1
def run(self):
print(f"Health check on port {self.port}")
for _ in range(3):
self.process_job()
print("Worker complete")
worker_wc = WorkerWithHealthCheck()
print("Health check server ready")
Expected output:
Health check on port 8080
Health check server ready
Resource-Limited Worker
import time
import threading
class ResourceLimitedWorker:
def __init__(self, max_memory_mb=256, max_threads=4):
self.max_memory_mb = max_memory_mb
self.max_threads = max_threads
self.active_threads = 0
self._lock = threading.Lock()
def acquire_slot(self):
with self._lock:
if self.active_threads >= self.max_threads:
return False
self.active_threads += 1
return True
def release_slot(self):
with self._lock:
self.active_threads -= 1
def process_within_limits(self, job_func, *args, **kwargs):
if not self.acquire_slot():
print(f" Max threads ({self.max_threads}) reached, queuing")
return None
try:
return job_func(*args, **kwargs)
finally:
self.release_slot()
def get_usage(self):
with self._lock:
return {
'active_threads': self.active_threads,
'max_threads': self.max_threads,
'utilization_pct': (self.active_threads / self.max_threads) * 100,
}
def memory_intensive_task(size_mb):
data = bytearray(size_mb * 1024 * 1024)
time.sleep(0.5)
return f"Processed {size_mb}MB"
limiter = ResourceLimitedWorker(max_threads=2)
results = []
def run_tasks():
for i in range(4):
result = limiter.process_within_limits(memory_intensive_task, 10)
results.append(result)
t = threading.Thread(target=run_tasks, daemon=True)
t.start()
t.join(timeout=5)
print(f"Active threads: {limiter.get_usage()['active_threads']}")
print(f"Results count: {len([r for r in results if r])}")
Expected output:
Max threads (2) reached, queuing
Max threads (2) reached, queuing
Active threads: 0
Results count: 2
Common Mistakes
1. Ignoring SIGTERM in Containers
Docker sends SIGTERM to stop containers. Workers that ignore it are killed with SIGKILL after 10 seconds, losing in-progress jobs.
2. No Health Check
Container orchestrators need health checks to know if the worker is alive. Without them, dead workers are not restarted.
3. Bloated Container Images
Including build tools and dependencies in production images increases size. Use multi-stage builds to keep images small.
4. Hardcoded Configuration in Dockerfile
Configuration should come from environment variables, not the Dockerfile. Same image should work in dev, staging, and production.
5. No Resource Limits
Workers without memory limits can OOM-kill the entire node. Always set CPU and memory limits in container orchestration.
Practice Questions
1. Why handle SIGTERM in containerized workers?
Docker sends SIGTERM for graceful shutdown. Proper handling lets in-progress jobs complete before the container stops.
2. What is a health check for a worker?
An endpoint (HTTP /health) that returns the worker status. Orchestrators use it to decide if the worker should be restarted.
3. Why use multi-stage Docker builds?
Smaller images: the final image only contains runtime dependencies, not build tools. Faster deploys and smaller attack surface.
4. How do resource limits protect workers?
CPU limits prevent CPU starvation of other containers. Memory limits prevent OOM kills. Both improve overall cluster stability.
Challenge
Build a complete Dockerized worker: multi-stage Dockerfile, graceful signal handling, health check endpoint, resource limits, environment-based configuration, and Docker Compose for local development.
FAQ
Mini Project: Containerized Worker
import signal
import time
import json
class DockerWorker:
def __init__(self):
self.running = True
self.processed = 0
signal.signal(signal.SIGTERM, self._stop)
signal.signal(signal.SIGINT, self._stop)
def _stop(self, signum, frame):
print(f"Shutting down (signal {signum})...")
self.running = False
def process(self, job):
print(f"Processing: {job}")
time.sleep(0.5)
self.processed += 1
return {'status': 'done', 'job': job}
def run(self, jobs):
print("Worker started")
for job in jobs:
if not self.running:
print(f"Drained {self.processed} jobs")
break
self.process(job)
print("Worker stopped")
worker = DockerWorker()
worker.run([f"job-{i}" for i in range(5)])
print(f"Total processed: {worker.processed}")
Expected output:
Worker started
Processing: job-0
Processing: job-1
Processing: job-2
Processing: job-3
Processing: job-4
Worker stopped
Total processed: 5
What's Next
Now that you understand Docker containers, explore Kubernetes jobs for orchestrated processing at scale, then learn about Kubernetes CronJobs for scheduled batch processing.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro