Master Kubernetes cluster automation using the kubernetes Python client for container orchestration, deployment management, and cluster operations
Kubernetes has become the de facto standard for container orchestration in modern DevOps environments. Python's kubernetes client library provides powerful automation capabilities for managing clusters, deployments, and workloads at scale.
Why Kubernetes Automation with Python?
Real-World Use Cases:
This lesson demonstrates production-ready patterns used by platform engineering teams at scale.
The kubernetes Python client provides comprehensive access to the Kubernetes API, enabling full cluster automation.
Core Client Components:
1from kubernetes import client, config, watch
2from kubernetes.client.rest import ApiException
3from kubernetes.stream import stream
4import yaml
5
6# Load kubeconfig
7config.load_kube_config() # From ~/.kube/config
8# Or for in-cluster:
9# config.load_incluster_config()
10
11# Create API clients
12v1 = client.CoreV1Api() # Pods, Services, ConfigMaps
13apps_v1 = client.AppsV1Api() # Deployments, StatefulSets
14batch_v1 = client.BatchV1Api() # Jobs, CronJobs
15rbac_v1 = client.RbacAuthorizationV1Api() # RBAC
16autoscaling_v1 = client.AutoscalingV1Api() # HPAAPI Resource Management:
1class PodManager:
2 def __init__(self, namespace='default'):
3 self.v1 = client.CoreV1Api()
4 self.namespace = namespace
5
6 def create_pod(self, name: str, image: str,
7 env_vars: dict = None) -> str:
8 """Create a pod with specified configuration"""
9 pod_manifest = {
10 'apiVersion': 'v1',
11 'kind': 'Pod',
12 'metadata': {'name': name},
13 'spec': {
14 'containers': [{
15 'name': name,
16 'image': image,
17 'env': [{'name': k, 'value': v}
18 for k, v in (env_vars or {}).items()]
19 }],
20 'restartPolicy': 'OnFailure'
21 }
22 }
23
24 response = self.v1.create_namespaced_pod(
25 namespace=self.namespace,
26 body=pod_manifest
27 )
28 return response.metadata.name
29
30 def get_pod_status(self, pod_name: str) -> dict:
31 """Get pod status and conditions"""
32 pod = self.v1.read_namespaced_pod(
33 name=pod_name,
34 namespace=self.namespace
35 )
36
37 return {
38 'phase': pod.status.phase,
39 'conditions': [
40 {'type': c.type, 'status': c.status}
41 for c in (pod.status.conditions or [])
42 ],
43 'container_statuses': [
44 {
45 'name': cs.name,
46 'ready': cs.ready,
47 'restart_count': cs.restart_count
48 }
49 for cs in (pod.status.container_statuses or [])
50 ]
51 }
52
53 def delete_pod(self, pod_name: str, grace_period: int = 30):
54 """Delete pod with grace period"""
55 self.v1.delete_namespaced_pod(
56 name=pod_name,
57 namespace=self.namespace,
58 grace_period_seconds=grace_period
59 )1def create_deployment(name: str, image: str,
2 replicas: int = 3) -> str:
3 """Create deployment with rolling update strategy"""
4 apps_v1 = client.AppsV1Api()
5
6 deployment = client.V1Deployment(
7 metadata=client.V1ObjectMeta(name=name),
8 spec=client.V1DeploymentSpec(
9 replicas=replicas,
10 selector=client.V1LabelSelector(
11 match_labels={'app': name}
12 ),
13 template=client.V1PodTemplateSpec(
14 metadata=client.V1ObjectMeta(
15 labels={'app': name}
16 ),
17 spec=client.V1PodSpec(
18 containers=[
19 client.V1Container(
20 name=name,
21 image=image,
22 ports=[client.V1ContainerPort(
23 container_port=8080
24 )],
25 resources=client.V1ResourceRequirements(
26 requests={'cpu': '100m', 'memory': '128Mi'},
27 limits={'cpu': '500m', 'memory': '512Mi'}
28 ),
29 liveness_probe=client.V1Probe(
30 http_get=client.V1HTTPGetAction(
31 path='/health',
32 port=8080
33 ),
34 initial_delay_seconds=30,
35 period_seconds=10
36 ),
37 readiness_probe=client.V1Probe(
38 http_get=client.V1HTTPGetAction(
39 path='/ready',
40 port=8080
41 ),
42 initial_delay_seconds=5,
43 period_seconds=5
44 )
45 )
46 ]
47 )
48 ),
49 strategy=client.V1DeploymentStrategy(
50 type='RollingUpdate',
51 rolling_update=client.V1RollingUpdateDeployment(
52 max_surge='25%',
53 max_unavailable='25%'
54 )
55 )
56 )
57 )
58
59 response = apps_v1.create_namespaced_deployment(
60 namespace='default',
61 body=deployment
62 )
63 return response.metadata.nameWatch API for Real-Time Monitoring:
1def watch_pod_events(namespace='default', timeout_seconds=60):
2 """Watch pod events in real-time"""
3 v1 = client.CoreV1Api()
4 w = watch.Watch()
5
6 for event in w.stream(
7 v1.list_namespaced_pod,
8 namespace=namespace,
9 timeout_seconds=timeout_seconds
10 ):
11 event_type = event['type'] # ADDED, MODIFIED, DELETED
12 pod = event['object']
13
14 print(f"Event: {event_type}")
15 print(f"Pod: {pod.metadata.name}")
16 print(f"Phase: {pod.status.phase}")
17 print(f"Node: {pod.spec.node_name}")
18 print("-" * 50)Error Handling Best Practices:
1from kubernetes.client.rest import ApiException
2
3def safe_create_resource(resource_func, *args, **kwargs):
4 """Wrapper for safe resource creation"""
5 try:
6 return resource_func(*args, **kwargs)
7 except ApiException as e:
8 if e.status == 409: # Already exists
9 print(f"Resource already exists: {e.reason}")
10 return None
11 elif e.status == 403: # Forbidden
12 print(f"Permission denied: {e.reason}")
13 raise
14 elif e.status == 422: # Invalid
15 print(f"Invalid resource: {e.body}")
16 raise
17 else:
18 print(f"API error ({e.status}): {e.reason}")
19 raiseThis architecture enables robust, production-grade Kubernetes automation with comprehensive error handling and monitoring.
Kubernetes auto-scaling enables applications to adapt dynamically to changing load conditions, optimizing resource utilization and cost.
Horizontal Pod Autoscaler (HPA):
1class AutoScalingManager:
2 def __init__(self, namespace='default'):
3 self.autoscaling_v1 = client.AutoscalingV1Api()
4 self.autoscaling_v2 = client.AutoscalingV2Api()
5 self.namespace = namespace
6
7 def create_hpa(self, deployment_name: str,
8 min_replicas: int = 2,
9 max_replicas: int = 10,
10 target_cpu_percent: int = 70) -> str:
11 """Create HPA for deployment"""
12
13 hpa = client.V1HorizontalPodAutoscaler(
14 metadata=client.V1ObjectMeta(
15 name=f"{deployment_name}-hpa"
16 ),
17 spec=client.V1HorizontalPodAutoscalerSpec(
18 scale_target_ref=client.V1CrossVersionObjectReference(
19 api_version='apps/v1',
20 kind='Deployment',
21 name=deployment_name
22 ),
23 min_replicas=min_replicas,
24 max_replicas=max_replicas,
25 target_cpu_utilization_percentage=target_cpu_percent
26 )
27 )
28
29 response = self.autoscaling_v1.create_namespaced_horizontal_pod_autoscaler(
30 namespace=self.namespace,
31 body=hpa
32 )
33 return response.metadata.name
34
35 def create_advanced_hpa(self, deployment_name: str) -> str:
36 """Create HPA with multiple metrics (v2 API)"""
37
38 hpa = client.V2HorizontalPodAutoscaler(
39 metadata=client.V2ObjectMeta(name=f"{deployment_name}-hpa"),
40 spec=client.V2HorizontalPodAutoscalerSpec(
41 scale_target_ref=client.V2CrossVersionObjectReference(
42 api_version='apps/v1',
43 kind='Deployment',
44 name=deployment_name
45 ),
46 min_replicas=2,
47 max_replicas=20,
48 metrics=[
49 # CPU metric
50 client.V2MetricSpec(
51 type='Resource',
52 resource=client.V2ResourceMetricSource(
53 name='cpu',
54 target=client.V2MetricTarget(
55 type='Utilization',
56 average_utilization=70
57 )
58 )
59 ),
60 # Memory metric
61 client.V2MetricSpec(
62 type='Resource',
63 resource=client.V2ResourceMetricSource(
64 name='memory',
65 target=client.V2MetricTarget(
66 type='Utilization',
67 average_utilization=80
68 )
69 )
70 ),
71 # Custom metric (requests per second)
72 client.V2MetricSpec(
73 type='Pods',
74 pods=client.V2PodsMetricSource(
75 metric=client.V2MetricIdentifier(
76 name='http_requests_per_second'
77 ),
78 target=client.V2MetricTarget(
79 type='AverageValue',
80 average_value='1000'
81 )
82 )
83 )
84 ],
85 behavior=client.V2HorizontalPodAutoscalerBehavior(
86 scale_up=client.V2HPAScalingRules(
87 stabilization_window_seconds=60,
88 policies=[
89 client.V2HPAScalingPolicy(
90 type='Percent',
91 value=50,
92 period_seconds=60
93 )
94 ]
95 ),
96 scale_down=client.V2HPAScalingRules(
97 stabilization_window_seconds=300,
98 policies=[
99 client.V2HPAScalingPolicy(
100 type='Pods',
101 value=1,
102 period_seconds=60
103 )
104 ]
105 )
106 )
107 )
108 )
109
110 response = self.autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(
111 namespace=self.namespace,
112 body=hpa
113 )
114 return response.metadata.name
115
116 def get_hpa_status(self, hpa_name: str) -> dict:
117 """Get current HPA status and metrics"""
118 hpa = self.autoscaling_v1.read_namespaced_horizontal_pod_autoscaler(
119 name=hpa_name,
120 namespace=self.namespace
121 )
122
123 return {
124 'current_replicas': hpa.status.current_replicas,
125 'desired_replicas': hpa.status.desired_replicas,
126 'current_cpu_utilization': hpa.status.current_cpu_utilization_percentage,
127 'target_cpu_utilization': hpa.spec.target_cpu_utilization_percentage
128 }Resource Quota Management:
1def create_resource_quota(namespace: str) -> str:
2 """Create resource quota for namespace"""
3 v1 = client.CoreV1Api()
4
5 quota = client.V1ResourceQuota(
6 metadata=client.V1ObjectMeta(name='compute-quota'),
7 spec=client.V1ResourceQuotaSpec(
8 hard={
9 'requests.cpu': '10',
10 'requests.memory': '20Gi',
11 'limits.cpu': '20',
12 'limits.memory': '40Gi',
13 'persistentvolumeclaims': '10',
14 'pods': '50',
15 'services': '20',
16 'services.loadbalancers': '5'
17 }
18 )
19 )
20
21 response = v1.create_namespaced_resource_quota(
22 namespace=namespace,
23 body=quota
24 )
25 return response.metadata.name
26
27def get_namespace_resource_usage(namespace: str) -> dict:
28 """Get current resource usage in namespace"""
29 v1 = client.CoreV1Api()
30
31 pods = v1.list_namespaced_pod(namespace=namespace)
32
33 total_cpu_requests = 0
34 total_memory_requests = 0
35 total_cpu_limits = 0
36 total_memory_limits = 0
37
38 for pod in pods.items:
39 for container in pod.spec.containers:
40 if container.resources.requests:
41 cpu = container.resources.requests.get('cpu', '0')
42 mem = container.resources.requests.get('memory', '0')
43 total_cpu_requests += parse_cpu(cpu)
44 total_memory_requests += parse_memory(mem)
45
46 if container.resources.limits:
47 cpu = container.resources.limits.get('cpu', '0')
48 mem = container.resources.limits.get('memory', '0')
49 total_cpu_limits += parse_cpu(cpu)
50 total_memory_limits += parse_memory(mem)
51
52 return {
53 'pod_count': len(pods.items),
54 'cpu_requests': f"{total_cpu_requests}m",
55 'memory_requests': f"{total_memory_requests}Mi",
56 'cpu_limits': f"{total_cpu_limits}m",
57 'memory_limits': f"{total_memory_limits}Mi"
58 }Vertical Pod Autoscaler (VPA) Integration:
1def create_vpa_recommendation(deployment_name: str) -> dict:
2 """Create VPA for resource recommendation"""
3 # VPA requires custom API group
4 from kubernetes import client
5
6 vpa_spec = {
7 'apiVersion': 'autoscaling.k8s.io/v1',
8 'kind': 'VerticalPodAutoscaler',
9 'metadata': {'name': f"{deployment_name}-vpa"},
10 'spec': {
11 'targetRef': {
12 'apiVersion': 'apps/v1',
13 'kind': 'Deployment',
14 'name': deployment_name
15 },
16 'updatePolicy': {
17 'updateMode': 'Off' # Recommendation only
18 },
19 'resourcePolicy': {
20 'containerPolicies': [{
21 'containerName': '*',
22 'minAllowed': {
23 'cpu': '100m',
24 'memory': '128Mi'
25 },
26 'maxAllowed': {
27 'cpu': '2',
28 'memory': '2Gi'
29 }
30 }]
31 }
32 }
33 }
34
35 # Create using custom objects API
36 api = client.CustomObjectsApi()
37 response = api.create_namespaced_custom_object(
38 group='autoscaling.k8s.io',
39 version='v1',
40 namespace='default',
41 plural='verticalpodautoscalers',
42 body=vpa_spec
43 )
44 return responseThis comprehensive auto-scaling strategy ensures optimal resource utilization while maintaining application performance.
Modern Kubernetes deployments require sophisticated strategies for zero-downtime updates and custom automation logic through operators.
Blue-Green Deployment Implementation:
1class BlueGreenDeployment:
2 def __init__(self, namespace='default'):
3 self.apps_v1 = client.AppsV1Api()
4 self.v1 = client.CoreV1Api()
5 self.namespace = namespace
6
7 def deploy_new_version(self, app_name: str,
8 new_image: str,
9 current_color: str = 'blue') -> str:
10 """Deploy new version alongside current"""
11 new_color = 'green' if current_color == 'blue' else 'blue'
12 deployment_name = f"{app_name}-{new_color}"
13
14 # Create new deployment
15 deployment = self._create_deployment_spec(
16 name=deployment_name,
17 image=new_image,
18 color=new_color,
19 app_label=app_name
20 )
21
22 self.apps_v1.create_namespaced_deployment(
23 namespace=self.namespace,
24 body=deployment
25 )
26
27 # Wait for deployment to be ready
28 self._wait_for_deployment(deployment_name)
29
30 return new_color
31
32 def switch_traffic(self, app_name: str, target_color: str):
33 """Switch service to point to new deployment"""
34 service_name = f"{app_name}-service"
35
36 # Update service selector
37 service = self.v1.read_namespaced_service(
38 name=service_name,
39 namespace=self.namespace
40 )
41
42 service.spec.selector = {
43 'app': app_name,
44 'version': target_color
45 }
46
47 self.v1.patch_namespaced_service(
48 name=service_name,
49 namespace=self.namespace,
50 body=service
51 )
52
53 def cleanup_old_version(self, app_name: str, old_color: str):
54 """Remove old deployment after successful switch"""
55 deployment_name = f"{app_name}-{old_color}"
56
57 self.apps_v1.delete_namespaced_deployment(
58 name=deployment_name,
59 namespace=self.namespace
60 )
61
62 def rollback(self, app_name: str, previous_color: str):
63 """Rollback to previous version"""
64 self.switch_traffic(app_name, previous_color)
65
66 def _create_deployment_spec(self, name: str, image: str,
67 color: str, app_label: str):
68 """Create deployment specification"""
69 return client.V1Deployment(
70 metadata=client.V1ObjectMeta(name=name),
71 spec=client.V1DeploymentSpec(
72 replicas=3,
73 selector=client.V1LabelSelector(
74 match_labels={
75 'app': app_label,
76 'version': color
77 }
78 ),
79 template=client.V1PodTemplateSpec(
80 metadata=client.V1ObjectMeta(
81 labels={
82 'app': app_label,
83 'version': color
84 }
85 ),
86 spec=client.V1PodSpec(
87 containers=[
88 client.V1Container(
89 name=app_label,
90 image=image,
91 ports=[client.V1ContainerPort(
92 container_port=8080
93 )]
94 )
95 ]
96 )
97 )
98 )
99 )
100
101 def _wait_for_deployment(self, deployment_name: str,
102 timeout: int = 300):
103 """Wait for deployment to be ready"""
104 import time
105 start_time = time.time()
106
107 while time.time() - start_time < timeout:
108 deployment = self.apps_v1.read_namespaced_deployment(
109 name=deployment_name,
110 namespace=self.namespace
111 )
112
113 if (deployment.status.ready_replicas ==
114 deployment.spec.replicas):
115 return True
116
117 time.sleep(5)
118
119 raise TimeoutError(f"Deployment {deployment_name} not ready")Canary Deployment Strategy:
1class CanaryDeployment:
2 def __init__(self, namespace='default'):
3 self.apps_v1 = client.AppsV1Api()
4 self.namespace = namespace
5
6 def create_canary(self, app_name: str, new_image: str,
7 canary_percent: int = 10) -> str:
8 """Create canary deployment with specified traffic %"""
9
10 # Get current deployment replica count
11 stable = self.apps_v1.read_namespaced_deployment(
12 name=f"{app_name}-stable",
13 namespace=self.namespace
14 )
15 total_replicas = stable.spec.replicas
16
17 # Calculate canary replicas
18 canary_replicas = max(1, int(total_replicas * canary_percent / 100))
19 stable_replicas = total_replicas - canary_replicas
20
21 # Scale down stable
22 stable.spec.replicas = stable_replicas
23 self.apps_v1.patch_namespaced_deployment(
24 name=f"{app_name}-stable",
25 namespace=self.namespace,
26 body=stable
27 )
28
29 # Create canary deployment
30 canary = self._create_canary_spec(
31 app_name=app_name,
32 image=new_image,
33 replicas=canary_replicas
34 )
35
36 self.apps_v1.create_namespaced_deployment(
37 namespace=self.namespace,
38 body=canary
39 )
40
41 return f"{app_name}-canary"
42
43 def promote_canary(self, app_name: str):
44 """Promote canary to stable"""
45 canary = self.apps_v1.read_namespaced_deployment(
46 name=f"{app_name}-canary",
47 namespace=self.namespace
48 )
49
50 # Update stable deployment with canary image
51 stable = self.apps_v1.read_namespaced_deployment(
52 name=f"{app_name}-stable",
53 namespace=self.namespace
54 )
55
56 stable.spec.template.spec.containers[0].image = \
57 canary.spec.template.spec.containers[0].image
58
59 self.apps_v1.patch_namespaced_deployment(
60 name=f"{app_name}-stable",
61 namespace=self.namespace,
62 body=stable
63 )
64
65 # Delete canary
66 self.apps_v1.delete_namespaced_deployment(
67 name=f"{app_name}-canary",
68 namespace=self.namespace
69 )
70
71 def rollback_canary(self, app_name: str, original_replicas: int):
72 """Rollback canary deployment"""
73 # Delete canary
74 self.apps_v1.delete_namespaced_deployment(
75 name=f"{app_name}-canary",
76 namespace=self.namespace
77 )
78
79 # Restore stable replicas
80 stable = self.apps_v1.read_namespaced_deployment(
81 name=f"{app_name}-stable",
82 namespace=self.namespace
83 )
84 stable.spec.replicas = original_replicas
85
86 self.apps_v1.patch_namespaced_deployment(
87 name=f"{app_name}-stable",
88 namespace=self.namespace,
89 body=stable
90 )Custom Operator Pattern:
1class DatabaseOperator:
2 """Custom operator for database lifecycle management"""
3
4 def __init__(self, namespace='default'):
5 self.v1 = client.CoreV1Api()
6 self.apps_v1 = client.AppsV1Api()
7 self.custom_api = client.CustomObjectsApi()
8 self.namespace = namespace
9
10 def reconcile_loop(self):
11 """Main reconciliation loop"""
12 w = watch.Watch()
13
14 # Watch custom resources
15 for event in w.stream(
16 self.custom_api.list_namespaced_custom_object,
17 group='databases.example.com',
18 version='v1',
19 namespace=self.namespace,
20 plural='databases'
21 ):
22 event_type = event['type']
23 db_resource = event['object']
24
25 if event_type == 'ADDED':
26 self._handle_create(db_resource)
27 elif event_type == 'MODIFIED':
28 self._handle_update(db_resource)
29 elif event_type == 'DELETED':
30 self._handle_delete(db_resource)
31
32 def _handle_create(self, db_resource: dict):
33 """Handle database creation"""
34 name = db_resource['metadata']['name']
35 spec = db_resource['spec']
36
37 # Create StatefulSet for database
38 statefulset = self._create_database_statefulset(
39 name=name,
40 image=spec['image'],
41 replicas=spec.get('replicas', 1),
42 storage=spec.get('storage', '10Gi')
43 )
44
45 self.apps_v1.create_namespaced_stateful_set(
46 namespace=self.namespace,
47 body=statefulset
48 )
49
50 # Create Service
51 service = self._create_database_service(name)
52 self.v1.create_namespaced_service(
53 namespace=self.namespace,
54 body=service
55 )
56
57 # Update status
58 self._update_status(name, 'Provisioning')
59
60 def _create_database_statefulset(self, name: str, image: str,
61 replicas: int, storage: str):
62 """Create StatefulSet specification"""
63 return client.V1StatefulSet(
64 metadata=client.V1ObjectMeta(name=name),
65 spec=client.V1StatefulSetSpec(
66 service_name=name,
67 replicas=replicas,
68 selector=client.V1LabelSelector(
69 match_labels={'app': name}
70 ),
71 template=client.V1PodTemplateSpec(
72 metadata=client.V1ObjectMeta(
73 labels={'app': name}
74 ),
75 spec=client.V1PodSpec(
76 containers=[
77 client.V1Container(
78 name='database',
79 image=image,
80 ports=[client.V1ContainerPort(
81 container_port=5432
82 )],
83 volume_mounts=[
84 client.V1VolumeMount(
85 name='data',
86 mount_path='/var/lib/postgresql/data'
87 )
88 ]
89 )
90 ]
91 )
92 ),
93 volume_claim_templates=[
94 client.V1PersistentVolumeClaim(
95 metadata=client.V1ObjectMeta(name='data'),
96 spec=client.V1PersistentVolumeClaimSpec(
97 access_modes=['ReadWriteOnce'],
98 resources=client.V1ResourceRequirements(
99 requests={'storage': storage}
100 )
101 )
102 )
103 ]
104 )
105 )These advanced patterns enable sophisticated deployment strategies and custom automation for complex Kubernetes workflows.
Build a comprehensive Kubernetes automation platform that manages deployments, auto-scaling, monitoring, and self-healing across multiple namespaces with advanced deployment strategies.
1#!/usr/bin/env python3
2"""
3☸️ Production Kubernetes Automation Platform
4Topic: Complete K8s cluster automation with deployment management
5Author: DevOps Automation Team
6Dependencies: pip install kubernetes pyyaml
7"""
8
9from kubernetes import client, config, watch
10from kubernetes.client.rest import ApiException
11from kubernetes.stream import stream
12import yaml
13import time
14import logging
15import sys
16from datetime import datetime
17from typing import Dict, List, Optional, Any
18from dataclasses import dataclass, field
19from enum import Enum
20
21# Configure logging
22logging.basicConfig(
23 level=logging.INFO,
24 format='%(asctime)s - %(levelname)s - %(message)s',
25 handlers=[
26 logging.FileHandler('kubernetes-automation.log'),
27 logging.StreamHandler(sys.stdout)
28 ]
29)
30logger = logging.getLogger(__name__)
31
32
33class DeploymentStrategy(Enum):
34 """Deployment strategy types"""
35 ROLLING_UPDATE = "RollingUpdate"
36 RECREATE = "Recreate"
37 BLUE_GREEN = "BlueGreen"
38 CANARY = "Canary"
39
40
41@dataclass
42class DeploymentConfig:
43 """Deployment configuration"""
44 name: str
45 image: str
46 replicas: int = 3
47 namespace: str = 'default'
48 strategy: DeploymentStrategy = DeploymentStrategy.ROLLING_UPDATE
49 cpu_request: str = '100m'
50 cpu_limit: str = '500m'
51 memory_request: str = '128Mi'
52 memory_limit: str = '512Mi'
53 env_vars: Dict[str, str] = field(default_factory=dict)
54 port: int = 8080
55 health_check_path: str = '/health'
56
57
58class KubernetesClusterManager:
59 """
60 Comprehensive Kubernetes cluster management
61 """
62
63 def __init__(self, kubeconfig_path: Optional[str] = None):
64 """
65 Initialize Kubernetes client
66
67 Args:
68 kubeconfig_path: Path to kubeconfig file (None for in-cluster)
69 """
70 try:
71 if kubeconfig_path:
72 config.load_kube_config(config_file=kubeconfig_path)
73 else:
74 # Try in-cluster first, fallback to default kubeconfig
75 try:
76 config.load_incluster_config()
77 logger.info("✓ Loaded in-cluster configuration")
78 except config.ConfigException:
79 config.load_kube_config()
80 logger.info("✓ Loaded kubeconfig from default location")
81
82 self.v1 = client.CoreV1Api()
83 self.apps_v1 = client.AppsV1Api()
84 self.batch_v1 = client.BatchV1Api()
85 self.autoscaling_v1 = client.AutoscalingV1Api()
86 self.rbac_v1 = client.RbacAuthorizationV1Api()
87
88 logger.info("🚀 Kubernetes cluster manager initialized")
89
90 except Exception as e:
91 logger.error(f"Failed to initialize Kubernetes client: {e}")
92 raise
93
94 def get_cluster_info(self) -> Dict[str, Any]:
95 """Get cluster information"""
96 try:
97 version = client.VersionApi().get_code()
98 nodes = self.v1.list_node()
99 namespaces = self.v1.list_namespace()
100
101 cluster_info = {
102 'version': f"{version.major}.{version.minor}",
103 'git_version': version.git_version,
104 'node_count': len(nodes.items),
105 'namespace_count': len(namespaces.items),
106 'nodes': []
107 }
108
109 for node in nodes.items:
110 node_info = {
111 'name': node.metadata.name,
112 'status': self._get_node_status(node),
113 'capacity': {
114 'cpu': node.status.capacity.get('cpu'),
115 'memory': node.status.capacity.get('memory'),
116 'pods': node.status.capacity.get('pods')
117 },
118 'allocatable': {
119 'cpu': node.status.allocatable.get('cpu'),
120 'memory': node.status.allocatable.get('memory'),
121 'pods': node.status.allocatable.get('pods')
122 },
123 'kubelet_version': node.status.node_info.kubelet_version
124 }
125 cluster_info['nodes'].append(node_info)
126
127 return cluster_info
128
129 except ApiException as e:
130 logger.error(f"Failed to get cluster info: {e}")
131 raise
132
133 def _get_node_status(self, node) -> str:
134 """Get node status"""
135 for condition in node.status.conditions:
136 if condition.type == 'Ready':
137 return 'Ready' if condition.status == 'True' else 'NotReady'
138 return 'Unknown'
139
140 def create_namespace(self, name: str, labels: Dict[str, str] = None) -> str:
141 """Create namespace with labels"""
142 try:
143 namespace = client.V1Namespace(
144 metadata=client.V1ObjectMeta(
145 name=name,
146 labels=labels or {}
147 )
148 )
149
150 self.v1.create_namespace(body=namespace)
151 logger.info(f"✓ Created namespace: {name}")
152 return name
153
154 except ApiException as e:
155 if e.status == 409:
156 logger.info(f"Namespace {name} already exists")
157 return name
158 logger.error(f"Failed to create namespace: {e}")
159 raise
160
161 def deploy_application(self, config: DeploymentConfig) -> Dict[str, Any]:
162 """
163 Deploy application with specified configuration
164
165 Args:
166 config: Deployment configuration
167
168 Returns:
169 Deployment information
170 """
171 try:
172 logger.info(f"📦 Deploying {config.name} to {config.namespace}")
173
174 # Create deployment
175 deployment = self._create_deployment_spec(config)
176
177 try:
178 response = self.apps_v1.create_namespaced_deployment(
179 namespace=config.namespace,
180 body=deployment
181 )
182 logger.info(f"✓ Created deployment: {config.name}")
183 except ApiException as e:
184 if e.status == 409:
185 logger.info(f"Deployment exists, updating...")
186 response = self.apps_v1.patch_namespaced_deployment(
187 name=config.name,
188 namespace=config.namespace,
189 body=deployment
190 )
191
192 # Create service
193 service = self._create_service_spec(config)
194
195 try:
196 svc_response = self.v1.create_namespaced_service(
197 namespace=config.namespace,
198 body=service
199 )
200 logger.info(f"✓ Created service: {config.name}-service")
201 except ApiException as e:
202 if e.status == 409:
203 logger.info(f"Service exists")
204
205 # Wait for deployment to be ready
206 self._wait_for_deployment(config.name, config.namespace)
207
208 return {
209 'deployment': config.name,
210 'namespace': config.namespace,
211 'replicas': config.replicas,
212 'image': config.image,
213 'status': 'deployed'
214 }
215
216 except Exception as e:
217 logger.error(f"Deployment failed: {e}")
218 raise
219
220 def _create_deployment_spec(self, config: DeploymentConfig):
221 """Create deployment specification"""
222 return client.V1Deployment(
223 metadata=client.V1ObjectMeta(
224 name=config.name,
225 labels={'app': config.name, 'managed-by': 'automation'}
226 ),
227 spec=client.V1DeploymentSpec(
228 replicas=config.replicas,
229 selector=client.V1LabelSelector(
230 match_labels={'app': config.name}
231 ),
232 template=client.V1PodTemplateSpec(
233 metadata=client.V1ObjectMeta(
234 labels={'app': config.name}
235 ),
236 spec=client.V1PodSpec(
237 containers=[
238 client.V1Container(
239 name=config.name,
240 image=config.image,
241 ports=[client.V1ContainerPort(
242 container_port=config.port
243 )],
244 env=[
245 client.V1EnvVar(name=k, value=v)
246 for k, v in config.env_vars.items()
247 ],
248 resources=client.V1ResourceRequirements(
249 requests={
250 'cpu': config.cpu_request,
251 'memory': config.memory_request
252 },
253 limits={
254 'cpu': config.cpu_limit,
255 'memory': config.memory_limit
256 }
257 ),
258 liveness_probe=client.V1Probe(
259 http_get=client.V1HTTPGetAction(
260 path=config.health_check_path,
261 port=config.port
262 ),
263 initial_delay_seconds=30,
264 period_seconds=10,
265 timeout_seconds=5,
266 failure_threshold=3
267 ),
268 readiness_probe=client.V1Probe(
269 http_get=client.V1HTTPGetAction(
270 path=config.health_check_path,
271 port=config.port
272 ),
273 initial_delay_seconds=5,
274 period_seconds=5,
275 timeout_seconds=3,
276 failure_threshold=3
277 )
278 )
279 ],
280 restart_policy='Always'
281 )
282 ),
283 strategy=client.V1DeploymentStrategy(
284 type=config.strategy.value if config.strategy != DeploymentStrategy.BLUE_GREEN else 'RollingUpdate',
285 rolling_update=client.V1RollingUpdateDeployment(
286 max_surge='25%',
287 max_unavailable='25%'
288 ) if config.strategy == DeploymentStrategy.ROLLING_UPDATE else None
289 )
290 )
291 )
292
293 def _create_service_spec(self, config: DeploymentConfig):
294 """Create service specification"""
295 return client.V1Service(
296 metadata=client.V1ObjectMeta(
297 name=f"{config.name}-service",
298 labels={'app': config.name}
299 ),
300 spec=client.V1ServiceSpec(
301 selector={'app': config.name},
302 ports=[
303 client.V1ServicePort(
304 port=80,
305 target_port=config.port,
306 protocol='TCP'
307 )
308 ],
309 type='ClusterIP'
310 )
311 )
312
313 def _wait_for_deployment(self, name: str, namespace: str,
314 timeout: int = 300):
315 """Wait for deployment to be ready"""
316 logger.info(f"⏳ Waiting for deployment {name} to be ready...")
317 start_time = time.time()
318
319 while time.time() - start_time < timeout:
320 try:
321 deployment = self.apps_v1.read_namespaced_deployment(
322 name=name,
323 namespace=namespace
324 )
325
326 if deployment.status.ready_replicas == deployment.spec.replicas:
327 logger.info(f"✓ Deployment {name} is ready!")
328 return True
329
330 time.sleep(5)
331
332 except ApiException as e:
333 logger.error(f"Error checking deployment status: {e}")
334 time.sleep(5)
335
336 raise TimeoutError(f"Deployment {name} not ready after {timeout}s")
337
338 def scale_deployment(self, name: str, namespace: str,
339 replicas: int) -> Dict[str, Any]:
340 """Scale deployment to specified replica count"""
341 try:
342 deployment = self.apps_v1.read_namespaced_deployment(
343 name=name,
344 namespace=namespace
345 )
346
347 old_replicas = deployment.spec.replicas
348 deployment.spec.replicas = replicas
349
350 self.apps_v1.patch_namespaced_deployment(
351 name=name,
352 namespace=namespace,
353 body=deployment
354 )
355
356 logger.info(f"✓ Scaled {name} from {old_replicas} to {replicas} replicas")
357
358 return {
359 'deployment': name,
360 'old_replicas': old_replicas,
361 'new_replicas': replicas
362 }
363
364 except ApiException as e:
365 logger.error(f"Failed to scale deployment: {e}")
366 raise
367
368 def create_hpa(self, deployment_name: str, namespace: str,
369 min_replicas: int = 2, max_replicas: int = 10,
370 target_cpu_percent: int = 70) -> str:
371 """Create Horizontal Pod Autoscaler"""
372 try:
373 hpa = client.V1HorizontalPodAutoscaler(
374 metadata=client.V1ObjectMeta(
375 name=f"{deployment_name}-hpa"
376 ),
377 spec=client.V1HorizontalPodAutoscalerSpec(
378 scale_target_ref=client.V1CrossVersionObjectReference(
379 api_version='apps/v1',
380 kind='Deployment',
381 name=deployment_name
382 ),
383 min_replicas=min_replicas,
384 max_replicas=max_replicas,
385 target_cpu_utilization_percentage=target_cpu_percent
386 )
387 )
388
389 self.autoscaling_v1.create_namespaced_horizontal_pod_autoscaler(
390 namespace=namespace,
391 body=hpa
392 )
393
394 logger.info(f"✓ Created HPA for {deployment_name}")
395 return f"{deployment_name}-hpa"
396
397 except ApiException as e:
398 if e.status == 409:
399 logger.info(f"HPA already exists")
400 return f"{deployment_name}-hpa"
401 logger.error(f"Failed to create HPA: {e}")
402 raise
403
404 def rollout_restart(self, deployment_name: str, namespace: str):
405 """Restart deployment pods"""
406 try:
407 deployment = self.apps_v1.read_namespaced_deployment(
408 name=deployment_name,
409 namespace=namespace
410 )
411
412 # Update annotation to trigger rollout
413 if not deployment.spec.template.metadata.annotations:
414 deployment.spec.template.metadata.annotations = {}
415
416 deployment.spec.template.metadata.annotations['kubectl.kubernetes.io/restartedAt'] = \
417 datetime.now().isoformat()
418
419 self.apps_v1.patch_namespaced_deployment(
420 name=deployment_name,
421 namespace=namespace,
422 body=deployment
423 )
424
425 logger.info(f"✓ Rollout restart initiated for {deployment_name}")
426
427 except ApiException as e:
428 logger.error(f"Failed to restart deployment: {e}")
429 raise
430
431 def get_pod_logs(self, pod_name: str, namespace: str,
432 tail_lines: int = 100) -> str:
433 """Get pod logs"""
434 try:
435 logs = self.v1.read_namespaced_pod_log(
436 name=pod_name,
437 namespace=namespace,
438 tail_lines=tail_lines
439 )
440 return logs
441
442 except ApiException as e:
443 logger.error(f"Failed to get pod logs: {e}")
444 raise
445
446 def execute_pod_command(self, pod_name: str, namespace: str,
447 command: List[str]) -> str:
448 """Execute command in pod"""
449 try:
450 resp = stream(
451 self.v1.connect_get_namespaced_pod_exec,
452 pod_name,
453 namespace,
454 command=command,
455 stderr=True,
456 stdin=False,
457 stdout=True,
458 tty=False
459 )
460 return resp
461
462 except ApiException as e:
463 logger.error(f"Failed to execute command: {e}")
464 raise
465
466 def watch_pod_events(self, namespace: str = 'default',
467 timeout_seconds: int = 60):
468 """Watch pod events in real-time"""
469 logger.info(f"👀 Watching pod events in {namespace}")
470
471 w = watch.Watch()
472 for event in w.stream(
473 self.v1.list_namespaced_pod,
474 namespace=namespace,
475 timeout_seconds=timeout_seconds
476 ):
477 event_type = event['type']
478 pod = event['object']
479
480 logger.info(f"Event: {event_type} | Pod: {pod.metadata.name} | "
481 f"Phase: {pod.status.phase}")
482
483 def cleanup_namespace(self, namespace: str):
484 """Delete all resources in namespace"""
485 logger.info(f"🧹 Cleaning up namespace: {namespace}")
486
487 try:
488 # Delete all deployments
489 deployments = self.apps_v1.list_namespaced_deployment(
490 namespace=namespace
491 )
492 for deployment in deployments.items:
493 self.apps_v1.delete_namespaced_deployment(
494 name=deployment.metadata.name,
495 namespace=namespace
496 )
497 logger.info(f" Deleted deployment: {deployment.metadata.name}")
498
499 # Delete all services
500 services = self.v1.list_namespaced_service(namespace=namespace)
501 for service in services.items:
502 self.v1.delete_namespaced_service(
503 name=service.metadata.name,
504 namespace=namespace
505 )
506 logger.info(f" Deleted service: {service.metadata.name}")
507
508 logger.info(f"✓ Namespace {namespace} cleaned up")
509
510 except ApiException as e:
511 logger.error(f"Cleanup failed: {e}")
512 raise
513
514
515def main():
516 """
517 Main entry point - Kubernetes automation demonstration
518 """
519 print("\n" + "="*70)
520 print("☸️ Production Kubernetes Automation Platform")
521 print("="*70 + "\n")
522
523 try:
524 # Initialize cluster manager
525 k8s = KubernetesClusterManager()
526
527 # Get cluster information
528 logger.info("📊 Retrieving cluster information...")
529 cluster_info = k8s.get_cluster_info()
530
531 print(f"\n{'='*70}")
532 print("CLUSTER INFORMATION")
533 print('='*70)
534 print(f"Kubernetes Version: {cluster_info['version']}")
535 print(f"Nodes: {cluster_info['node_count']}")
536 print(f"Namespaces: {cluster_info['namespace_count']}")
537 print("\nNode Details:")
538 for node in cluster_info['nodes']:
539 print(f" - {node['name']}: {node['status']} | "
540 f"CPU: {node['capacity']['cpu']} | "
541 f"Memory: {node['capacity']['memory']}")
542 print('='*70)
543
544 # Create test namespace
545 namespace = 'automation-demo'
546 k8s.create_namespace(namespace, labels={'env': 'demo'})
547
548 # Deploy sample application
549 app_config = DeploymentConfig(
550 name='nginx-demo',
551 image='nginx:latest',
552 replicas=3,
553 namespace=namespace,
554 cpu_request='50m',
555 cpu_limit='100m',
556 memory_request='64Mi',
557 memory_limit='128Mi',
558 port=80,
559 health_check_path='/',
560 env_vars={'ENV': 'demo'}
561 )
562
563 deployment_info = k8s.deploy_application(app_config)
564
565 print(f"\n{'='*70}")
566 print("DEPLOYMENT RESULTS")
567 print('='*70)
568 print(f"Application: {deployment_info['deployment']}")
569 print(f"Namespace: {deployment_info['namespace']}")
570 print(f"Replicas: {deployment_info['replicas']}")
571 print(f"Image: {deployment_info['image']}")
572 print(f"Status: {deployment_info['status']}")
573 print('='*70)
574
575 # Create HPA
576 k8s.create_hpa(
577 deployment_name='nginx-demo',
578 namespace=namespace,
579 min_replicas=2,
580 max_replicas=5,
581 target_cpu_percent=60
582 )
583
584 # Scale deployment
585 logger.info("\n⚙️ Testing manual scaling...")
586 scale_result = k8s.scale_deployment('nginx-demo', namespace, 5)
587 print(f"\nScaled from {scale_result['old_replicas']} to "
588 f"{scale_result['new_replicas']} replicas")
589
590 # Wait a bit
591 time.sleep(10)
592
593 # Cleanup
594 logger.info("\n🧹 Cleaning up demo resources...")
595 k8s.cleanup_namespace(namespace)
596
597 logger.info("\n🎉 Kubernetes automation demonstration completed!")
598
599 except KeyboardInterrupt:
600 logger.info("\n⚠️ Interrupted by user")
601 sys.exit(130)
602 except Exception as e:
603 logger.error(f"\n❌ Fatal error: {e}")
604 import traceback
605 traceback.print_exc()
606 sys.exit(1)
607
608
609if __name__ == "__main__":
610 main()🔍 Comprehensive Implementation Breakdown:
1. Kubernetes Client Architecture:
2. Cluster Management Features:
1cluster_info = k8s.get_cluster_info()
2# Returns: version, node count, namespaces, node details with capacity3. Application Deployment:
4. Auto-Scaling Implementation:
1k8s.create_hpa(
2 deployment_name='nginx-demo',
3 namespace='automation-demo',
4 min_replicas=2,
5 max_replicas=10,
6 target_cpu_percent=70
7)5. Operational Capabilities:
scale_deployment() for immediate scalingrollout_restart() to restart pods without downtimeget_pod_logs() for debuggingexecute_pod_command() for pod interaction6. Production Best Practices:
✅ Idempotent Operations: Safe to run multiple times (checks for existing resources)
✅ Graceful Error Handling: Specific handling for 409 (exists), 403 (forbidden), etc.
✅ Wait Mechanisms: _wait_for_deployment() ensures deployment readiness
✅ Resource Cleanup: cleanup_namespace() for proper teardown
✅ Comprehensive Logging: Detailed logging for operations and errors
✅ Type Safety: Type hints throughout for IDE support and validation
7. Real-World Patterns:
8. Security Considerations:
This implementation provides a solid foundation for building enterprise-grade Kubernetes automation platforms used by DevOps and SRE teams to manage containerized workloads at scale!
Your progress is saved in this browser, on this device only.