影刀RPA Kubernetes自动化：Pod管理与日志采集

影刀RPA Kubernetes自动化：Pod管理与日志采集

什么情况用什么 → 怎么做 → 有什么坑
作者：林焱 | 飞行社出品

什么情况用什么

企业用K8s管理容器，每天要手动检查Pod状态、查看日志、重启故障服务，重复性工作多到爆。

这套方案适合：

• 运维团队自动化K8s资源管理
• 定时清理已完成/失败的Pod
• 自动采集Pod日志并分析

核心工具：影刀RPA + kubernetes Python客户端（kubernetes） + 日志分析

怎么做

第一步：安装K8s Python客户端并配置连接

# 安装K8s Python客户端pipinstallkubernetes[video(video-GP5mx6H8-1783576802836)(type-csdn)(url-https://live.csdn.net/v/embed/526818)(image-https://v-blog.csdnimg.cn/asset/582d14c3bd0451c5399cd990b56e2a0d/cover/Cover0.jpg)(title-拼多多店群自动化报活动上架！)]

importkubernetesimportosdefinit_k8s_client(kubeconfig_path=None):""" 初始化K8s客户端 kubeconfig_path: kubeconfig文件路径（如 "~/.kube/config"） """try:ifkubeconfig_path:# 方法1：使用kubeconfig文件（推荐）kubernetes.config.load_kube_config(config_file=kubeconfig_path)else:# 方法2：使用默认kubeconfig（~/.kube/config）kubernetes.config.load_kube_config()# 验证连接v1=kubernetes.client.CoreV1Api()pods=v1.list_pod_for_all_namespaces(watch=False)print(f"K8s连接成功，集群中有{len(pods.items)}个Pod")returnTrueexceptExceptionase:print(f"⚠️ K8s连接失败:{e}")returnFalse# 使用示例success=init_k8s_client(kubeconfig_path="C:/Users/32108/.kube/config")

第二步：自动化Pod管理

deflist_pods(namespace=None,label_selector=None):"""列出所有Pod"""v1=kubernetes.client.CoreV1Api()ifnamespace:pods=v1.list_namespaced_pod(namespace=namespace,label_selector=label_selector)else:pods=v1.list_pod_for_all_namespaces(label_selector=label_selector)pod_list=[]forpodinpods.items:pod_list.append({"名称":pod.metadata.name,"命名空间":pod.metadata.namespace,"状态":pod.status.phase,"Pod IP":pod.status.pod_ip,"节点":pod.spec.node_name,"重启次数":pod.status.container_statuses[0].restart_countifpod.status.container_statuseselse0,"创建时间":pod.metadata.creation_timestamp})returnpod_listdefrestart_deployment(namespace,deployment_name):"""重启Deployment（通过滚动重启）"""apps_v1=kubernetes.client.AppsV1Api()try:# 方法1：通过patch annotations触发滚动重启body={"spec":{"template":{"metadata":{"annotations":{"kubectl.kubernetes.io/restartedAt":datetime.now().isoformat()}}}}}apps_v1.patch_namespaced_deployment(name=deployment_name,namespace=namespace,body=body)print(f"Deployment{deployment_name}已触发滚动重启")returnTrueexceptExceptionase:print(f"重启Deployment失败:{e}")returnFalsedefclean_completed_pods(namespace="default",age_hours=24):"""清理已完成的Pod（Succeeded/Failed状态，超过指定时间）"""v1=kubernetes.client.CoreV1Api()# 获取Pod列表pods=v1.list_namespaced_pod(namespace=namespace,field_selector="status.phase in (Succeeded,Failed)")deleted_count=0cutoff_time=datetime.now()-timedelta(hours=age_hours)forpodinpods.items:# 检查Pod完成时间ifpod.status.start_time:finish_time=pod.status.start_timeiffinish_time<cutoff_time:# 删除Podv1.delete_namespaced_pod(name=pod.metadata.name,namespace=namespace,body=kubernetes.client.V1DeleteOptions(propagation_policy='Foreground'))print(f"已删除Pod:{pod.metadata.name}")deleted_count+=1print(f"共删除{deleted_count}个已完成/失败的Pod")returndeleted_count

第三步：自动采集Pod日志

