影刀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_pathNone): 初始化K8s客户端 kubeconfig_path: kubeconfig文件路径如 ~/.kube/config try:ifkubeconfig_path:# 方法1使用kubeconfig文件推荐kubernetes.config.load_kube_config(config_filekubeconfig_path)else:# 方法2使用默认kubeconfig~/.kube/configkubernetes.config.load_kube_config()# 验证连接v1kubernetes.client.CoreV1Api()podsv1.list_pod_for_all_namespaces(watchFalse)print(fK8s连接成功集群中有{len(pods.items)}个Pod)returnTrueexceptExceptionase:print(f⚠️ K8s连接失败:{e})returnFalse# 使用示例successinit_k8s_client(kubeconfig_pathC:/Users/32108/.kube/config)第二步自动化Pod管理deflist_pods(namespaceNone,label_selectorNone):列出所有Podv1kubernetes.client.CoreV1Api()ifnamespace:podsv1.list_namespaced_pod(namespacenamespace,label_selectorlabel_selector)else:podsv1.list_pod_for_all_namespaces(label_selectorlabel_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_v1kubernetes.client.AppsV1Api()try:# 方法1通过patch annotations触发滚动重启body{spec:{template:{metadata:{annotations:{kubectl.kubernetes.io/restartedAt:datetime.now().isoformat()}}}}}apps_v1.patch_namespaced_deployment(namedeployment_name,namespacenamespace,bodybody)print(fDeployment{deployment_name}已触发滚动重启)returnTrueexceptExceptionase:print(f重启Deployment失败:{e})returnFalsedefclean_completed_pods(namespacedefault,age_hours24):清理已完成的PodSucceeded/Failed状态超过指定时间v1kubernetes.client.CoreV1Api()# 获取Pod列表podsv1.list_namespaced_pod(namespacenamespace,field_selectorstatus.phase in (Succeeded,Failed))deleted_count0cutoff_timedatetime.now()-timedelta(hoursage_hours)forpodinpods.items:# 检查Pod完成时间ifpod.status.start_time:finish_timepod.status.start_timeiffinish_timecutoff_time:# 删除Podv1.delete_namespaced_pod(namepod.metadata.name,namespacenamespace,bodykubernetes.client.V1DeleteOptions(propagation_policyForeground))print(f已删除Pod:{pod.metadata.name})deleted_count1print(f共删除{deleted_count}个已完成/失败的Pod)returndeleted_count第三步自动采集Pod日志defget_pod_logs(namespace,pod_name,container_nameNone,tail_lines100):获取Pod日志v1kubernetes.client.CoreV1Api()try:logsv1.read_namespaced_pod_log(namepod_name,namespacenamespace,containercontainer_name,tail_linestail_lines)returnlogsexceptExceptionase:print(f获取Pod日志失败:{e})returndefcollect_pod_logs(namespace,label_selector,save_dir,tail_lines1000):批量采集Pod日志importos v1kubernetes.client.CoreV1Api()# 创建保存目录os.makedirs(save_dir,exist_okTrue)# 获取Pod列表podsv1.list_namespaced_pod(namespacenamespace,label_selectorlabel_selector)collected[]forpodinpods.items:pod_namepod.metadata.name# 获取所有容器的日志forcontainerinpod.spec.containers:container_namecontainer.name logsget_pod_logs(namespace,pod_name,container_name,tail_lines)iflogs:# 保存到文件file_pathos.path.join(save_dir,f{pod_name}_{container_name}.log)withopen(file_path,w,encodingutf-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,encodingutf-8,errorsignore)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(namespaceNone):检查Pod健康状态v1kubernetes.client.CoreV1Api()ifnamespace:podsv1.list_namespaced_pod(namespacenamespace)else:podsv1.list_pod_for_all_namespaces()alerts[]forpodinpods.items:pod_namepod.metadata.name namespacepod.metadata.namespace# 1. 检查Pod状态ifpod.status.phase!Running:alerts.append({pod:pod_name,namespace:namespace,issue:fPod状态异常:{pod.status.phase},severity:high})# 2. 检查容器重启次数ifpod.status.container_statuses:forcontainer_statusinpod.status.container_statuses:ifcontainer_status.restart_count5:alerts.append({pod:pod_name,namespace:namespace,issue:f容器{container_status.name}重启次数过多:{container_status.restart_count}次,severity:medium})# 3. 检查资源使用情况需要Metrics Servertry:# 获取Pod资源使用需要metrics-servermetricsv1.read_namespaced_pod_metrics(pod_name,namespace)forcontainerinmetrics.containers:# CPU使用率需要自定义阈值cpu_usagecontainer.usage[cpu]ifnincpu_usage:# 纳秒cpu_valueint(cpu_usage.replace(n,))/1e9# 转换为核ifcpu_value0.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\nhigh_severity[aforainalertsifa[severity]high]medium_severity[aforainalertsifa[severity]medium]ifhigh_severity:contentf**高优先级告警 ({len(high_severity)}项):**\nforalertinhigh_severity[:5]:# 最多显示5条contentf-{alert[pod]}({alert[namespace]}):{alert[issue]}\ncontent\nifmedium_severity:contentf**中优先级告警 ({len(medium_severity)}项):**\nforalertinmedium_severity[:5]:# 最多显示5条contentf-{alert[pod]}({alert[namespace]}):{alert[issue]}\n# 发送到企微payload{msgtype:markdown,markdown:{content:content}}importrequests responserequests.post(webhook_url,jsonpayload)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列表、健康状态、清理记录、错误日志 ↓ 【发送邮件】将报告发送给运维团队有什么坑坑1K8s配置文件权限问题kubeconfig文件包含敏感信息权限设置不当可能导致安全问题。解决方案限制文件权限chmod 600 ~/.kube/config使用Service Account在Pod内运行时使用Service Account自动认证使用RBAC限制K8s API访问权限# 在Pod内运行时无需kubeconfig使用默认Service Account# 只需将Pod的Service Account绑定到具有相应权限的ClusterRole# 例如# kubectl create clusterrole pod-reader --verbget,list,watch --resourcepods# kubectl create clusterrolebinding read-pods --clusterrolepod-reader --serviceaccountdefault:default坑2Pod日志过大采集耗时大流量应用的Pod日志可能非常大GB级别全部采集会占用大量磁盘空间和网络带宽。解决方案只采集最近N行tail_lines1000过滤关键字只采集包含错误关键字的行压缩存储采集后压缩日志文件defcollect_pod_logs_optimized(namespace,pod_name,container_name,save_dir):优化版日志采集只采集错误日志logsget_pod_logs(namespace,pod_name,container_name,tail_lines10000)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_okTrue)file_pathos.path.join(save_dir,f{pod_name}_{container_name}_errors.log)withopen(file_path,w,encodingutf-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坑3K8s API Server访问频率限制K8s API Server有访问频率限制频繁调用API可能导致限流。解决方案TEMU店群矩阵自动化运营核价报活动加入缓存缓存Pod列表避免频繁调用list API使用Watch机制监听Pod变化而不是轮询控制并发限制同时调用的线程数# 使用Watch机制监听Pod变化更高效fromkubernetesimportwatchdefwatch_pod_changes(namespace):监听Pod变化事件驱动v1kubernetes.client.CoreV1Api()wwatch.Watch()foreventinw.stream(v1.list_namespaced_pod,namespacenamespace):print(f事件:{event[type]}Pod:{event[object].metadata.name})# 处理事件例如Pod删除时发送告警ifevent[type]DELETED:send_pod_deleted_alert(event[object])坑4跨集群管理复杂如果有多个K8s集群需要分别配置kubeconfig管理复杂。解决方案使用kubeconfig文件合并将多个集群配置合并到一个kubeconfig文件使用K8s Federation管理多个集群循环处理多个集群defmanage_multiple_clusters(cluster_configs):管理多个K8s集群forcluster_name,kubeconfigincluster_configs.items():print(f处理集群:{cluster_name})# 切换到对应集群的kubeconfigos.environ[KUBECONFIG]kubeconfig# 初始化客户端successinit_k8s_client()ifsuccess:# 执行操作podslist_pods()print(f 集群{cluster_name}中有{len(pods)}个Pod)# 检查健康状态alertscheck_pod_health()ifalerts:send_k8s_alert(alerts,WEBHOOK_URL)else:print(f 连接集群{cluster_name}失败)总结功能节省时间附加价值Pod状态监控每天省30分钟及时发现故障日志自动采集每天省1小时便于问题排查自动清理Pod每周省30分钟节省集群资源健康状态告警—提高系统稳定性实际落地建议先小范围测试在一个非生产环境测试完整流程使用RBAC限制权限只给必要的权限降低安全风险做好异常处理网络超时、API限流要有重试机制遵守K8s最佳实践使用Deployment而不是裸Pod便于管理K8s自动化能为运维团队节省50%以上的日常管理时间同时提高系统稳定性和可靠性。