kubernetes

# Kubernetes编排技能 ## 概述 Kubernetes是容器编排的事实标准。不当的K8s配置会导致资源浪费、性能问题和运维困难。在设计Kubernetes架构前需要仔细分析应用需求。 **核心原则**: 好的Kubernetes编排应该提升应用可用性和可扩展性，同时保证资源利用率。坏的编排会增加运维复杂性，甚至影响服务稳定性。 ## 何时使用 **始终:** - 设计微服务部署架构时 - 实现容器编排和调度时 - 优化Kubernetes集群性能时 - 解决服务发现和负载均衡问题时 - 建立自动化运维流程时 **触发短语:** - "Kubernetes部署" - "K8s集群优化" - "Pod调度策略" - "服务网格配置" - "自动扩缩容" - "Kubernetes安全" ## Kubernetes编排功能 ### 集群架构设计 - Master节点配置 - Worker节点规划 - 网络插件选择 - 存储方案设计 - 高可用架构 ### 资源管理 - Pod配置优化 - Deployment策略 - Service和Ingress配置 - ConfigMap和Secret管理 - 资源配额设置 ### 调度和扩缩容 - 调度策略配置 - 自动扩缩容设置 - 节点亲和性规则 - 污点和容忍度 - 优先级和抢占 ### 监控和运维 - 健康检查配置 - 日志收集方案 - 监控指标设置 - 告警规则配置 - 故障排查 ## 常见Kubernetes问题 ### 资源配置不当 ``` 问题: Pod资源配置不合理导致性能问题 错误示例: - 没有设置资源请求和限制 - CPU和内存配置过高或过低 - 忽略资源使用监控 - 不合理的QoS类别 解决方案: 1. 设置合理的requests和limits 2. 实施资源监控和告警 3. 优化Pod调度策略 4. 配置合适的QoS类别 ``` ### 网络通信问题 ``` 问题: Kubernetes网络配置导致通信故障 错误示例: - Service无法访问Pod - Ingress配置错误 - 网络策略阻止通信 - DNS解析失败 解决方案: 1. 检查Service和Endpoint配置 2. 验证Ingress规则设置 3. 调整NetworkPolicy策略 4. 排查CoreDNS问题 ``` ### 存储挂载问题 ``` 问题: 持久化存储挂载失败或数据丢失 错误示例: - PV和PVC配置不匹配 - 存储类选择错误 - 权限配置问题 - 数据备份策略缺失 解决方案: 1. 正确配置PV和PVC 2. 选择合适的StorageClass 3. 设置正确的访问权限 4. 实施定期备份策略 ``` ## 代码实现示例 ### Kubernetes集群分析器 ```python import kubernetes import yaml import json import time from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass from collections import defaultdict @dataclass class ClusterMetrics: """集群指标""" node_count: int pod_count: int service_count: int deployment_count: int namespace_count: int cpu_capacity: float memory_capacity: float cpu_usage: float memory_usage: float @dataclass class PodMetrics: """Pod指标""" name: str namespace: str status: str node_name: str cpu_requests: float cpu_limits: float memory_requests: float memory_limits: float restart_count: int age_seconds: int @dataclass class ClusterIssue: """集群问题""" severity: str # critical, high, medium, low type: str resource: str namespace: str message: str suggestion: str class KubernetesClusterAnalyzer: def __init__(self, kubeconfig_path: Optional[str] = None): self.kubeconfig_path = kubeconfig_path self.load_kubernetes_config() self.v1 = kubernetes.client.CoreV1Api() self.apps_v1 = kubernetes.client.AppsV1Api() self.networking_v1 = kubernetes.client.NetworkingV1Api() self.issues: List[ClusterIssue] = [] def load_kubernetes_config(self): """加载Kubernetes配置""" try: if self.kubeconfig_path: kubernetes.config.load_kube_config(config_file=self.kubeconfig_path) else: kubernetes.config.load_incluster_config() except Exception as e: try: kubernetes.config.load_kube_config() except Exception as e2: raise Exception(f'无法加载Kubernetes配置: {e2}') def analyze_cluster(self) -> Dict[str, Any]: """分析整个集群""" try: # 获取集群指标 cluster_metrics = self.get_cluster_metrics() # 分析Pod pod_analysis = self.analyze_pods() # 分析节点 node_analysis = self.analyze_nodes() # 分析资源使用 resource_analysis = self.analyze_resource_usage() # 分析网络 network_analysis = self.analyze_network() # 生成报告 report = { 'cluster_metrics': cluster_metrics, 'pod_analysis': pod_analysis, 'node_analysis': node_analysis, 'resource_analysis': resource_analysis, 'network_analysis': network_analysis, 'issues': self.issues, 'recommendations': self.generate_recommendations(), 'health_score': self.calculate_health_score(cluster_metrics) } return report except Exception as e: return {'error': f'分析集群失败: {e}'} def get_cluster_metrics(self) -> ClusterMetrics: """获取集群指标""" try: # 获取节点数量 nodes = self.v1.list_node() node_count = len(nodes.items) # 获取Pod数量 pods = self.v1.list_pod_for_all_namespaces() pod_count = len(pods.items) # 获取Service数量 services = self.v1.list_service_for_all_namespaces() service_count = len(services.items) # 获取Deployment数量 deployments = self.apps_v1.list_deployment_for_all_namespaces() deployment_count = len(deployments.items) # 获取Namespace数量 namespaces = self.v1.list_namespace() namespace_count = len(namespaces.items) # 计算资源容量 cpu_capacity = 0 memory_capacity = 0 for node in nodes.items: cpu_capacity += self.parse_cpu(node.status.capacity.get('cpu', '0')) memory_capacity += self.parse_memory(node.status.capacity.get('memory', '0')) return ClusterMetrics( node_count=node_count, pod_count=pod_count, service_count=service_count, deployment_count=deployment_count, namespace_count=namespace_count, cpu_capacity=cpu_capacity, memory_capacity=memory_capacity, cpu_usage=0, # 需要通过metrics server获取 memory_usage=0 ) except Exception as e: raise Exception(f'获取集群指标失败: {e}') def analyze_pods(self) -> Dict[str, Any]: """分析Pod状态""" try: pods = self.v1.list_pod_for_all_namespaces() pod_metrics = [] status_counts = defaultdict(int) namespace_counts = defaultdict(int) for pod in pods.items: # 获取Pod指标 metrics = self.get_pod_metrics(pod) pod_metrics.append(metrics) # 统计状态 status_counts[pod.status.phase] += 1 namespace_counts[pod.metadata.namespace] += 1 # 检查Pod问题 self.check_pod_issues(pod, metrics) return { 'total_pods': len(pods.items), 'status_distribution': dict(status_counts), 'namespace_distribution': dict(namespace_counts), 'pod_metrics': pod_metrics, 'issues': [issue for issue in self.issues if issue.type.startswith('pod_')] } except Exception as e: return {'error': f'分析Pod失败: {e}'} def get_pod_metrics(self, pod) -> PodMetrics: """获取Pod指标""" try: # 解析资源请求和限制 cpu_requests = 0 cpu_limits = 0 memory_requests = 0 memory_limits = 0 if pod.spec.containers: for container in pod.spec.containers: if container.resources: if container.resources.requests: cpu_requests += self.parse_cpu(container.resources.requests.get('cpu', '0')) memory_requests += self.parse_memory(container.resources.requests.get('memory', '0')) if container.resources.limits: cpu_limits += self.parse_cpu(container.resources.limits.get('cpu', '0')) memory_limits += self.parse_memory(container.resources.limits.get('memory', '0')) # 计算重启次数 restart_count = 0 if pod.status.container_statuses: for container_status in pod.status.container_statuses: restart_count += container_status.restart_count # 计算Pod年龄 creation_time = pod.metadata.creation_timestamp age_seconds = int((time.time() - creation_time.timestamp())) return PodMetrics( name=pod.metadata.name, namespace=pod.metadata.namespace, status=pod.status.phase, node_name=pod.spec.node_name or '', cpu_requests=cpu_requests, cpu_limits=cpu_limits, memory_requests=memory_requests, memory_limits=memory_limits, restart_count=restart_count, age_seconds=age_seconds ) except Exception as e: raise Exception(f'获取Pod指标失败: {e}') def check_pod_issues(self, pod, metrics: PodMetrics) -> None: """检查Pod问题""" # 检查Pod状态 if pod.status.phase == 'Pending': self.issues.append(ClusterIssue( severity='medium', type='pod_pending', resource=pod.metadata.name, namespace=pod.metadata.namespace, message='Pod处于Pending状态', suggestion='检查资源配额、节点调度或PVC绑定状态' )) elif pod.status.phase == 'Failed': self.issues.append(ClusterIssue( severity='high', type='pod_failed', resource=pod.metadata.name, namespace=pod.metadata.namespace, message='Pod处于Failed状态', suggestion='检查Pod日志和事件，确定失败原因' )) # 检查重启次数 if metrics.restart_count > 5: self.issues.append(ClusterIssue( severity='high', type='pod_crash_loop', resource=pod.metadata.name, namespace=pod.metadata.namespace, message=f'Pod重启次数过多: {metrics.restart_count}', suggestion='检查应用日志，可能存在CrashLoopBackOff问题' )) # 检查资源配置 if metrics.cpu_requests == 0 or metrics.memory_requests == 0: self.issues.append(ClusterIssue( severity='medium', type='pod_no_requests', resource=pod.metadata.name, namespace=pod.metadata.namespace, message='Pod没有设置资源请求', suggestion='设置CPU和内存请求以确保调度稳定性' )) if metrics.cpu_limits == 0 or metrics.memory_limits == 0: self.issues.append(ClusterIssue( severity='low', type='pod_no_limits', resource=pod.metadata.name, namespace=pod.metadata.namespace, message='Pod没有设置资源限制', suggestion='设置资源限制防止资源耗尽' )) def analyze_nodes(self) -> Dict[str, Any]: """分析节点状态""" try: nodes = self.v1.list_node() node_analysis = [] for node in nodes.items: analysis = { 'name': node.metadata.name, 'status': self.get_node_status(node), 'roles': self.get_node_roles(node), 'capacity': node.status.capacity, 'allocatable': node.status.allocatable, 'conditions': node.status.conditions, 'issues': [] } # 检查节点问题 for condition in node.status.conditions or []: if condition.type == 'Ready' and condition.status != 'True': analysis['issues'].append({ 'severity': 'high', 'message': f'节点{node.metadata.name}不可用', 'suggestion': '检查节点状态和网络连接' }) node_analysis.append(analysis) return { 'total_nodes': len(nodes.items), 'nodes': node_analysis, 'ready_nodes': len([n for n in node_analysis if n['status'] == 'Ready']) } except Exception as e: return {'error': f'分析节点失败: {e}'} def get_node_status(self, node) -> str: """获取节点状态""" for condition in node.status.conditions or []: if condition.type == 'Ready': return 'Ready' if condition.status == 'True' else 'NotReady' return 'Unknown' def get_node_roles(self, node) -> List[str]: """获取节点角色""" roles = [] labels = node.metadata.labels or {} for label in labels: if label.startswith('node-role.kubernetes.io/'): role = label.replace('node-role.kubernetes.io/', '') if role: roles.append(role) return roles or ['worker'] def analyze_resource_usage(self) -> Dict[str, Any]: """分析资源使用情况""" try: # 这里需要metrics-server支持 # 简化实现，返回基本信息 return { 'message': '需要安装metrics-server获取详细资源使用信息', 'recommendation': '请安装metrics-server: kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/latest/download/components.yaml' } except Exception as e: return {'error': f'分析资源使用失败: {e}'} def analyze_network(self) -> Dict[str, Any]: """分析网络配置""" try: # 获取所有Service services = self.v1.list_service_for_all_namespaces() service_analysis = [] for service in services.items: analysis = { 'name': service.metadata.name, 'namespace': service.metadata.namespace, 'type': service.spec.type,