# Kubernetes编排 > 当实施Kubernetes编排时，分析集群架构设计，优化容器编排性能，解决K8s相关问题。验证资源配置，设计服务部署策略，和最佳实践。 - Skill: `microwind/kubernetes` (Agent Skill, multi-file: 3 files) - Install (CLI): `npx skillmds@latest add microwind/kubernetes` - Raw SKILL.md: https://api.skillmd.com/api/skills/microwind/kubernetes/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: DevOps & Infra - License: MIT - Author: microwind (https://skillmd.com/u/microwind) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/microwind/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, 'cluster_ip': service.spec.cluster_ip, 'external_ips': service.spec.external_i_ps or [], 'ports': service.spec.ports or [], 'selector': service.spec.selector or {}, 'issues': [] } # 检查Service问题 if service.spec.type == 'LoadBalancer' and not service.spec.cluster_ip: analysis['issues'].append({ 'severity': 'medium', 'message': 'LoadBalancer Service没有分配IP', 'suggestion': '检查云提供商配置或网络策略' }) if service.spec.selector and not service.spec.selector: analysis['issues'].append({ 'severity': 'low', 'message': 'Service没有选择器', 'suggestion': '添加选择器或将Service类型改为ExternalName' }) service_analysis.append(analysis) # 获取Ingress try: ingresses = self.networking_v1.list_ingress_for_all_namespaces() ingress_analysis = [] for ingress in ingresses.items: analysis = { 'name': ingress.metadata.name, 'namespace': ingress.metadata.namespace, 'rules': ingress.spec.rules or [], 'tls': ingress.spec.tls or [], 'backend': ingress.spec.backend, 'issues': [] } ingress_analysis.append(analysis) except: ingress_analysis = [] return { 'total_services': len(services.items), 'services': service_analysis, 'total_ingresses': len(ingress_analysis) if ingress_analysis else 0, 'ingresses': ingress_analysis } except Exception as e: return {'error': f'分析网络失败: {e}'} def generate_recommendations(self) -> List[Dict[str, str]]: """生成优化建议""" recommendations = [] # 基于问题生成建议 issue_counts = defaultdict(int) for issue in self.issues: issue_counts[issue.type] += 1 if issue_counts['pod_pending'] > 0: recommendations.append({ 'priority': 'high', 'type': 'scheduling', 'message': f'{issue_counts["pod_pending"]}个Pod处于Pending状态', 'suggestion': '检查集群资源可用性和调度配置' }) if issue_counts['pod_failed'] > 0: recommendations.append({ 'priority': 'critical', 'type': 'reliability', 'message': f'{issue_counts["pod_failed"]}个Pod处于Failed状态', 'suggestion': '检查Pod日志和事件，修复应用问题' }) if issue_counts['pod_no_requests'] > 5: recommendations.append({ 'priority': 'medium', 'type': 'resource_management', 'message': f'{issue_counts["pod_no_requests"]}个Pod没有设置资源请求', 'suggestion': '为所有Pod设置资源请求确保调度稳定性' }) return recommendations def calculate_health_score(self, metrics: ClusterMetrics) -> int: """计算集群健康评分""" score = 100 # 根据问题扣分 for issue in self.issues: if issue.severity == 'critical': score -= 20 elif issue.severity == 'high': score -= 10 elif issue.severity == 'medium': score -= 5 elif issue.severity == 'low': score -= 2 # 根据节点状态调整 if metrics.node_count == 0: score = 0 return max(0, int(score)) def parse_cpu(self, cpu_str: str) -> float: """解析CPU字符串""" if not cpu_str: return 0 cpu_str = cpu_str.lower() if cpu_str.endswith('m'): return float(cpu_str[:-1]) / 1000 else: return float(cpu_str) def parse_memory(self, memory_str: str) -> int: """解析内存字符串""" if not memory_str: return 0 memory_str = memory_str.upper() if memory_str.endswith('KI'): return int(memory_str[:-2]) * 1024 elif memory_str.endswith('MI'): return int(memory_str[:-2]) * 1024 * 1024 elif memory_str.endswith('GI'): return int(memory_str[:-2]) * 1024 * 1024 * 1024 elif memory_str.endswith('K'): return int(memory_str[:-1]) * 1000 elif memory_str.endswith('M'): return int(memory_str[:-1]) * 1000 * 1000 elif memory_str.endswith('G'): return int(memory_str[:-1]) * 1000 * 1000 * 1000 else: return int(memory_str) # Kubernetes部署优化器 class KubernetesDeploymentOptimizer: def __init__(self, kubeconfig_path: Optional[str] = None): self.kubeconfig_path = kubeconfig_path self.load_kubernetes_config() self.apps_v1 = kubernetes.client.AppsV1Api() self.v1 = kubernetes.client.CoreV1Api() 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 optimize_deployment(self, namespace: str, deployment_name: str) -> Dict[str, Any]: """优化Deployment配置""" try: # 获取当前Deployment deployment = self.apps_v1.read_namespaced_deployment( name=deployment_name, namespace=namespace ) # 分析当前配置 current_analysis = self.analyze_deployment_config(deployment) # 生成优化建议 optimization_plan = self.generate_deployment_optimizations(deployment) # 生成优化后的配置 optimized_deployment = self.generate_optimized_deployment(deployment, optimization_plan) return { 'namespace': namespace, 'deployment_name': deployment_name, 'current_analysis': current_analysis, 'optimization_plan': optimization_plan, 'optimized_deployment': optimized_deployment, 'estimated_improvements': self.estimate_deployment_improvements(deployment, optimized_deployment) } except Exception as e: return {'error': f'优化Deployment失败: {e}'} def analyze_deployment_config(self, deployment) -> Dict[str, Any]: """分析Deployment配置""" analysis = { 'replicas': deployment.spec.replicas, 'strategy': deployment.spec.strategy.type if deployment.spec.strategy else 'RollingUpdate', 'pod_template': { 'containers': [] }, 'issues': [] } # 分析容器配置 if deployment.spec.template.spec.containers: for container in deployment.spec.template.spec.containers: container_analysis = { 'name': container.name, 'image': container.image, 'resources': { 'requests': container.resources.requests if container.resources else {}, 'limits': container.resources.limits if container.resources else {} }, 'ports': container.ports or [], 'env_vars': len(container.env or []), 'volume_mounts': len(container.volume_mounts or []), 'liveness_probe': container.liveness_probe is not None, 'readiness_probe': container.readiness_probe is not None, 'startup_probe': container.startup_probe is not None } # 检查容器问题 if not container.resources: analysis['issues'].append({ 'container': container.name, 'severity': 'high', 'message': '容器没有设置资源配置', 'suggestion': '设置CPU和内存的requests和limits' }) elif not container.resources.requests: analysis['issues'].append({ 'container': container.name, 'severity': 'medium', 'message': '容器没有设置资源请求', 'suggestion': '设置CPU和内存的requests' }) if not container.liveness_probe: analysis['issues'].append({ 'container': container.name, 'severity': 'medium', 'message': '容器没有设置存活探针', 'suggestion': '添加livenessProbe检测容器健康状态' }) if not container.readiness_probe: analysis['issues'].append({ 'container': container.name, 'severity': 'medium', 'message': '容器没有设置就绪探针', 'suggestion': '添加readinessProbe检测服务就绪状态' }) analysis['pod_template']['containers'].append(container_analysis) return analysis def generate_deployment_optimizations(self, deployment) -> List[Dict[str, Any]]: """生成Deployment优化建议""" optimizations = [] # 检查副本数 if deployment.spec.replicas == 1: optimizations.append({ 'type': 'availability', 'priority': 'high', 'message': '副本数为1，存在单点故障风险', 'suggestion': '增加副本数提高可用性', 'recommended_value': 3 }) # 检查更新策略 if not deployment.spec.strategy or deployment.spec.strategy.type == 'RollingUpdate': strategy = deployment.spec.strategy if strategy and strategy.rolling_update: max_unavailable = strategy.rolling_update.max_unavailable max_surge = strategy.rolling_update.max_surge if not max_unavailable or max_unavailable == '25%': optimizations.append({ 'type': 'update_strategy', 'priority': 'medium', 'message': '使用默认的滚动更新配置', 'suggestion': '根据业务需求调整maxUnavailable和maxSurge' }) # 检查镜像版本 if deployment.spec.template.spec.containers: for container in deployment.spec.template.spec.containers: if ':latest' in container.image: optimizations.append({ 'type': 'image_management', 'priority': 'high', 'message': f'容器{container.name}使用latest标签', 'suggestion': '使用具体的镜像版本标签确保部署一致性' }) return optimizations def generate_optimized_deployment(self, original_deployment, optimizations: List[Dict[str, Any]]) -> Dict[str, Any]: """生成优化后的Deployment配置""" optimized_deployment = original_deployment # 应用优化建议 for optimization in optimizations: if optimization['type'] == 'availability' and optimization.get('recommended_value'): optimized_deployment.spec.replicas = optimization['recommended_value'] elif optimization['type'] == 'update_strategy': if not optimized_deployment.spec.strategy: optimized_deployment.spec.strategy = kubernetes.client.V1DeploymentStrategy( type='RollingUpdate', rolling_update=kubernetes.client.V1RollingUpdateDeployment( max_unavailable='1', max_surge='1' ) ) return optimized_deployment def estimate_deployment_improvements(self, original, optimized) -> Dict[str, Any]: """估算改进效果""" improvements = { 'availability_score': 0, 'stability_score': 0, 'maintainability_score': 0, 'overall_score': 0 } # 可用性改进 if original.spec.replicas == 1 and optimized.spec.replicas > 1: improvements['availability_score'] += 40 # 稳定性改进 if optimized.spec.template.spec.containers: for container in optimized.spec.template.spec.containers: if container.resources and container.resources.requests: improvements['stability_score'] += 20 # 可维护性改进 if optimized.spec.template.spec.containers: for container in optimized.spec.template.spec.containers: if ':latest' not in container.image: improvements['maintainability_score'] += 15 # 计算总体改进 improvements['overall_score'] = ( improvements['availability_score'] + improvements['stability_score'] + improvements['maintainability_score'] ) // 3 return improvements # 使用示例 def main(): # 集群分析 analyzer = KubernetesClusterAnalyzer() cluster_report = analyzer.analyze_cluster() print("Kubernetes集群分析报告:") print(f"节点数: {cluster_report['cluster_metrics'].node_count}") print(f"Pod数: {cluster_report['cluster_metrics'].pod_count}") print(f"健康评分: {cluster_report['health_score']}") # Deployment优化 optimizer = KubernetesDeploymentOptimizer() deployment_optimization = optimizer.optimize_deployment('default', 'my-app') print(f"\nDeployment优化建议:") for opt in deployment_optimization['optimization_plan']: print(f"- {opt['message']}: {opt['suggestion']}") if __name__ == '__main__': main() ``` ### Kubernetes资源管理器 ```python import kubernetes import yaml import json from typing import List, Dict, Any, Optional from pathlib import Path class KubernetesResourceManager: 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() 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 create_resource_from_yaml(self, yaml_file: str, namespace: str = 'default') -> Dict[str, Any]: """从YAML文件创建资源""" try: with open(yaml_file, 'r', encoding='utf-8') as f: yaml_content = yaml.safe_load_all(f) created_resources = [] for resource_dict in yaml_content: if not resource_dict: continue # 根据资源类型创建相应的资源 resource_kind = resource_dict.get('kind') resource_name = resource_dict.get('metadata', {}).get('name', '') if resource_kind == 'Deployment': deployment = self.create_deployment_from_dict(resource_dict, namespace) created_resources.append({ 'type': 'Deployment', 'name': resource_name, 'namespace': namespace, 'status': 'created' }) elif resource_kind == 'Service': service = self.create_service_from_dict(resource_dict, namespace) created_resources.append({ 'type': 'Service', 'name': resource_name, 'namespace': namespace, 'status': 'created' }) elif resource_kind == 'ConfigMap': configmap = self.create_configmap_from_dict(resource_dict, namespace) created_resources.append({ 'type': 'ConfigMap', 'name': resource_name, 'namespace': namespace, 'status': 'created' }) elif resource_kind == 'Secret': secret = self.create_secret_from_dict(resource_dict, namespace) created_resources.append({ 'type': 'Secret', 'name': resource_name, 'namespace': namespace, 'status': 'created' }) return { 'created_resources': created_resources, 'total_created': len(created_resources), 'namespace': namespace } except Exception as e: return {'error': f'创建资源失败: {e}'} def create_deployment_from_dict(self, deployment_dict: Dict, namespace: str): """从字典创建Deployment""" deployment = kubernetes.client.V1Deployment( api_version='apps/v1', kind='Deployment', metadata=kubernetes.client.V1ObjectMeta( name=deployment_dict['metadata']['name'], namespace=namespace, labels=deployment_dict['metadata'].get('labels', {}) ), spec=kubernetes.client.V1DeploymentSpec( replicas=deployment_dict['spec'].get('replicas', 1), selector=kubernetes.client.V1LabelSelector( match_labels=deployment_dict['spec']['selector']['match_labels'] ), template=self.create_pod_template_from_dict(deployment_dict['spec']['template']) ) ) return self.apps_v1.create_namespaced_deployment( namespace=namespace, body=deployment ) def create_pod_template_from_dict(self, template_dict: Dict) -> kubernetes.client.V1PodTemplateSpec: """从字典创建Pod模板""" containers = [] for container_dict in template_dict['spec']['containers']: container = kubernetes.client.V1Container( name=container_dict['name'], image=container_dict['image'], ports=[ kubernetes.client.V1ContainerPort( container_port=port['containerPort'], protocol=port.get('protocol', 'TCP') ) for port in container_dict.get('ports', []) ], env=[ kubernetes.client.V1EnvVar( name=env['name'], value=env.get('value'), value_from=env.get('valueFrom') ) for env in container_dict.get('env', []) ], resources=self.create_resource_requirements_from_dict( container_dict.get('resources', {}) ) if 'resources' in container_dict else None, liveness_probe=self.create_probe_from_dict( container_dict.get('livenessProbe', {}) ) if 'livenessProbe' in container_dict else None, readiness_probe=self.create_probe_from_dict( container_dict.get('readinessProbe', {}) ) if 'readinessProbe' in container_dict else None ) containers.append(container) return kubernetes.client.V1PodTemplateSpec( metadata=kubernetes.client.V1ObjectMeta( labels=template_dict['metadata'].get('labels', {}), annotations=template_dict['metadata'].get('annotations', {}) ), spec=kubernetes.client.V1PodSpec( containers=containers, restart_policy=template_dict['spec'].get('restartPolicy', 'Always') ) ) def create_resource_requirements_from_dict(self, resources_dict: Dict) -> kubernetes.client.V1ResourceRequirements: """从字典创建资源需求""" return kubernetes.client.V1ResourceRequirements( requests=resources_dict.get('requests', {}), limits=resources_dict.get('limits', {}) ) def create_probe_from_dict(self, probe_dict: Dict) -> kubernetes.client.V1Probe: """从字典创建探针""" return kubernetes.client.V1Probe( http_get=kubernetes.client.V1HTTPGetAction( path=probe_dict.get('httpGet', {}).get('path', '/'), port=probe_dict.get('httpGet', {}).get('port', 80) ) if 'httpGet' in probe_dict else None, tcp_socket=kubernetes.client.V1TCPSocketAction( port=probe_dict.get('tcpSocket', {}).get('port', 80) ) if 'tcpSocket' in probe_dict else None, initial_delay_seconds=probe_dict.get('initialDelaySeconds', 30), period_seconds=probe_dict.get('periodSeconds', 10), timeout_seconds=probe_dict.get('timeoutSeconds', 5), failure_threshold=probe_dict.get('failureThreshold', 3) ) def create_service_from_dict(self, service_dict: Dict, namespace: str): """从字典创建Service""" service = kubernetes.client.V1Service( api_version='v1', kind='Service', metadata=kubernetes.client.V1ObjectMeta( name=service_dict['metadata']['name'], namespace=namespace, labels=service_dict['metadata'].get('labels', {}) ), spec=kubernetes.client.V1ServiceSpec( type=service_dict['spec'].get('type', 'ClusterIP'), selector=service_dict['spec'].get('selector', {}), ports=[ kubernetes.client.V1ServicePort( port=port['port'], target_port=port.get('targetPort', port['port']), protocol=port.get('protocol', 'TCP') ) for port in service_dict['spec']['ports'] ] ) ) return self.v1.create_namespaced_service( namespace=namespace, body=service ) def create_configmap_from_dict(self, configmap_dict: Dict, namespace: str): """从字典创建ConfigMap""" configmap = kubernetes.client.V1ConfigMap( api_version='v1', kind='ConfigMap', metadata=kubernetes.client.V1ObjectMeta( name=configmap_dict['metadata']['name'], namespace=namespace, labels=configmap_dict['metadata'].get('labels', {}) ), data=configmap_dict.get('data', {}), binary_data=configmap_dict.get('binaryData', {}) ) return self.v1.create_namespaced_config_map( namespace=namespace, body=configmap ) def create_secret_from_dict(self, secret_dict: Dict, namespace: str): """从字典创建Secret""" import base64 # 编码敏感数据 data = {} if 'data' in secret_dict: for key, value in secret_dict['data'].items(): if isinstance(value, str): data[key] = base64.b64encode(value.encode()).decode() else: data[key] = value secret = kubernetes.client.V1Secret( api_version='v1', kind='Secret', metadata=kubernetes.client.V1ObjectMeta( name=secret_dict['metadata']['name'], namespace=namespace, labels=secret_dict['metadata'].get('labels', {}) ), type=secret_dict.get('type', 'Opaque'), data=data, string_data=secret_dict.get('stringData', {}) ) return self.v1.create_namespaced_secret( namespace=namespace, body=secret ) def scale_deployment(self, namespace: str, deployment_name: str, replicas: int) -> Dict[str, Any]: """扩缩容Deployment""" try: # 获取当前Deployment deployment = self.apps_v1.read_namespaced_deployment( name=deployment_name, namespace=namespace ) # 更新副本数 deployment.spec.replicas = replicas # 应用更新 updated_deployment = self.apps_v1.patch_namespaced_deployment( name=deployment_name, namespace=namespace, body=deployment ) return { 'deployment_name': deployment_name, 'namespace': namespace, …(truncated)