defget_pod_logs(namespace,pod_name,container_name=None,tail_lines=100):"""获取Pod日志"""v1=kubernetes.client.CoreV1Api()try:logs=v1.read_namespaced_pod_log(name=pod_name,namespace=namespace,container=container_name,tail_lines=tail_lines)returnlogsexceptExceptionase:print(f"获取Pod日志失败:{e}")return""defcollect_pod_logs(namespace,label_selector,save_dir,tail_lines=1000):"""批量采集Pod日志"""importos v1=kubernetes.client.CoreV1Api()# 创建保存目录os.makedirs(save_dir,exist_ok=True)# 获取Pod列表pods=v1.list_namespaced_pod(namespace=namespace,label_selector=label_selector)collected=[]forpodinpods.items:pod_name=pod.metadata.name# 获取所有容器的日志forcontainerinpod.spec.containers:container_name=container.name logs=get_pod_logs(namespace,pod_name,container_name,tail_lines)iflogs:# 保存到文件file_path=os.path.join(save_dir,f"{pod_name}_{container_name}.log")withopen(file_path,"w",encoding="utf-8")asf:f.write(logs)collected.append({"pod":pod_name,"container":container_name,"log_file":file_path,"size":len(logs)})print(f"共采集{len(collected)}个容器的日志")returncollecteddefanalyze_logs_for_errors(log_file):"""分析日志中的错误"""error_keywords=["error","exception","failed","fatal","错误","异常","失败"]errors=[]withopen(log_file,"r",encoding="utf-8",errors="ignore")asf:forline_num,lineinenumerate(f,1):forkeywordinerror_keywords:ifkeywordinline.lower():errors.append({"line_num":line_num,"keyword":keyword,"content":line.strip()})breakreturnerrors

第四步：监控Pod健康状态

defcheck_pod_health(namespace=None):"""检查Pod健康状态"""v1=kubernetes.client.CoreV1Api()ifnamespace:pods=v1.list_namespaced_pod(namespace=namespace)else:pods=v1.list_pod_for_all_namespaces()alerts=[]forpodinpods.items:pod_name=pod.metadata.name namespace=pod.metadata.namespace# 1. 检查Pod状态ifpod.status.phase!="Running":alerts.append({"pod":pod_name,"namespace":namespace,"issue":f"Pod状态异常:{pod.status.phase}","severity":"high"})# 2. 检查容器重启次数ifpod.status.container_statuses:forcontainer_statusinpod.status.container_statuses:ifcontainer_status.restart_count>5:alerts.append({"pod":pod_name,"namespace":namespace,"issue":f"容器{container_status.name}重启次数过多:{container_status.restart_count}次","severity":"medium"})# 3. 检查资源使用情况（需要Metrics Server）try:# 获取Pod资源使用（需要metrics-server）metrics=v1.read_namespaced_pod_metrics(pod_name,namespace)forcontainerinmetrics.containers:# CPU使用率（需要自定义阈值）cpu_usage=container.usage["cpu"]if"n"incpu_usage:# 纳秒cpu_value=int(cpu_usage.replace("n",""))/1e9# 转换为核ifcpu_value>0.8:# 超过0.8核alerts.append({"pod":pod_name,"namespace":namespace,"issue":f"容器{container.name}CPU使用率过高:{cpu_value:.2f}核","severity":"medium"})exceptException:# Metrics Server可能未安装passreturnalertsdefsend_k8s_alert(alerts,webhook_url):"""发送K8s告警到企微"""ifnotalerts:return# 构造消息内容content="⚠️ **Kubernetes Pod健康告警**\n\n"high_severity=[aforainalertsifa["severity"]=="high"]medium_severity=[aforainalertsifa["severity"]=="medium"]ifhigh_severity:content+=f"**高优先级告警 ({len(high_severity)}项):**\n"foralertinhigh_severity[:5]:# 最多显示5条content+=f"-{alert['pod']}({alert['namespace']}):{alert['issue']}\n"content+="\n"ifmedium_severity:content+=f"**中优先级告警 ({len(medium_severity)}项):**\n"foralertinmedium_severity[:5]:# 最多显示5条content+=f"-{alert['pod']}({alert['namespace']}):{alert['issue']}\n"# 发送到企微payload={"msgtype":"markdown","markdown":{"content":content}}importrequests response=requests.post(webhook_url,json=payload)ifresponse.json().get("errcode")==0:print(f"成功发送{len(alerts)}条K8s告警")else:print(f"发送告警失败:{response.text}")

第五步：影刀RPA完整流程编排

【定时触发】每天早上9点、下午3点各运行一次 ↓ 【Python节点】init_k8s_client() → 初始化K8s客户端 ↓ 【Python节点】list_pods() → 列出所有Pod ↓ 【Python节点】check_pod_health() → 检查Pod健康状态 ↓ 【条件判断】是否有告警？ ├─ 是 → 【企微通知】发送K8s告警 └─ 否 → 继续 ↓ 【Python节点】clean_completed_pods() → 清理已完成的Pod ↓ 【Python节点】collect_pod_logs() → 采集Pod日志 ↓ 【Python节点】analyze_logs_for_errors() → 分析日志错误 ↓ 【条件判断】是否发现错误？ ├─ 是 → 【企微通知】发送错误日志告警 └─ 否 → 继续 ↓ 【生成报告】"K8s运维日报.xlsx" → 包含：Pod列表、健康状态、清理记录、错误日志 ↓ 【发送邮件】将报告发送给运维团队

