如何在Python中获取Shell命令执行输出?kubectl报错捕获问题
如何捕获kubectl命令的错误输出?
我明白你的问题了——你用Python的subprocess串联执行kubectl和grep命令,但kubectl产生的错误信息完全没被捕获,只得到空的out_ns和err。这是因为你没处理第一个进程(kubectl)的错误输出通道。
问题根源
你当前的代码只重定向了kubectl的stdout到管道,但是它的stderr(错误输出)默认是直接输出到终端的,根本没进入后续的grep进程或你的communicate()捕获流程。你看到的终端错误,是kubectl直接打印的,不是通过你的Python代码输出的。
解决方案
这里有两种实用的处理方式,你可以根据需求选择:
方法1:分开处理kubectl和grep的输出与错误
这种方式更清晰,能分别捕获kubectl本身的错误,再处理正常输出的过滤:
import subprocess NAMESPACE = "namespace-tests" # 先执行kubectl,同时捕获它的标准输出和错误输出 kubectl_proc = subprocess.Popen( ['kubectl', 'get', 'ns'], stdout=subprocess.PIPE, stderr=subprocess.PIPE ) # 获取kubectl的执行结果和错误 kubectl_out, kubectl_err = kubectl_proc.communicate() # 先检查kubectl是否执行出错 if kubectl_proc.returncode != 0: print(f"kubectl执行错误: {kubectl_err.decode('utf-8').strip()}") else: # kubectl执行正常,再用grep过滤结果 grep_proc = subprocess.Popen( ["grep", "-E", r'(^|\s)' + NAMESPACE + r'($|\s)'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) grep_out, grep_err = grep_proc.communicate(input=kubectl_out) print(f"过滤结果: {grep_out.decode('utf-8').strip()}") print(f"grep执行错误: {grep_err.decode('utf-8').strip()}")
方法2:将kubectl的错误合并到标准输出
如果你想把kubectl的错误也传入管道(比如让grep也能处理错误信息),可以把stderr合并到stdout:
import subprocess NAMESPACE = "namespace-tests" # 将kubectl的错误输出合并到标准输出 kubectl_proc = subprocess.Popen( ['kubectl', 'get', 'ns'], stdout=subprocess.PIPE, stderr=subprocess.STDOUT # 关键:把stderr重定向到stdout ) grep_proc = subprocess.Popen( ["grep", "-E", r'(^|\s)' + NAMESPACE + r'($|\s)'], stdin=kubectl_proc.stdout, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) kubectl_proc.stdout.close() combined_out, grep_err = grep_proc.communicate() print(f"输出(含kubectl错误): {combined_out.decode('utf-8').strip()}") print(f"grep执行错误: {grep_err.decode('utf-8').strip()}")
额外小提示
- 使用原始字符串(
r'')来写正则表达式,避免转义字符的混乱,比如r'(^|\s)'比"\(^|\s)"更可靠。 - 记得用
.decode('utf-8')把字节流转换成字符串,否则直接打印会显示b'xxx'的字节格式。
内容的提问来源于stack exchange,提问作者BhagyaKolitha Jayalath
相关产品推荐
相关产品推荐

