执行 Linux 命令（subprocess）

运维自动化离不开调用系统命令：kubectl get pods、docker ps、systemctl is-active nginx。标准库 subprocess 是首选；关键是用 list 传参、设置 timeout、避免随意 shell=True。

---

基本模式：list 传参

#!/usr/bin/env python3
"""安全执行外部命令：list 形式、捕获输出、超时控制。"""
import subprocess
import sys

def run_cmd(cmd: list[str], timeout: int = 30) -> subprocess.CompletedProcess:
    """
    cmd 示例: ["kubectl", "get", "pods", "-n", "default", "-o", "wide"]
    切勿: f"kubectl get pods -n {ns}" + shell=True  # 有注入风险
    """
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout,
            check=False,  # 非零退出码不抛异常，由调用方判断
        )
        return result
    except subprocess.TimeoutExpired as e:
        print(f"命令超时 ({timeout}s): {' '.join(cmd)}", file=sys.stderr)
        raise

# 示例：检查 nginx 服务状态
r = run_cmd(["systemctl", "is-active", "nginx"], timeout=10)
if r.returncode == 0 and r.stdout.strip() == "active":
    print("nginx 运行中")
else:
    print(f"nginx 异常: rc={r.returncode}, stderr={r.stderr.strip()}")

---

运维命令封装

import json
import subprocess

def kubectl_pods(namespace: str = "default") -> list[dict]:
    """获取 Pod 列表，JSON 输出便于解析。"""
    cmd = [
        "kubectl", "get", "pods",
        "-n", namespace,
        "-o", "json",
    ]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
    if r.returncode != 0:
        raise RuntimeError(f"kubectl 失败: {r.stderr}")
    data = json.loads(r.stdout)
    return data.get("items", [])

def docker_container_names() -> list[str]:
    cmd = ["docker", "ps", "--format", "{{.Names}}"]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    r.check_returncode()
    return [line.strip() for line in r.stdout.splitlines() if line.strip()]

def systemctl_status(unit: str) -> dict:
    cmd = ["systemctl", "show", unit, "--property=ActiveState,SubState"]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
    props = {}
    for line in r.stdout.splitlines():
        if "=" in line:
            k, v = line.split("=", 1)
            props[k] = v
    return props

---

何时可以用 shell=True（谨慎）

仅在固定管道、无用户输入且命令本身无法用 list 表达时考虑，并限制 shell：

# 统计 ERROR 行数：管道在 shell 里较自然，但 ns/pod 必须来自可信变量
namespace = "kube-system"
cmd = f"kubectl logs -n {namespace} -l app=coredns --tail=500 2>/dev/null | grep -c ERROR || true"
# 更推荐：分两步，kubectl 用 list，grep 用 list 或 Python 字符串处理

禁止：把用户输入、Web 参数直接拼进 shell=True 字符串。

---

错误处理 checklist

项	说明
timeout	防止 kubectl/docker pull 挂死
check / returncode	区分成功与业务失败
stderr	失败时打印完整 stderr 便于排障
绝对路径	定时任务里用 /usr/bin/kubectl 避免 PATH 不一致
环境变量	env={**os.environ, "KUBECONFIG": "/path/to/config"}

import os
import subprocess

r = subprocess.run(
    ["/usr/bin/kubectl", "cluster-info"],
    capture_output=True,
    text=True,
    timeout=30,
    env={**os.environ, "KUBECONFIG": os.path.expanduser("~/.kube/config")},
)
if r.returncode != 0:
    raise SystemExit(f"kubectl cluster-info 失败:\n{r.stderr}")
print(r.stdout)

注意

生产脚本不要 shell=True + 外部输入。复杂逻辑用 Python 解析 JSON/YAML，比解析 shell 文本更稳。