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

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

---

什么情况用什么

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

这套方案适合：

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

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

---

怎么做

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

# 安装K8s Python客户端
pip install kubernetes

[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-拼多多店群自动化报活动上架！)]

import kubernetes
import os

def init_k8s_client(kubeconfig_path=None):
    """
    初始化K8s客户端
    kubeconfig_path: kubeconfig文件路径（如 "~/.kube/config"）
    """
    
    try:
        if kubeconfig_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")
        return True
        
    except Exception as e:
        print(f"⚠️ K8s连接失败: {e}")
        return False

# 使用示例
success = init_k8s_client(kubeconfig_path="C:/Users/32108/.kube/config")

第二步：自动化Pod管理

def list_pods(namespace=None, label_selector=None):
    """列出所有Pod"""
    
    v1 = kubernetes.client.CoreV1Api()
    
    if namespace:
        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 = []
    for pod in pods.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_count if pod.status.container_statuses else 0,
            "创建时间": pod.metadata.creation_timestamp
        })
    
    return pod_list

def restart_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} 已触发滚动重启")
        return True
        
    except Exception as e:
        print(f"重启Deployment失败: {e}")
        return False

def clean_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 = 0
    cutoff_time = datetime.now() - timedelta(hours=age_hours)
    
    for pod in pods.items:
        # 检查Pod完成时间
        if pod.status.start_time:
            finish_time = pod.status.start_time
            
            if finish_time < cutoff_time:
                # 删除Pod
                v1.delete_namespaced_pod(
                    name=pod.metadata.name,
                    namespace=namespace,
                    body=kubernetes.client.V1DeleteOptions(
                        propagation_policy='Foreground'
                    )
                )
                print(f"已删除Pod: {pod.metadata.name}")
                deleted_count += 1
    
    print(f"共删除 {deleted_count} 个已完成/失败的Pod")
    return deleted_count

第三步：自动采集Pod日志

def get_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
        )
        
        return logs
    
    except Exception as e:
        print(f"获取Pod日志失败: {e}")
        return ""

def collect_pod_logs(namespace, label_selector, save_dir, tail_lines=1000):
    """批量采集Pod日志"""
    
    import os
    
    v1 = kubernetes.client.CoreV1Api()
    
    # 创建保存目录
    os.makedirs(save_dir, exist_ok=True)
    
    # 获取Pod列表
    pods = v1.list_namespaced_pod(
        namespace=namespace,
        label_selector=label_selector
    )
    
    collected = []
    
    for pod in pods.items:
        pod_name = pod.metadata.name
        
        # 获取所有容器的日志
        for container in pod.spec.containers:
            container_name = container.name
            
            logs = get_pod_logs(namespace, pod_name, container_name, tail_lines)
            
            if logs:
                # 保存到文件
                file_path = os.path.join(save_dir, f"{pod_name}_{container_name}.log")
                
                with open(file_path, "w", encoding="utf-8") as f:
                    f.write(logs)
                
                collected.append({
                    "pod": pod_name,
                    "container": container_name,
                    "log_file": file_path,
                    "size": len(logs)
                })
    
    print(f"共采集 {len(collected)} 个容器的日志")
    return collected

def analyze_logs_for_errors(log_file):
    """分析日志中的错误"""
    
    error_keywords = ["error", "exception", "failed", "fatal", "错误", "异常", "失败"]
    
    errors = []
    
    with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
        for line_num, line in enumerate(f, 1):
            for keyword in error_keywords:
                if keyword in line.lower():
                    errors.append({
                        "line_num": line_num,
                        "keyword": keyword,
                        "content": line.strip()
                    })
                    break
    
    return errors

第四步：监控Pod健康状态

