一、核心技术逻辑（先讲透底层思路，不整虚的）

作为 10 年运维老炮，咱们先把 DevOps 工具集的核心逻辑掰明白 —— 本质是把运维的 “重复手工操作” 抽象成 “可自动化的代码逻辑”，核心围绕 3 个维度：

  1. 数据采集：从 K8S 集群、服务器、日志等源头捞取标准化数据（兼容 K8S 1.33 的 API）；
  2. 逻辑处理：按运维规则过滤、分析、判断（比如 Pod 状态异常、磁盘使用率超 80%）；
  3. 动作执行 / 结果输出：自动修复（重启 Pod）、批量操作（改配置）、可视化展示（巡检报告）。

针对 K8S 1.33，核心适配点：

  • 用官方最新的 client-go（Go）/kubernetes（Python）客户端库，兼容 1.33 的 API 组（比如 apps/v1、core/v1 无变化，但部分 beta API 已转正，需注意）；
  • 支持 K8S 1.33 新增的特性（比如容器运行时接口 CRI v1 稳定版、PodSecurity Admission 默认启用）；
  • 认证方式兼容：kubeconfig 文件、ServiceAccount、Token 认证（1.33 仍主流）。

二、通用操作步骤（所有工具的底层流程，通用于巡检 / 日志 / 批量操作）

不管是做巡检平台、日志脚本还是批量工具，都逃不开这 6 步，咱们用 “说人话” 的方式拆解：

步骤 1：环境准备（先搭好干活的架子）

  • 依赖工具：
    • Go：装 1.21+（兼容 client-go v0.33），拉取k8s.io/client-go@v0.33.0；
    • Python：装 3.9+，装kubernetes==29.0.0（适配 1.33）、pandas（数据处理）、paramiko（批量 SSH）；
    • K8S 侧：确保能通过 kubeconfig（~/.kube/config）或 ServiceAccount 访问集群，权限至少包含：读取 Pod/Node/Deployment、执行 exec/scale 等（按需配置 RBAC）。
  • 核心准备：写一个 “通用 K8S 连接模块”（复用！），不管啥工具都先连集群，避免重复写代码。

步骤 2：需求拆解（把运维需求翻译成代码逻辑）

比如 “集群巡检” 需求：检查 Node 状态、Pod 重启次数、磁盘使用率、容器日志错误 —— 拆解成：

  • 要采集哪些数据？（Node 的 status、Pod 的 restartCount、Node 的 filesystem、容器日志的 ERROR 关键字）；
  • 判定规则是什么？（Node NotReady 告警、Pod 重启 > 3 次告警、磁盘使用率 > 85% 告警、日志 ERROR>10 条告警）；
  • 输出什么结果？（HTML 巡检报告、钉钉告警、本地日志）。

步骤 3：数据采集（捞数据，标准化）

  • K8S 资源采集：通过 client-go/kubernetes 库调用 K8S API，比如列所有 Node、列所有 Namespace 的 Pod；
  • 宿主机数据采集：通过 K8S 的exec进入 Node 的容器（或 SSH），执行df -h、top、dmesg等命令，解析返回结果；
  • 日志采集：通过kubectl logs（代码调用）拉取容器日志，或对接 ELK（如果集群有）。

步骤 4：逻辑处理（核心，按规则判断）

  • 数据清洗：把采集到的原始数据（比如 df -h 的字符串）转成结构化数据（字典 / 结构体）；
  • 规则匹配：写 if/else 或配置化规则（推荐配置化，比如 yaml），判断是否触发告警 / 执行操作；
  • 数据聚合：把分散的数据汇总（比如按 Node 维度汇总所有 Pod 问题）。

步骤 5：动作执行 / 结果输出

  • 被动输出：生成巡检报告（HTML/Excel）、日志文件；
  • 主动操作：批量重启 Pod、扩容 Deployment、发送告警（钉钉 / 企业微信）；
  • 幂等性：确保重复执行不会出问题（比如批量操作前先判断状态）。

步骤 6：测试 & 部署

  • 测试：先在测试集群跑，模拟异常场景（比如手动把 Node 设为 NotReady），验证工具是否能识别；
  • 部署：做成定时任务（crontab）、K8S CronJob，或集成到 CI/CD 平台。

三、详细案例：K8S 1.33 集群巡检工具（Python 版，易理解）

咱们做一个最常用的 “K8S 集群巡检工具”，覆盖核心运维场景，兼容 1.33，代码可直接跑，步骤拆到每一行。

案例目标

  1. 巡检内容：
    • Node 状态（Ready/NotReady）、CPU / 内存使用率；
    • Pod 状态（Running/Failed/CrashLoopBackOff）、重启次数；
    • Deployment 副本数（是否达标）、容器镜像版本；
    • 节点磁盘使用率（/ 根目录）。
  2. 输出：生成 HTML 巡检报告，触发阈值的问题标红；
  3. 兼容 K8S 1.33：用最新的 kubernetes 客户端，适配 API。

步骤 1：环境安装

# 安装依赖
pip install kubernetes==29.0.0 psutil==5.9.8 pandas==2.2.2 jinja2==3.1.3

步骤 2：编写核心代码（分模块，注释拉满）

新建文件k8s_inspector.py，代码分 4 个模块：连接 K8S、数据采集、规则判断、生成报告。

模块 1：通用配置 & K8S 连接（核心复用）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import json
import subprocess
from datetime import datetime
from kubernetes import client, config
from kubernetes.client.rest import ApiException
import pandas as pd
from jinja2 import Template

# ====================== 通用配置（改这里就行，不用动代码） ======================
# 巡检阈值配置（说人话：超过这个值就告警）
THRESHOLDS = {
    "node_disk_usage": 85,          # 节点磁盘使用率>85%告警
    "node_cpu_usage": 80,           # 节点CPU使用率>80%告警
    "node_mem_usage": 85,           # 节点内存使用率>85%告警
    "pod_restart_count": 3,         # Pod重启次数>3次告警
    "deployment_replicas_diff": 0, # Deployment副本数不达标告警（期望-实际≠0）
}

# HTML报告模板（简单易改）
HTML_TEMPLATE = """
<!DOCTYPE html>
<html lang="zh-CN">
<head>
    <meta charset="UTF-8">
    <title>K8S集群巡检报告 {{ report_time }}</title>
    <style>
        .warning { color: red; font-weight: bold; }
        table { border-collapse: collapse; width: 100%; margin: 10px 0; }
        th, td { border: 1px solid #ccc; padding: 8px; text-align: left; }
        th { background-color: #f2f2f2; }
    </style>
</head>
<body>
    <h1>K8S 1.33集群巡检报告</h1>
    <p>巡检时间：{{ report_time }}</p>
    <p>集群名称：{{ cluster_name }}</p>

    <h2>1. 节点状态巡检</h2>
    <table>
        <tr><th>节点名称</th><th>状态</th><th>CPU使用率(%)</th><th>内存使用率(%)</th><th>磁盘使用率(%)</th><th>问题</th></tr>
        {% for node in node_data %}
        <tr>
            <td>{{ node.name }}</td>
            <td {% if node.status != "Ready" %}class="warning"{% endif %}>{{ node.status }}</td>
            <td {% if node.cpu_usage > thresholds.node_cpu_usage %}class="warning"{% endif %}>{{ node.cpu_usage }}</td>
            <td {% if node.mem_usage > thresholds.node_mem_usage %}class="warning"{% endif %}>{{ node.mem_usage }}</td>
            <td {% if node.disk_usage > thresholds.node_disk_usage %}class="warning"{% endif %}>{{ node.disk_usage }}</td>
            <td {% if node.issues %}class="warning"{% endif %}>{{ node.issues or "无" }}</td>
        </tr>
        {% endfor %}
    </table>

    <h2>2. Pod状态巡检</h2>
    <table>
        <tr><th>命名空间</th><th>Pod名称</th><th>状态</th><th>重启次数</th><th>所在节点</th><th>问题</th></tr>
        {% for pod in pod_data %}
        <tr>
            <td>{{ pod.namespace }}</td>
            <td>{{ pod.name }}</td>
            <td {% if pod.status not in ["Running", "Completed"] %}class="warning"{% endif %}>{{ pod.status }}</td>
            <td {% if pod.restart_count > thresholds.pod_restart_count %}class="warning"{% endif %}>{{ pod.restart_count }}</td>
            <td>{{ pod.node_name }}</td>
            <td {% if pod.issues %}class="warning"{% endif %}>{{ pod.issues or "无" }}</td>
        </tr>
        {% endfor %}
    </table>

    <h2>3. Deployment巡检</h2>
    <table>
        <tr><th>命名空间</th><th>Deployment名称</th><th>期望副本数</th><th>可用副本数</th><th>镜像版本</th><th>问题</th></tr>
        {% for deploy in deploy_data %}
        <tr>
            <td>{{ deploy.namespace }}</td>
            <td>{{ deploy.name }}</td>
            <td>{{ deploy.replicas_expected }}</td>
            <td {% if deploy.replicas_available != deploy.replicas_expected %}class="warning"{% endif %}>{{ deploy.replicas_available }}</td>
            <td>{{ deploy.image }}</td>
            <td {% if deploy.issues %}class="warning"{% endif %}>{{ deploy.issues or "无" }}</td>
        </tr>
        {% endfor %}
    </table>
</body>
</html>
"""

# ====================== 连接K8S集群（兼容1.33） ======================
def connect_k8s():
    """
    连接K8S集群，优先用kubeconfig（本地/运维机），其次用ServiceAccount（集群内运行）
    兼容K8S 1.33：用最新的client配置
    """
    try:
        # 本地运维机：用~/.kube/config
        config.load_kube_config()
        print("✅ 成功通过kubeconfig连接K8S集群")
    except:
        # 集群内Pod运行：用ServiceAccount（/var/run/secrets/kubernetes.io/serviceaccount）
        config.load_incluster_config()
        print("✅ 成功通过ServiceAccount连接K8S集群")
    
    # 初始化客户端（兼容1.33的核心：用对应版本的API）
    core_api = client.CoreV1Api()
    apps_api = client.AppsV1Api()
    return core_api, apps_api

模块 2：数据采集（捞取 K8S 和节点数据）

# ====================== 数据采集模块 ======================
def get_node_metrics(node_name):
    """
    获取节点CPU/内存/磁盘使用率（兼容K8S 1.33：通过exec执行宿主机命令）
    思路：K8S 1.33中，Node是通过CRI管理的，直接exec到节点的pause容器（或用node-exporter，这里简化）
    """
    core_api, _ = connect_k8s()
    metrics = {
        "cpu_usage": 0.0,
        "mem_usage": 0.0,
        "disk_usage": 0.0
    }
    
    try:
        # 1. 执行top获取CPU使用率（简化版：取1分钟平均）
        cpu_cmd = "top -bn1 | grep 'Cpu(s)' | sed 's/.*, *\\([0-9.]*\\)%* id.*/\\1/' | awk '{print 100 - $1}'"
        cpu_result = core_api.connect_get_namespaced_pod_exec(
            name="dummy-pod",  # 实际用节点上的任意运行中Pod，或提前创建一个
            namespace="default",
            command=["sh", "-c", cpu_cmd],
            stderr=True, stdin=False, stdout=True, tty=False
        )
        metrics["cpu_usage"] = round(float(cpu_result.strip()), 2)
        
        # 2. 执行free获取内存使用率
        mem_cmd = "free | grep Mem | awk '{print $3/$2 * 100.0}'"
        mem_result = core_api.connect_get_namespaced_pod_exec(
            name="dummy-pod", namespace="default", command=["sh", "-c", mem_cmd],
            stderr=True, stdin=False, stdout=True, tty=False
        )
        metrics["mem_usage"] = round(float(mem_result.strip()), 2)
        
        # 3. 执行df获取根目录磁盘使用率
        disk_cmd = "df -h / | grep / | awk '{print $5}' | sed 's/%//g'"
        disk_result = core_api.connect_get_namespaced_pod_exec(
            name="dummy-pod", namespace="default", command=["sh", "-c", disk_cmd],
            stderr=True, stdin=False, stdout=True, tty=False
        )
        metrics["disk_usage"] = round(float(disk_result.strip()), 2)
        
    except Exception as e:
        print(f"⚠️ 获取节点{node_name}指标失败：{str(e)}")
    
    return metrics

def collect_node_data():
    """采集所有Node的基础信息+指标"""
    core_api, _ = connect_k8s()
    node_list = core_api.list_node().items
    node_data = []
    
    for node in node_list:
        # 提取Node状态（Ready/NotReady）
        node_status = "Unknown"
        for condition in node.status.conditions:
            if condition.type == "Ready":
                node_status = "Ready" if condition.status == "True" else "NotReady"
                break
        
        # 获取节点指标（CPU/内存/磁盘）
        metrics = get_node_metrics(node.metadata.name)
        
        # 汇总节点数据
        node_info = {
            "name": node.metadata.name,
            "status": node_status,
            "cpu_usage": metrics["cpu_usage"],
            "mem_usage": metrics["mem_usage"],
            "disk_usage": metrics["disk_usage"],
            "issues": ""
        }
        
        # 规则判断：节点是否有问题
        issues = []
        if node_status != "Ready":
            issues.append("节点状态异常（NotReady）")
        if metrics["cpu_usage"] > THRESHOLDS["node_cpu_usage"]:
            issues.append(f"CPU使用率超标（{metrics['cpu_usage']}% > {THRESHOLDS['node_cpu_usage']}%）")
        if metrics["mem_usage"] > THRESHOLDS["node_mem_usage"]:
            issues.append(f"内存使用率超标（{metrics['mem_usage']}% > {THRESHOLDS['node_mem_usage']}%）")
        if metrics["disk_usage"] > THRESHOLDS["node_disk_usage"]:
            issues.append(f"磁盘使用率超标（{metrics['disk_usage']}% > {THRESHOLDS['node_disk_usage']}%）")
        
        node_info["issues"] = "; ".join(issues) if issues else ""
        node_data.append(node_info)
    
    return node_data

def collect_pod_data():
    """采集所有Pod的基础信息"""
    core_api, _ = connect_k8s()
    pod_list = core_api.list_pod_for_all_namespaces().items
    pod_data = []
    
    for pod in pod_list:
        # 提取Pod核心信息
        pod_info = {
            "namespace": pod.metadata.namespace,
            "name": pod.metadata.name,
            "status": pod.status.phase,
            "restart_count": 0,
            "node_name": pod.spec.node_name,
            "issues": ""
        }
        
        # 计算Pod重启次数（所有容器的重启次数之和）
        for container_status in pod.status.container_statuses or []:
            pod_info["restart_count"] += container_status.restart_count or 0
        
        # 规则判断：Pod是否有问题
        issues = []
        if pod_info["status"] not in ["Running", "Completed"]:
            issues.append(f"状态异常（{pod_info['status']}）")
        if pod_info["restart_count"] > THRESHOLDS["pod_restart_count"]:
            issues.append(f"重启次数超标（{pod_info['restart_count']} > {THRESHOLDS['pod_restart_count']}）")
        
        pod_info["issues"] = "; ".join(issues) if issues else ""
        pod_data.append(pod_info)
    
    return pod_data

def collect_deploy_data():
    """采集所有Deployment的基础信息"""
    _, apps_api = connect_k8s()
    deploy_list = apps_api.list_deployment_for_all_namespaces().items
    deploy_data = []
    
    for deploy in deploy_list:
        # 提取Deployment核心信息
        deploy_info = {
            "namespace": deploy.metadata.namespace,
            "name": deploy.metadata.name,
            "replicas_expected": deploy.spec.replicas or 0,
            "replicas_available": deploy.status.available_replicas or 0,
            "image": deploy.spec.template.spec.containers[0].image if deploy.spec.template.spec.containers else "",
            "issues": ""
        }
        
        # 规则判断：Deployment是否有问题
        issues = []
        if deploy_info["replicas_available"] != deploy_info["replicas_expected"]:
            issues.append(f"副本数不达标（期望{deploy_info['replicas_expected']}，可用{deploy_info['replicas_available']}）")
        
        deploy_info["issues"] = "; ".join(issues) if issues else ""
        deploy_data.append(deploy_info)
    
    return deploy_data

模块 3：生成巡检报告

# ====================== 生成报告模块 ======================
def generate_report():
    """生成HTML巡检报告"""
    # 1. 采集所有数据
    node_data = collect_node_data()
    pod_data = collect_pod_data()
    deploy_data = collect_deploy_data()
    
    # 2. 填充模板
    template = Template(HTML_TEMPLATE)
    report_content = template.render(
        report_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        cluster_name="K8S 1.33生产集群",  # 可从kubeconfig中提取
        thresholds=THRESHOLDS,
        node_data=node_data,
        pod_data=pod_data,
        deploy_data=deploy_data
    )
    
    # 3. 保存报告
    report_path = f"k8s_inspect_report_{datetime.now().strftime('%Y%m%d%H%M%S')}.html"
    with open(report_path, "w", encoding="utf-8") as f:
        f.write(report_content)
    
    print(f"✅ 巡检报告已生成：{os.path.abspath(report_path)}")
    return report_path

# ====================== 主函数 ======================
if __name__ == "__main__":
    try:
        print("🚀 开始K8S 1.33集群巡检...")
        generate_report()
        print("🎉 巡检完成！")
    except Exception as e:
        print(f"❌ 巡检失败：{str(e)}")
        exit(1)

步骤 3：代码调整 & 测试（关键！适配你的集群）

  1. 修改 dummy-pod：代码中get_node_metrics里的dummy-pod要换成你集群中存在的 Pod（任意运行中的 Pod，比如 default 命名空间的 nginx-pod）；
  2. 权限配置：确保运行代码的账号有以下权限（K8S RBAC）： 
    # 新建inspector-role.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: k8s-inspector-role
rules:
- apiGroups: [""]
  resources: ["nodes", "pods", "pods/exec"]
  verbs: ["get", "list"]
- apiGroups: ["apps"]
  resources: ["deployments"]
  verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: k8s-inspector-binding
subjects:
- kind: User
  name: admin  # 换成你的用户名/ServiceAccount
  apiGroup: rbac.authorization.k8s.io
roleRef:
  kind: ClusterRole
  name: k8s-inspector-role
  apiGroup: rbac.authorization.k8s.io
  3. 测试运行： 
    # 本地运行（有kubeconfig）
python k8s_inspector.py
# 集群内运行（挂载ServiceAccount）
kubectl run k8s-inspector --image=python:3.9 -it --rm --restart=Never -- python /app/k8s_inspector.py

步骤 4：扩展优化（运维老炮必备）

  1. 配置化：把阈值、巡检项写到 yaml 文件，不用改代码；
  2. 告警集成：检测到问题后，调用钉钉 / 企业微信 API 发送告警；
  3. 定时执行：做成 K8S CronJob，每天凌晨巡检： 
    apiVersion: batch/v1
kind: CronJob
metadata:
  name: k8s-inspector-cron
spec:
  schedule: "0 0 * * *"  # 每天0点
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: k8s-inspector-sa  # 绑定上面的Role
          containers:
          - name: k8s-inspector
            image: python:3.9
            volumeMounts:
            - name: script
              mountPath: /app
            command: ["python", "/app/k8s_inspector.py"]
          volumes:
          - name: script
            configMap:
              name: k8s-inspector-script  # 把代码放到ConfigMap
          restartPolicy: OnFailure
  4. 日志分析集成：在采集 Pod 数据时，拉取最近 1 小时的日志，统计 ERROR 关键字数量，加到巡检报告。

四、关键注意事项（避坑指南）

  1. K8S 1.33 兼容性：
    • 避免使用已废弃的 API（比如extensions/v1beta1，1.33 已完全移除）；
    • client-go/kubernetes 库版本必须和 K8S 版本匹配（1.33 对应 client v0.33）；
    • PodSecurity Admission：如果集群启用了 PSA，确保工具的 ServiceAccount 有对应的权限（比如 privileged 级别）。
  2. 性能问题：巡检时如果集群 Pod 多（上千个），要分页查询（limit/continue），避免一次性拉取所有数据导致超时；
  3. 幂等性：批量操作工具（比如重启 Pod）要加判断，比如只重启状态为 Failed 的 Pod，避免重复重启；
  4. 权限最小化：工具的 ServiceAccount 只给必要权限，不要用 cluster-admin。

五、总结

DevOps 工具集的核心就是 “把运维手工活写成代码”，先抽象通用逻辑（连接、采集、判断、输出），再针对具体场景（巡检 / 日志 / 批量）填充细节。这个案例覆盖了 K8S 1.33 的核心巡检场景，代码可直接复用，你可以基于这个框架扩展成自己的工具集（比如批量操作工具：改一行代码，把 “生成报告” 换成 “批量重启 Pod” 即可）。

# golang # python # devops
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

别让你的 Agent 一口吃成胖子：AI Agent 任务拆分的工程实践指南

avatar AI Agent技术社区
cover

AI Agent律师行业智能体的实践步骤

avatar AI Agent技术社区

MCP + 企业IM架构：AI Agent调用OA、ERP、MES时，身份和权限如何传递？

<think>我们只需要根据内容生成摘要，不超过150字。先理解文章核心：MCP+企业IM架构，AI Agent调用OA/ERP/MES时身份和权限传递。要点：IM从消息入口变任务入口；MCP解决连接，但不解决权限；需要五层身份映射；权限分层（可见性、调用、数据范围、确认）；建议组合MCP+API网关。摘要需简洁。 参考生成： "本文探讨MCP与企业IM集成中AI Agent调用业务系统时的身份权

avatar AI Agent技术社区