有什么坑

坑1：K8s配置文件权限问题

kubeconfig文件包含敏感信息，权限设置不当可能导致安全问题。

解决方案：

1. 限制文件权限：chmod 600 ~/.kube/config
2. 使用Service Account：在Pod内运行时，使用Service Account自动认证
3. 使用RBAC：限制K8s API访问权限

# 在Pod内运行时，无需kubeconfig，使用默认Service Account# 只需将Pod的Service Account绑定到具有相应权限的ClusterRole# 例如：# kubectl create clusterrole pod-reader --verb=get,list,watch --resource=pods# kubectl create clusterrolebinding read-pods --clusterrole=pod-reader --serviceaccount=default:default

坑2：Pod日志过大，采集耗时

大流量应用的Pod日志可能非常大（GB级别），全部采集会占用大量磁盘空间和网络带宽。

解决方案：

1. 只采集最近N行：tail_lines=1000
2. 过滤关键字：只采集包含错误关键字的行
3. 压缩存储：采集后压缩日志文件

defcollect_pod_logs_optimized(namespace,pod_name,container_name,save_dir):"""优化版日志采集（只采集错误日志）"""logs=get_pod_logs(namespace,pod_name,container_name,tail_lines=10000)ifnotlogs:returnNone# 只保留包含错误关键字的行error_keywords=["error","exception","failed","fatal","错误","异常","失败"]error_lines=[]forlineinlogs.split("\n"):ifany(keywordinline.lower()forkeywordinerror_keywords):error_lines.append(line)ifnoterror_lines:returnNone# 保存到文件os.makedirs(save_dir,exist_ok=True)file_path=os.path.join(save_dir,f"{pod_name}_{container_name}_errors.log")withopen(file_path,"w",encoding="utf-8")asf:f.write("\n".join(error_lines))# 压缩文件importgzipwithopen(file_path,"rb")asf_in:withgzip.open(f"{file_path}.gz","wb")asf_out:shutil.copyfileobj(f_in,f_out)os.remove(file_path)# 删除原始文件print(f"已采集错误日志并压缩:{file_path}.gz")returnf"{file_path}.gz"

坑3：K8s API Server访问频率限制

K8s API Server有访问频率限制，频繁调用API可能导致限流。

解决方案：

1. 加入缓存：缓存Pod列表，避免频繁调用list API
2. 使用Watch机制：监听Pod变化，而不是轮询
3. 控制并发：限制同时调用的线程数

# 使用Watch机制监听Pod变化（更高效）fromkubernetesimportwatchdefwatch_pod_changes(namespace):"""监听Pod变化（事件驱动）"""v1=kubernetes.client.CoreV1Api()w=watch.Watch()foreventinw.stream(v1.list_namespaced_pod,namespace=namespace):print(f"事件:{event['type']}Pod:{event['object'].metadata.name}")# 处理事件（例如：Pod删除时发送告警）ifevent['type']=="DELETED":send_pod_deleted_alert(event['object'])

坑4：跨集群管理复杂

如果有多个K8s集群，需要分别配置kubeconfig，管理复杂。

解决方案：

1. 使用kubeconfig文件合并：将多个集群配置合并到一个kubeconfig文件
2. 使用K8s Federation：管理多个集群
3. 循环处理多个集群：

defmanage_multiple_clusters(cluster_configs):"""管理多个K8s集群"""forcluster_name,kubeconfigincluster_configs.items():print(f"处理集群:{cluster_name}")# 切换到对应集群的kubeconfigos.environ["KUBECONFIG"]=kubeconfig# 初始化客户端success=init_k8s_client()ifsuccess:# 执行操作pods=list_pods()print(f" 集群{cluster_name}中有{len(pods)}个Pod")# 检查健康状态alerts=check_pod_health()ifalerts:send_k8s_alert(alerts,WEBHOOK_URL)else:print(f" 连接集群{cluster_name}失败")

总结

实际落地建议：

1. 先小范围测试：在一个非生产环境测试完整流程
2. 使用RBAC限制权限：只给必要的权限，降低安全风险
3. 做好异常处理：网络超时、API限流要有重试机制
4. 遵守K8s最佳实践：使用Deployment而不是裸Pod，便于管理

K8s自动化能为运维团队节省50%以上的日常管理时间，同时提高系统稳定性和可靠性。