def check_pod_health(namespace=None):
    """检查Pod健康状态"""
    
    v1 = kubernetes.client.CoreV1Api()
    
    if namespace:
        pods = v1.list_namespaced_pod(namespace=namespace)
    else:
        pods = v1.list_pod_for_all_namespaces()
    
    alerts = []
    
    for pod in pods.items:
        pod_name = pod.metadata.name
        namespace = pod.metadata.namespace
        
        # 1. 检查Pod状态
        if pod.status.phase != "Running":
            alerts.append({
                "pod": pod_name,
                "namespace": namespace,
                "issue": f"Pod状态异常: {pod.status.phase}",
                "severity": "high"
            })
        
        # 2. 检查容器重启次数
        if pod.status.container_statuses:
            for container_status in pod.status.container_statuses:
                if container_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)
            
            for container in metrics.containers:
                # CPU使用率（需要自定义阈值）
                cpu_usage = container.usage["cpu"]
                if "n" in cpu_usage:  # 纳秒
                    cpu_value = int(cpu_usage.replace("n", "")) / 1e9  # 转换为核
                    if cpu_value > 0.8:  # 超过0.8核
                        alerts.append({
                            "pod": pod_name,
                            "namespace": namespace,
                            "issue": f"容器 {container.name} CPU使用率过高: {cpu_value:.2f}核",
                            "severity": "medium"
                        })
        
        except Exception:
            # Metrics Server可能未安装
            pass
    
    return alerts

def send_k8s_alert(alerts, webhook_url):
    """发送K8s告警到企微"""
    
    if not alerts:
        return
    
    # 构造消息内容
    content = "⚠️ **Kubernetes Pod健康告警**\n\n"
    
    high_severity = [a for a in alerts if a["severity"] == "high"]
    medium_severity = [a for a in alerts if a["severity"] == "medium"]
    
    if high_severity:
        content += f"**高优先级告警 ({len(high_severity)}项):**\n"
        for alert in high_severity[:5]:  # 最多显示5条
            content += f"- {alert['pod']} ({alert['namespace']}): {alert['issue']}\n"
        content += "\n"
    
    if medium_severity:
        content += f"**中优先级告警 ({len(medium_severity)}项):**\n"
        for alert in medium_severity[:5]:  # 最多显示5条
            content += f"- {alert['pod']} ({alert['namespace']}): {alert['issue']}\n"
    
    # 发送到企微
    payload = {
        "msgtype": "markdown",
        "markdown": {"content": content}
    }
    
    import requests
    response = requests.post(webhook_url, json=payload)
    
    if response.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. 压缩存储：采集后压缩日志文件

def collect_pod_logs_optimized(namespace, pod_name, container_name, save_dir):
    """优化版日志采集（只采集错误日志）"""
    
    logs = get_pod_logs(namespace, pod_name, container_name, tail_lines=10000)
    
    if not logs:
        return None
    
    # 只保留包含错误关键字的行
    error_keywords = ["error", "exception", "failed", "fatal", "错误", "异常", "失败"]
    
    error_lines = []
    for line in logs.split("\n"):
        if any(keyword in line.lower() for keyword in error_keywords):
            error_lines.append(line)
    
    if not error_lines:
        return None
    
    # 保存到文件
    os.makedirs(save_dir, exist_ok=True)
    file_path = os.path.join(save_dir, f"{pod_name}_{container_name}_errors.log")
    
    with open(file_path, "w", encoding="utf-8") as f:
        f.write("\n".join(error_lines))
    
    # 压缩文件
    import gzip
    with open(file_path, "rb") as f_in:
        with gzip.open(f"{file_path}.gz", "wb") as f_out:
            shutil.copyfileobj(f_in, f_out)
    
    os.remove(file_path)  # 删除原始文件
    
    print(f"已采集错误日志并压缩: {file_path}.gz")
    return f"{file_path}.gz"

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

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

解决方案：

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

# 使用Watch机制监听Pod变化（更高效）
from kubernetes import watch

def watch_pod_changes(namespace):
    """监听Pod变化（事件驱动）"""
    
    v1 = kubernetes.client.CoreV1Api()
    w = watch.Watch()
    
    for event in w.stream(v1.list_namespaced_pod, namespace=namespace):
        print(f"事件: {event['type']} Pod: {event['object'].metadata.name}")
        
        # 处理事件（例如：Pod删除时发送告警）
        if event['type'] == "DELETED":
            send_pod_deleted_alert(event['object'])

坑4：跨集群管理复杂

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

解决方案：

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

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

---

总结

功能	节省时间	附加价值
Pod状态监控	每天省30分钟	及时发现故障
日志自动采集	每天省1小时	便于问题排查
自动清理Pod	每周省30分钟	节省集群资源
健康状态告警	—	提高系统稳定性

实际落地建议：

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

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