Kubernetes CronJobs for Scheduled Jobs
In this tutorial, you will learn about Kubernetes CronJobs for Scheduled Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Schedule Background Jobs using Kubernetes CronJobs with proper configuration, concurrency policies, history limits, and integration with job queues.
What You Learn
You will learn how to create and manage Kubernetes CronJobs, configure concurrency policies, set history limits, handle job failures, and integrate CronJobs with existing job queues.
Why It Matters
Kubernetes CronJobs provide a managed, scalable alternative to traditional cron. They handle scheduling, concurrency control, history management, and integration with K8s logging and monitoring.
Real-World Use
DodaTech uses CronJobs for: database backups every 6 hours, cache warming every 30 minutes, log rotation daily, and malware signature updates every 2 hours. Each CronJob has resource limits and concurrency policies.
CronJob YAML Configuration
apiVersion: batch/v1
kind: CronJob
metadata:
name: database-backup
spec:
schedule: "0 */6 * * *"
startingDeadlineSeconds: 300
concurrencyPolicy: Forbid
successfulJobsHistoryLimit: 3
failedJobsHistoryLimit: 1
suspend: false
jobTemplate:
spec:
template:
spec:
containers:
- name: backup
image: dodatech/backup:latest
command: ["python", "backup.py"]
env:
- name: BACKUP_TYPE
value: "full"
resources:
requests:
cpu: "500m"
memory: "256Mi"
limits:
cpu: "1000m"
memory: "512Mi"
restartPolicy: OnFailure
Expected output:
CronJob Concurrency Policies
import time
import threading
class CronJobHandler:
def __init__(self, name, concurrency_policy='Forbid'):
self.name = name
self.concurrency_policy = concurrency_policy
self.current_job = None
self._lock = threading.Lock()
def should_run(self):
with self._lock:
if self.concurrency_policy == 'Forbid':
if self.current_job is not None and self.current_job['status'] == 'running':
print(f"[{self.name}] Skipped: previous still running (Forbid)")
return False
elif self.concurrency_policy == 'Replace':
if self.current_job is not None and self.current_job['status'] == 'running':
print(f"[{self.name}] Replacing previous job (Replace)")
self.current_job['cancel'] = True
elif self.concurrency_policy == 'Allow':
print(f"[{self.name}] Allowing concurrent execution (Allow)")
self.current_job = {'status': 'running', 'started': time.time()}
return True
def complete(self, success=True):
with self._lock:
if self.current_job:
self.current_job['status'] = 'completed' if success else 'failed'
self.current_job['duration'] = time.time() - self.current_job['started']
def run_job(self, func, *args, **kwargs):
if not self.should_run():
return False
try:
result = func(*args, **kwargs)
self.complete(True)
return result
except Exception as e:
self.complete(False)
raise
def long_running_backup(duration=2):
time.sleep(duration)
return "Backup complete"
handler = CronJobHandler('backup', concurrency_policy='Forbid')
# First run
result1 = handler.run_job(long_running_backup, 1)
print(f"First run: {result1}")
# Second run while first is running would be skipped
Expected output:
First run: Backup complete
CronJob with Deadline
import time
import threading
class CronJobWithDeadline:
def __init__(self, name, deadline_seconds=300):
self.name = name
self.deadline_seconds = deadline_seconds
def execute(self, func, *args, **kwargs):
result_container = []
exception_container = []
def target():
try:
result = func(*args, **kwargs)
result_container.append(result)
except Exception as e:
exception_container.append(e)
t = threading.Thread(target=target, daemon=True)
t.start()
t.join(timeout=self.deadline_seconds)
if t.is_alive():
print(f"[{self.name}] Job exceeded deadline ({self.deadline_seconds}s)")
return {'status': 'deadline_exceeded'}
if exception_container:
raise exception_container[0]
return {'status': 'completed', 'result': result_container[0] if result_container else None}
def should_start(self, scheduled_time, now):
elapsed = now - scheduled_time
if elapsed > self.deadline_seconds:
print(f"[{self.name}] Skipping: {elapsed:.0f}s past deadline")
return False
return True
cj = CronJobWithDeadline('nightly-cleanup', deadline_seconds=120)
def cleanup_old_files():
time.sleep(1)
return "Cleaned 150 files"
# Simulating running at scheduled time
scheduled = time.time()
time.sleep(0.5)
now = time.time()
if cj.should_start(scheduled, now):
result = cj.execute(cleanup_old_files)
print(f"Result: {result}")
Expected output:
Result: {'status': 'completed', 'result': 'Cleaned 150 files'}
CronJob History Management
import time
import json
class CronJobHistory:
def __init__(self, successful_limit=3, failed_limit=1):
self.successful_limit = successful_limit
self.failed_limit = failed_limit
self.history = []
def record_run(self, job_name, success, duration, details=None):
entry = {
'job': job_name,
'timestamp': time.time(),
'success': success,
'duration': duration,
'details': details or {},
}
self.history.append(entry)
self._cleanup()
def _cleanup(self):
successful = [h for h in self.history if h['success']]
failed = [h for h in self.history if not h['success']]
if len(successful) > self.successful_limit:
for h in successful[:-self.successful_limit]:
self.history.remove(h)
if len(failed) > self.failed_limit:
for h in failed[:-self.failed_limit]:
self.history.remove(h)
def get_recent_successes(self, count=3):
successes = [h for h in self.history if h['success']]
return successes[-count:]
def get_recent_failures(self, count=1):
failures = [h for h in self.history if not h['success']]
return failures[-count:]
def get_stats(self):
total = len(self.history)
successes = sum(1 for h in self.history if h['success'])
failures = total - successes
return {
'total_runs': total,
'successes': successes,
'failures': failures,
'success_rate': (successes / total * 100) if total > 0 else 0,
}
history = CronJobHistory(successful_limit=3, failed_limit=1)
history.record_run('backup', True, 45.2)
history.record_run('backup', True, 42.1)
history.record_run('backup', False, 60.0, {'error': 'timeout'})
history.record_run('backup', True, 44.0)
history.record_run('backup', True, 46.5)
print(json.dumps(history.get_stats(), indent=2))
print(f"Recent failures: {len(history.get_recent_failures())}")
Expected output:
{
"total_runs": 5,
"successes": 4,
"failures": 1,
"success_rate": 80.0
}
Recent failures: 1
Common Mistakes
1. No Concurrency Policy
Default is Allow, which can cause overlapping job executions. Use Forbid for database operations and Replace for idempotent tasks.
2. No Deadline
Jobs that hang without deadline run forever. Set startingDeadlineSeconds to limit how late a job can start.
3. Retaining Too Much History
SuccessfulJobsHistoryLimit and failedJobsHistoryLimit default to 3 and 1. Increase for auditing, but watch for etcd storage limits.
4. No Resource Limits
CronJob pods without limits can consume all cluster resources. Always set CPU and memory requests and limits.
5. Forgetting Timezone
CronJob schedule is based on the kube-controller-manager timezone, usually UTC. Document this and convert if needed.
Practice Questions
1. What concurrency policies does CronJob support?
Allow (multiple can run), Forbid (skip if running), Replace (stop old, start new). Choose based on job idempotency.
2. Why set startingDeadlineSeconds?
If the scheduler misses a schedule (downtime), the job has this long to start. After that, it is skipped as missed.
3. How do history limits affect the cluster?
More history means more completed Pod objects in etcd. High limits increase etcd storage usage. Set reasonable limits.
4. Can CronJob be suspended?
Yes. Set suspend: true. Suspended CronJobs do not create new jobs. Useful for maintenance Windows.
Challenge
Create a CronJob configuration for a data pipeline: database backup (daily at 2 AM, Forbid policy, 30-minute deadline), cache warming (every 30 minutes, Replace policy, 5-minute deadline), report generation (weekly Monday, Allow policy, 1-hour deadline).
FAQ
Mini Project: CronJob Manager
import time
import json
import threading
class CronJobManager:
def __init__(self):
self.jobs = {}
self.history = []
def add(self, name, schedule_minutes, func, concurrency='Forbid'):
self.jobs[name] = {
'interval_minutes': schedule_minutes,
'func': func,
'concurrency': concurrency,
'last_run': 0,
'running': False,
}
def check_and_run(self):
now = time.time()
for name, job in self.jobs.items():
interval = job['interval_minutes'] * 60
if now - job['last_run'] >= interval:
if job['concurrency'] == 'Forbid' and job['running']:
print(f"[{name}] Skipped (Forbid)")
continue
job['running'] = True
job['last_run'] = now
start = time.time()
try:
job['func']()
duration = time.time() - start
self.history.append({'name': name, 'success': True, 'duration': duration})
print(f"[{name}] Completed in {duration:.1f}s")
except Exception as e:
self.history.append({'name': name, 'success': False, 'error': str(e)})
print(f"[{name}] Failed: {e}")
finally:
job['running'] = False
manager = CronJobManager()
manager.add('health_check', 1, lambda: time.sleep(0.2), 'Allow')
manager.add('backup', 2, lambda: time.sleep(0.3), 'Forbid')
for _ in range(3):
manager.check_and_run()
time.sleep(1)
stats = {'total_jobs': len(manager.jobs), 'runs': len(manager.history)}
print(json.dumps(stats))
Expected output:
[health_check] Completed in 0.2s
[backup] Completed in 0.3s
[health_check] Completed in 0.2s
{"total_jobs": 2, "runs": ...}
What's Next
Now that you understand Kubernetes CronJobs, explore job security permissions for securing workers, then learn about advanced failure handling.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro