Prerequisites
- Basic understanding of programming concepts 📝
- Python installation (3.8+) 🐍
- VS Code or preferred IDE 💻
What you'll learn
- Understand the concept fundamentals 🎯
- Apply the concept in real projects 🏗️
- Debug common issues 🐛
- Write clean, Pythonic code ✨
🎯 Introduction
Welcome to the world of Kubernetes Deployments! 🎉 Today we’re diving into one of the most powerful tools for managing containerized applications at scale.
Kubernetes Deployments are like having a smart assistant that manages your applications 24/7 🤖. They ensure your apps are always running, handle updates smoothly, and scale automatically based on demand. Whether you’re deploying a simple web app or a complex microservices architecture, Deployments are your best friend in the Kubernetes ecosystem!
By the end of this tutorial, you’ll be confidently deploying and managing Python applications in Kubernetes like a pro! Let’s embark on this exciting journey! 🚀
📚 Understanding Kubernetes Deployments
🤔 What are Kubernetes Deployments?
Think of a Kubernetes Deployment as a smart restaurant manager 🍴. Just like a manager ensures there are always enough waiters during busy hours and replaces any who call in sick, a Deployment ensures your application always has the right number of running instances (pods) and replaces any that fail.
In Python terms, a Deployment is a declarative way to manage your application’s lifecycle. You tell Kubernetes what you want (e.g., “I want 3 instances of my Flask app running”), and it figures out how to make that happen and keep it that way! This means you can:
- ✨ Automatically restart crashed applications
- 🚀 Scale up during high traffic
- 🛡️ Roll out updates without downtime
💡 Why Use Kubernetes Deployments?
Here’s why developers love Deployments:
- Self-Healing 🏥: Applications automatically restart if they crash
- Rolling Updates 🔄: Deploy new versions without downtime
- Easy Scaling 📈: Handle traffic spikes effortlessly
- Version Control 📦: Roll back to previous versions instantly
Real-world example: Imagine running an e-commerce site during Black Friday 🛒. With Deployments, your app automatically scales to handle millions of shoppers and recovers instantly if any instance crashes!
🔧 Basic Syntax and Usage
📝 Creating Your First Deployment
Let’s start with a simple Python Flask application deployment:
# 👋 First, let's create a simple Flask app (app.py)
from flask import Flask, jsonify
import os
import socket
app = Flask(__name__)
@app.route('/')
def hello():
# 🎨 Return some useful info about our pod
return jsonify({
'message': 'Hello from Kubernetes! 🚀',
'hostname': socket.gethostname(),
'version': '1.0',
'pod_ip': os.environ.get('POD_IP', 'unknown')
})
@app.route('/health')
def health():
# 🏥 Health check endpoint for Kubernetes
return jsonify({'status': 'healthy! 💪'}), 200
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)Now, let’s create a Dockerfile:
# 🐳 Dockerfile for our Python app
FROM python:3.9-slim
WORKDIR /app
# 📦 Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 🚀 Copy our app
COPY app.py .
# 🎯 Expose port and run
EXPOSE 5000
CMD ["python", "app.py"]🎯 Deployment Manifest
Here’s how to define a Kubernetes Deployment using Python’s kubernetes client:
# 🏗️ deployment.py - Create a Deployment programmatically
from kubernetes import client, config
def create_deployment():
# 🔧 Load Kubernetes config
config.load_incluster_config() # Use this inside cluster
# config.load_kube_config() # Use this for local development
# 🎨 Create API instance
apps_v1 = client.AppsV1Api()
# 📦 Define the deployment
deployment = client.V1Deployment(
metadata=client.V1ObjectMeta(
name="flask-app",
labels={"app": "flask-app"}
),
spec=client.V1DeploymentSpec(
replicas=3, # 🚀 Run 3 instances
selector=client.V1LabelSelector(
match_labels={"app": "flask-app"}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": "flask-app"}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name="flask-app",
image="myregistry/flask-app:1.0",
ports=[client.V1ContainerPort(container_port=5000)],
env=[
# 🌟 Pass pod IP as environment variable
client.V1EnvVar(
name="POD_IP",
value_from=client.V1EnvVarSource(
field_ref=client.V1ObjectFieldSelector(
field_path="status.podIP"
)
)
)
],
# 🏥 Health checks
liveness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/health",
port=5000
),
initial_delay_seconds=30,
period_seconds=10
),
readiness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/health",
port=5000
),
initial_delay_seconds=5,
period_seconds=5
)
)
]
)
)
)
)
# 🚀 Create the deployment
apps_v1.create_namespaced_deployment(
namespace="default",
body=deployment
)
print("✅ Deployment created successfully!")💡 Practical Examples
🛒 Example 1: E-Commerce App with Auto-Scaling
Let’s build a scalable e-commerce backend:
# 🛍️ ecommerce_deployment.py
from kubernetes import client, config
import yaml
class EcommerceDeployment:
def __init__(self, namespace="ecommerce"):
config.load_kube_config()
self.apps_v1 = client.AppsV1Api()
self.autoscaling_v1 = client.AutoscalingV1Api()
self.namespace = namespace
def deploy_app(self, name, image, replicas=2):
# 🎨 Create deployment for microservice
deployment = {
"apiVersion": "apps/v1",
"kind": "Deployment",
"metadata": {
"name": name,
"labels": {"app": name, "tier": "backend"}
},
"spec": {
"replicas": replicas,
"selector": {"matchLabels": {"app": name}},
"template": {
"metadata": {"labels": {"app": name}},
"spec": {
"containers": [{
"name": name,
"image": image,
"ports": [{"containerPort": 8000}],
"resources": {
# 📊 Resource management
"requests": {
"memory": "128Mi",
"cpu": "100m"
},
"limits": {
"memory": "256Mi",
"cpu": "200m"
}
},
"env": [
{"name": "SERVICE_NAME", "value": name},
{"name": "ENVIRONMENT", "value": "production"}
]
}]
}
}
}
}
# 🚀 Deploy it!
self.apps_v1.create_namespaced_deployment(
namespace=self.namespace,
body=deployment
)
print(f"✅ Deployed {name} successfully!")
# 📈 Create HorizontalPodAutoscaler
self.create_autoscaler(name)
def create_autoscaler(self, name):
# 🎯 Auto-scale based on CPU usage
hpa = client.V1HorizontalPodAutoscaler(
metadata=client.V1ObjectMeta(name=f"{name}-hpa"),
spec=client.V1HorizontalPodAutoscalerSpec(
scale_target_ref=client.V1CrossVersionObjectReference(
api_version="apps/v1",
kind="Deployment",
name=name
),
min_replicas=2,
max_replicas=10,
target_cpu_utilization_percentage=70
)
)
self.autoscaling_v1.create_namespaced_horizontal_pod_autoscaler(
namespace=self.namespace,
body=hpa
)
print(f"📈 Created autoscaler for {name}!")
def deploy_full_stack(self):
# 🛒 Deploy all microservices
services = [
("cart-service", "myregistry/cart:1.0", 3),
("product-service", "myregistry/products:1.0", 2),
("order-service", "myregistry/orders:1.0", 2),
("payment-service", "myregistry/payment:1.0", 4)
]
for name, image, replicas in services:
self.deploy_app(name, image, replicas)
print(f"🎉 {name} is up and running!")
# 🎮 Let's deploy!
ecommerce = EcommerceDeployment()
ecommerce.deploy_full_stack()🎮 Example 2: Game Server Fleet Management
Managing game servers with intelligent deployment strategies:
# 🏆 game_server_manager.py
from kubernetes import client, config
from datetime import datetime
import time
class GameServerFleet:
def __init__(self):
config.load_kube_config()
self.apps_v1 = client.AppsV1Api()
self.core_v1 = client.CoreV1Api()
def create_game_server(self, region, server_type="standard"):
# 🎮 Define game server deployment
deployment_name = f"game-server-{region}-{int(time.time())}"
# 🎯 Different configurations for different server types
configs = {
"standard": {"cpu": "500m", "memory": "1Gi", "players": 50},
"premium": {"cpu": "1000m", "memory": "2Gi", "players": 100},
"tournament": {"cpu": "2000m", "memory": "4Gi", "players": 200}
}
config = configs.get(server_type, configs["standard"])
deployment = client.V1Deployment(
metadata=client.V1ObjectMeta(
name=deployment_name,
labels={
"app": "game-server",
"region": region,
"type": server_type,
"created": datetime.now().isoformat()
}
),
spec=client.V1DeploymentSpec(
replicas=1,
selector=client.V1LabelSelector(
match_labels={"app": deployment_name}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": deployment_name}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name="game-server",
image="myregistry/game-server:latest",
ports=[
client.V1ContainerPort(container_port=7777), # Game port
client.V1ContainerPort(container_port=8080) # API port
],
env=[
{"name": "SERVER_REGION", "value": region},
{"name": "MAX_PLAYERS", "value": str(config["players"])},
{"name": "SERVER_TYPE", "value": server_type}
],
resources=client.V1ResourceRequirements(
requests={
"cpu": config["cpu"],
"memory": config["memory"]
},
limits={
"cpu": config["cpu"],
"memory": config["memory"]
}
),
# 🏥 Health checks for game servers
liveness_probe=client.V1Probe(
tcp_socket=client.V1TCPSocketAction(port=7777),
initial_delay_seconds=60,
period_seconds=30
),
readiness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/api/ready",
port=8080
),
initial_delay_seconds=30,
period_seconds=10
)
)
],
# 🎯 Place game servers on dedicated nodes
node_selector={"node-type": "game-server"},
# 🛡️ Ensure servers are spread across nodes
affinity=client.V1Affinity(
pod_anti_affinity=client.V1PodAntiAffinity(
preferred_during_scheduling_ignored_during_execution=[
client.V1WeightedPodAffinityTerm(
weight=100,
pod_affinity_term=client.V1PodAffinityTerm(
label_selector=client.V1LabelSelector(
match_labels={"app": "game-server"}
),
topology_key="kubernetes.io/hostname"
)
)
]
)
)
)
)
)
)
# 🚀 Deploy the game server
self.apps_v1.create_namespaced_deployment(
namespace="game-servers",
body=deployment
)
print(f"🎮 Game server {deployment_name} deployed in {region}!")
return deployment_name
def scale_for_event(self, event_type="normal"):
# 📈 Scale game servers based on events
scaling_profiles = {
"normal": {"us-east": 5, "eu-west": 3, "asia": 2},
"weekend": {"us-east": 10, "eu-west": 8, "asia": 5},
"tournament": {"us-east": 20, "eu-west": 15, "asia": 10}
}
profile = scaling_profiles.get(event_type, scaling_profiles["normal"])
for region, count in profile.items():
# 🎯 Deploy servers for each region
for i in range(count):
server_type = "tournament" if event_type == "tournament" else "standard"
self.create_game_server(region, server_type)
time.sleep(2) # Stagger deployments
print(f"🎊 Scaled fleet for {event_type} event!")
def get_server_status(self):
# 📊 Monitor all game servers
deployments = self.apps_v1.list_namespaced_deployment(
namespace="game-servers",
label_selector="app=game-server"
)
print("🎮 Game Server Fleet Status:")
for deployment in deployments.items:
region = deployment.metadata.labels.get("region", "unknown")
server_type = deployment.metadata.labels.get("type", "standard")
ready = deployment.status.ready_replicas or 0
total = deployment.spec.replicas
status_emoji = "✅" if ready == total else "⚠️"
print(f"{status_emoji} {deployment.metadata.name}: {ready}/{total} ready ({region}, {server_type})")
# 🎮 Manage your game server fleet!
fleet = GameServerFleet()
fleet.scale_for_event("tournament")
fleet.get_server_status()🚀 Advanced Concepts
🧙♂️ Advanced Deployment Strategies
When you’re ready to level up, implement sophisticated deployment patterns:
# 🎯 advanced_deployments.py
from kubernetes import client, config
import time
import hashlib
class AdvancedDeploymentManager:
def __init__(self):
config.load_kube_config()
self.apps_v1 = client.AppsV1Api()
self.core_v1 = client.CoreV1Api()
def blue_green_deployment(self, app_name, new_image, namespace="default"):
"""
🔄 Blue-Green deployment strategy
"""
# 🎨 Create "green" deployment alongside existing "blue"
green_deployment = f"{app_name}-green"
blue_deployment = f"{app_name}-blue"
print(f"🚀 Starting Blue-Green deployment for {app_name}")
# 📦 Deploy green version
deployment = client.V1Deployment(
metadata=client.V1ObjectMeta(
name=green_deployment,
labels={"app": app_name, "version": "green"}
),
spec=client.V1DeploymentSpec(
replicas=3,
selector=client.V1LabelSelector(
match_labels={"app": app_name, "version": "green"}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": app_name, "version": "green"}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name=app_name,
image=new_image,
ports=[client.V1ContainerPort(container_port=8080)]
)
]
)
)
)
)
self.apps_v1.create_namespaced_deployment(namespace=namespace, body=deployment)
print("✅ Green deployment created")
# ⏳ Wait for green deployment to be ready
self.wait_for_deployment(green_deployment, namespace)
# 🎯 Switch traffic to green
self.switch_service_to_green(app_name, namespace)
# 🗑️ Clean up blue deployment
time.sleep(30) # Grace period
self.apps_v1.delete_namespaced_deployment(
name=blue_deployment,
namespace=namespace
)
print("🎉 Blue-Green deployment completed!")
def canary_deployment(self, app_name, new_image, canary_percentage=10):
"""
🐤 Canary deployment with gradual rollout
"""
print(f"🐤 Starting Canary deployment for {app_name}")
# 🎯 Create canary deployment
canary_name = f"{app_name}-canary"
stable_name = f"{app_name}-stable"
# Calculate replicas based on percentage
total_replicas = 10
canary_replicas = max(1, total_replicas * canary_percentage // 100)
stable_replicas = total_replicas - canary_replicas
# 📊 Deploy canary version
canary_deployment = self.create_versioned_deployment(
canary_name, new_image, canary_replicas, "canary"
)
# 📈 Gradually increase canary traffic
steps = [10, 25, 50, 100]
for percentage in steps:
print(f"📊 Canary at {percentage}%")
canary_replicas = total_replicas * percentage // 100
stable_replicas = total_replicas - canary_replicas
# Update replica counts
self.scale_deployment(canary_name, canary_replicas)
self.scale_deployment(stable_name, stable_replicas)
# 🏥 Monitor metrics
if not self.check_canary_health(canary_name):
print("❌ Canary failed health checks, rolling back!")
self.rollback_canary(app_name)
return False
time.sleep(60) # Monitor for 1 minute at each stage
print("✅ Canary deployment successful!")
return True
def create_versioned_deployment(self, name, image, replicas, version):
"""
🏗️ Create a deployment with version labels
"""
deployment = client.V1Deployment(
metadata=client.V1ObjectMeta(
name=name,
labels={"app": name.split("-")[0], "version": version}
),
spec=client.V1DeploymentSpec(
replicas=replicas,
selector=client.V1LabelSelector(
match_labels={"app": name.split("-")[0], "version": version}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": name.split("-")[0], "version": version},
annotations={
"prometheus.io/scrape": "true",
"prometheus.io/port": "9090"
}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name="app",
image=image,
ports=[
client.V1ContainerPort(container_port=8080),
client.V1ContainerPort(container_port=9090, name="metrics")
],
# 🎯 Advanced probes
startup_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(path="/startup", port=8080),
failure_threshold=30,
period_seconds=10
),
liveness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(path="/health", port=8080),
initial_delay_seconds=0,
period_seconds=10,
timeout_seconds=5,
success_threshold=1,
failure_threshold=3
),
readiness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(path="/ready", port=8080),
initial_delay_seconds=0,
period_seconds=5,
timeout_seconds=3,
success_threshold=1,
failure_threshold=3
)
)
]
)
),
# 🚀 Advanced update strategy
strategy=client.V1DeploymentStrategy(
type="RollingUpdate",
rolling_update=client.V1RollingUpdateDeployment(
max_surge="25%",
max_unavailable="0" # Zero downtime
)
),
# 📊 Track deployment progress
progress_deadline_seconds=600,
revision_history_limit=10
)
)
return self.apps_v1.create_namespaced_deployment(
namespace="default",
body=deployment
)
# 🪄 Use advanced deployments
manager = AdvancedDeploymentManager()
manager.canary_deployment("my-app", "myregistry/my-app:2.0")🏗️ GitOps-Style Deployment Management
Implement GitOps patterns with Python:
# 🚀 gitops_deployment.py
import git
import yaml
from kubernetes import client, config
import hashlib
import json
class GitOpsDeploymentManager:
def __init__(self, repo_url, branch="main"):
self.repo_url = repo_url
self.branch = branch
config.load_kube_config()
self.apps_v1 = client.AppsV1Api()
self.repo = None
def sync_deployments(self):
"""
🔄 Sync deployments from Git repository
"""
# 📥 Clone or update repository
self.update_repo()
# 🎯 Find all deployment files
deployment_files = self.find_deployment_files()
for file_path in deployment_files:
print(f"📄 Processing {file_path}")
self.apply_deployment(file_path)
def apply_deployment(self, file_path):
"""
🚀 Apply deployment from YAML file with GitOps annotations
"""
with open(file_path, 'r') as f:
deployment_yaml = yaml.safe_load(f)
# 🏷️ Add GitOps metadata
deployment_yaml['metadata']['annotations'] = {
'gitops.io/repo': self.repo_url,
'gitops.io/branch': self.branch,
'gitops.io/path': file_path,
'gitops.io/commit': self.get_current_commit(),
'gitops.io/sync-time': datetime.now().isoformat()
}
# 🔍 Check if deployment exists
name = deployment_yaml['metadata']['name']
namespace = deployment_yaml['metadata'].get('namespace', 'default')
try:
existing = self.apps_v1.read_namespaced_deployment(name, namespace)
# 📊 Update if changed
if self.has_changed(existing, deployment_yaml):
self.apps_v1.patch_namespaced_deployment(
name=name,
namespace=namespace,
body=deployment_yaml
)
print(f"✅ Updated deployment: {name}")
else:
print(f"⏭️ No changes for: {name}")
except client.exceptions.ApiException as e:
if e.status == 404:
# 🆕 Create new deployment
self.apps_v1.create_namespaced_deployment(
namespace=namespace,
body=deployment_yaml
)
print(f"✨ Created deployment: {name}")
else:
raise
def has_changed(self, existing, desired):
"""
🔍 Check if deployment needs updating
"""
# Generate hash of spec
existing_hash = hashlib.sha256(
json.dumps(existing.spec.to_dict(), sort_keys=True).encode()
).hexdigest()
desired_hash = hashlib.sha256(
json.dumps(desired['spec'], sort_keys=True).encode()
).hexdigest()
return existing_hash != desired_hash
# 🎯 Implement GitOps workflow
gitops = GitOpsDeploymentManager("https://github.com/myorg/k8s-configs")
gitops.sync_deployments()⚠️ Common Pitfalls and Solutions
😱 Pitfall 1: Resource Limits Not Set
# ❌ Wrong way - no resource limits!
container = client.V1Container(
name="my-app",
image="myapp:latest"
# 💥 This pod could consume all node resources!
)
# ✅ Correct way - always set resource limits!
container = client.V1Container(
name="my-app",
image="myapp:latest",
resources=client.V1ResourceRequirements(
requests={
"memory": "256Mi", # 🎯 Minimum guaranteed
"cpu": "250m" # 🎯 250 millicores
},
limits={
"memory": "512Mi", # 🛡️ Maximum allowed
"cpu": "500m" # 🛡️ Prevents resource hogging
}
)
)🤯 Pitfall 2: Missing Health Checks
# ❌ Dangerous - no health checks!
container = client.V1Container(
name="api-server",
image="api:1.0",
ports=[client.V1ContainerPort(container_port=8080)]
)
# 💥 Kubernetes won't know if your app is healthy!
# ✅ Safe - comprehensive health checks!
container = client.V1Container(
name="api-server",
image="api:1.0",
ports=[client.V1ContainerPort(container_port=8080)],
# 🏥 Startup probe for slow-starting apps
startup_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/startup",
port=8080
),
failure_threshold=30,
period_seconds=10
),
# 💗 Liveness probe to detect hangs
liveness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/health",
port=8080
),
initial_delay_seconds=0,
period_seconds=10,
timeout_seconds=5,
failure_threshold=3
),
# ✅ Readiness probe for load balancing
readiness_probe=client.V1Probe(
http_get=client.V1HTTPGetAction(
path="/ready",
port=8080
),
initial_delay_seconds=0,
period_seconds=5,
timeout_seconds=3,
failure_threshold=3
)
)
print("🎉 Your app now has proper health monitoring!")🛠️ Best Practices
- 🎯 Use Declarative Configuration: Define desired state, let Kubernetes handle the rest
- 📊 Set Resource Limits: Always specify CPU and memory limits to prevent resource starvation
- 🏥 Implement Health Checks: Use all three probe types (startup, liveness, readiness)
- 🏷️ Label Everything: Use consistent labeling for easier management and selection
- 🔄 Rolling Updates: Configure proper update strategies to ensure zero downtime
🧪 Hands-On Exercise
🎯 Challenge: Build a Multi-Environment Deployment System
Create a Python-based deployment system that manages applications across different environments:
📋 Requirements:
- ✅ Deploy to dev, staging, and production environments
- 🏷️ Environment-specific configurations (replicas, resources)
- 👤 Implement RBAC for different teams
- 📅 Schedule deployments with maintenance windows
- 🎨 Each environment needs different scaling policies!
🚀 Bonus Points:
- Add deployment approval workflow
- Implement automatic rollback on failures
- Create deployment metrics dashboard
💡 Solution
🔍 Click to see solution
# 🎯 Multi-environment deployment system!
from kubernetes import client, config
from datetime import datetime, timedelta
import time
from typing import Dict, List
import schedule
class MultiEnvironmentDeploymentSystem:
def __init__(self):
config.load_kube_config()
self.apps_v1 = client.AppsV1Api()
self.core_v1 = client.CoreV1Api()
self.rbac_v1 = client.RbacAuthorizationV1Api()
# 🎨 Environment configurations
self.environments = {
"dev": {
"namespace": "development",
"replicas": 1,
"cpu_request": "100m",
"memory_request": "128Mi",
"cpu_limit": "200m",
"memory_limit": "256Mi",
"autoscale": False
},
"staging": {
"namespace": "staging",
"replicas": 2,
"cpu_request": "250m",
"memory_request": "256Mi",
"cpu_limit": "500m",
"memory_limit": "512Mi",
"autoscale": True,
"min_replicas": 2,
"max_replicas": 5
},
"production": {
"namespace": "production",
"replicas": 3,
"cpu_request": "500m",
"memory_request": "512Mi",
"cpu_limit": "1000m",
"memory_limit": "1Gi",
"autoscale": True,
"min_replicas": 3,
"max_replicas": 20,
"maintenance_window": {
"start_hour": 2, # 2 AM
"end_hour": 4 # 4 AM
}
}
}
def deploy_to_environment(self, app_name: str, image: str, env: str):
"""
🚀 Deploy application to specific environment
"""
if env not in self.environments:
raise ValueError(f"❌ Unknown environment: {env}")
env_config = self.environments[env]
# 🛡️ Check maintenance window for production
if env == "production" and not self.in_maintenance_window(env_config):
print("⏰ Waiting for maintenance window...")
self.wait_for_maintenance_window(env_config)
print(f"🚀 Deploying {app_name} to {env} environment")
# 📦 Create deployment
deployment = client.V1Deployment(
metadata=client.V1ObjectMeta(
name=f"{app_name}-{env}",
namespace=env_config["namespace"],
labels={
"app": app_name,
"environment": env,
"version": self.extract_version(image),
"managed-by": "multi-env-system"
},
annotations={
"deployment.kubernetes.io/revision": "1",
"deployed-at": datetime.now().isoformat(),
"deployed-by": "python-deployment-system"
}
),
spec=client.V1DeploymentSpec(
replicas=env_config["replicas"],
selector=client.V1LabelSelector(
match_labels={"app": app_name, "environment": env}
),
template=client.V1PodTemplateSpec(
metadata=client.V1ObjectMeta(
labels={"app": app_name, "environment": env}
),
spec=client.V1PodSpec(
containers=[
client.V1Container(
name=app_name,
image=image,
ports=[client.V1ContainerPort(container_port=8080)],
env=self.get_environment_variables(env),
resources=client.V1ResourceRequirements(
requests={
"cpu": env_config["cpu_request"],
"memory": env_config["memory_request"]
},
limits={
"cpu": env_config["cpu_limit"],
"memory": env_config["memory_limit"]
}
),
# 🏥 Environment-specific probes
liveness_probe=self.get_probe_for_env(env, "liveness"),
readiness_probe=self.get_probe_for_env(env, "readiness")
)
],
# 🎯 Production gets pod disruption budget
**self.get_pod_spec_extras(env)
)
),
strategy=self.get_deployment_strategy(env)
)
)
# 🚀 Create or update deployment
try:
self.apps_v1.create_namespaced_deployment(
namespace=env_config["namespace"],
body=deployment
)
print(f"✅ Created deployment in {env}")
except client.exceptions.ApiException as e:
if e.status == 409: # Already exists
self.apps_v1.patch_namespaced_deployment(
name=f"{app_name}-{env}",
namespace=env_config["namespace"],
body=deployment
)
print(f"✅ Updated deployment in {env}")
else:
raise
# 📈 Create autoscaler if needed
if env_config.get("autoscale", False):
self.create_autoscaler(app_name, env, env_config)
# 🛡️ Set up RBAC for the environment
self.setup_rbac(app_name, env)
# 📊 Monitor deployment progress
self.monitor_deployment(app_name, env, env_config["namespace"])
def get_environment_variables(self, env: str) -> List[client.V1EnvVar]:
"""
🌟 Environment-specific variables
"""
base_vars = [
client.V1EnvVar(name="ENVIRONMENT", value=env),
client.V1EnvVar(name="LOG_LEVEL", value="DEBUG" if env == "dev" else "INFO"),
client.V1EnvVar(name="METRICS_ENABLED", value="true" if env != "dev" else "false")
]
if env == "production":
base_vars.extend([
client.V1EnvVar(name="ENABLE_PROFILING", value="true"),
client.V1EnvVar(name="ALERT_CHANNEL", value="#prod-alerts")
])
return base_vars
def get_probe_for_env(self, env: str, probe_type: str) -> client.V1Probe:
"""
🏥 Environment-specific health checks
"""
if env == "dev":
# Relaxed probes for development
return client.V1Probe(
http_get=client.V1HTTPGetAction(path="/health", port=8080),
initial_delay_seconds=30,
period_seconds=30,
failure_threshold=5
)
elif env == "production":
# Strict probes for production
return client.V1Probe(
http_get=client.V1HTTPGetAction(path="/health", port=8080),
initial_delay_seconds=10,
period_seconds=5,
timeout_seconds=3,
failure_threshold=2,
success_threshold=2 if probe_type == "readiness" else 1
)
else:
# Standard probes for staging
return client.V1Probe(
http_get=client.V1HTTPGetAction(path="/health", port=8080),
initial_delay_seconds=15,
period_seconds=10,
failure_threshold=3
)
def setup_rbac(self, app_name: str, env: str):
"""
👤 Set up role-based access control
"""
# 🎯 Define roles for each environment
role_rules = {
"dev": [
client.V1PolicyRule(
api_groups=["*"],
resources=["*"],
verbs=["*"] # Developers have full access in dev
)
],
"staging": [
client.V1PolicyRule(
api_groups=["apps", ""],
resources=["deployments", "pods", "services"],
verbs=["get", "list", "watch", "update", "patch"]
)
],
"production": [
client.V1PolicyRule(
api_groups=["apps", ""],
resources=["deployments", "pods", "services"],
verbs=["get", "list", "watch"] # Read-only in prod
)
]
}
role_name = f"{app_name}-{env}-role"
# Create role
role = client.V1Role(
metadata=client.V1ObjectMeta(name=role_name),
rules=role_rules.get(env, role_rules["production"])
)
try:
self.rbac_v1.create_namespaced_role(
namespace=self.environments[env]["namespace"],
body=role
)
print(f"🔐 Created RBAC role for {env}")
except client.exceptions.ApiException:
pass # Role might already exist
def promote_deployment(self, app_name: str, from_env: str, to_env: str):
"""
📈 Promote deployment from one environment to another
"""
print(f"📦 Promoting {app_name} from {from_env} to {to_env}")
# Get source deployment
source_deployment = self.apps_v1.read_namespaced_deployment(
name=f"{app_name}-{from_env}",
namespace=self.environments[from_env]["namespace"]
)
# Extract image
image = source_deployment.spec.template.spec.containers[0].image
# Deploy to target environment
self.deploy_to_environment(app_name, image, to_env)
print(f"✅ Successfully promoted {app_name} to {to_env}!")
def rollback_deployment(self, app_name: str, env: str, revision: int = None):
"""
🔄 Rollback deployment to previous version
"""
deployment_name = f"{app_name}-{env}"
namespace = self.environments[env]["namespace"]
if revision is None:
# Get previous revision
deployment = self.apps_v1.read_namespaced_deployment(deployment_name, namespace)
revision = int(deployment.metadata.annotations.get("deployment.kubernetes.io/revision", "1")) - 1
print(f"🔄 Rolling back {deployment_name} to revision {revision}")
# Perform rollback
body = {
"kind": "DeploymentRollback",
"apiVersion": "apps/v1beta1",
"name": deployment_name,
"rollback": {"revision": revision}
}
self.apps_v1.create_namespaced_deployment_rollback(
name=deployment_name,
namespace=namespace,
body=body
)
print(f"✅ Rollback completed!")
# 🎮 Test the multi-environment system!
deployment_system = MultiEnvironmentDeploymentSystem()
# Deploy to dev first
deployment_system.deploy_to_environment("my-app", "myregistry/my-app:1.0", "dev")
# After testing, promote to staging
deployment_system.promote_deployment("my-app", "dev", "staging")
# Finally promote to production (will wait for maintenance window)
deployment_system.promote_deployment("my-app", "staging", "production")
# Check deployment status
deployment_system.get_deployment_status("my-app", "production")🎓 Key Takeaways
You’ve mastered Kubernetes Deployments! Here’s what you can now do:
- ✅ Create and manage Deployments with Python’s kubernetes client 💪
- ✅ Implement advanced deployment strategies like Blue-Green and Canary 🛡️
- ✅ Set up auto-scaling for handling traffic spikes 🎯
- ✅ Configure health checks to ensure high availability 🐛
- ✅ Build multi-environment deployment systems with RBAC! 🚀
Remember: Kubernetes Deployments are your Swiss Army knife for managing containerized applications. They handle the complexity so you can focus on building great software! 🤝
🤝 Next Steps
Congratulations! 🎉 You’ve conquered Kubernetes Deployments!
Here’s what to do next:
- 💻 Practice with the exercises above using a local Kubernetes cluster (minikube or kind)
- 🏗️ Build a complete CI/CD pipeline that uses these deployment strategies
- 📚 Move on to our next tutorial: Kubernetes Services and Networking
- 🌟 Share your Kubernetes journey with the community!
Remember: Every Kubernetes expert started where you are now. Keep deploying, keep learning, and most importantly, have fun! 🚀
Happy Kubernetes adventures! 🎉🚀✨