三国曹魏古城建设网站郑州seo技术

前言：公司云原生k8s二开工程师发了一串通用性命令用来查询以下数值，我想着能不能将这命令写成一个自动化脚本。
起初设计的 版本一：开头加一条环境变量，执行脚本后，提示输入：需要查询的命名空间，输出信息追加到以当前年月日时来命名自动生成的txt文件；
版本二：自动生成中文排头标题，并生成csv文件，这样就不用手动将txt转化成excl表了；
版本三：发现生成csv文件，其中 副本数 和 容器镜像信息区分不开，想着用在文本格式处理更有优势的python来写，最终成功了。

需要注意的是我们这边k8s容器平台是二开版，然后由于我目前试验的命名空间业务的特殊性，标题里通用的 就绪探针 和 存活探针，采集的参数为：资源限制 CPU 和 资源限制 Memory的，儿你使用时，具体的参数需要根据你的当前deployments.apps配置参数来判定。

查询数值：

namespaceName：部署所在的命名空间
deploymentName：部署的名称
replicas：部署的副本数量
image：容器的镜像
resourcesRequest：容器请求的资源
resourcesLimits：容器资源的限制
readinessProbe：就绪探针的配置
livenessProbe：存活探针的配置
skyworkingNamespace：环境变量 SW_AGENT_NAMESPACE 的值
lopLogsApplog：环境变量 lop_logs_applog 的值

通用命令：
kubectl get deployments.apps -n 需要查询的命名空间 -o jsonpath='{range .items[*]} {"\n\n"} namespaceName={.metadata.namespace}{"\t"} deploymentName={.metadata.name} {"\t"} replicas={.spec.replicas} {range .spec.template.spec.containers[*]} image={.image} {"\t"} resourcesRequest={.resources.requests} {"\t"} resourcesLimits={.resources.limits} {"\t"} readinessProbe={.readinessProbe} {"\t"} livenessProbe={.livenessProbe} {"\t"}{end} skyworkingNamespace={.spec.template.spec.containers[0].env[?(@.name=="SW_AGENT_NAMESPACE")].*} {"\t"} lopLogsApplog={.spec.template.spec.containers[0].env[?(@.name=="lop_logs_applog")].*} {end} {"\n"}'

版本一：shell脚本

#!/bin/bash# 设置环境变量
export ENV_VARIABLE="SomeValue"read -p "请输入需要查询的命名空间： " namespace# 获取日期和时间并格式化作为文件名
current_date=$(date +%Y%m%d%H)
filename="${current_date}.txt"# 执行 kubectl 命令并将结果追加到文件中
kubectl get deployments.apps -n "$namespace" -o jsonpath='{range.items[*]} {"\n\n"} namespaceName={.metadata.namespace}{"\t"} deploymentName={.metadata.name} {"\t"} replicas={.spec.replicas} {range.spec.template.spec.containers[*]} image={.image} {"\t"} resourcesRequest={.resources.requests} {"\t"} resourcesLimits={.resources.limits} {"\t"} readinessProbe={.readinessProbe} {"\t"} livenessProbe={.livenessProbe} {"\t"}{end} skyworkingNamespace={.spec.template.spec.containers[0].env[?(@.name=="SW_AGENT_NAMESPACE")].*} {"\t"} lopLogsApplog={.spec.template.spec.containers[0].env[?(@.name=="lop_logs_applog")].*} {end} {"\n"}' >> "$filename"echo "输出已追加到 $filename 文件中。"

执行结果：

版本二：shell脚本：

#!/bin/bashexport ENV_VARIABLE="SomeValue"read -p "请输入需要查询的命名空间： " namespacecurrent_date=$(date +%Y%m%d%H)
filename="${current_date}.csv"# 使用以下命令生成 CSV 格式的输出并追加到文件中
kubectl get deployments.apps -n "$namespace" -o jsonpath='{range.items[*]}{"\n"}'$'\t'"{.metadata.namespace}"$'\t'"{.metadata.name}"$'\t'"{.spec.replicas}"$'\t'"{range.spec.template.spec.containers[*]}{.image}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.resources.requests}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.resources.limits}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.readinessProbe}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.livenessProbe}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.spec.template.spec.containers[0].env[?(@.name==\"SW_AGENT_NAMESPACE\")].value}{end}"$'\t'"{range.spec.template.spec.containers[*]}{.spec.template.spec.containers[0].env[?(@.name==\"lop_logs_applog\")].value}{end}" >> "$filename"echo "输出已追加到 $filename 文件中。"

执行结果：
版本三：
改了很多版，这里就直接展示成功吧

通用版-py脚本：

import subprocess
import datetime# 获取用户输入的命名空间
namespace = input("请输入当前需要查询的命名空间： ")# 获取当前时间并生成文件名
timestamp = datetime.datetime.now().strftime("%Y%m%d%H")
output_file = f"{timestamp}.csv"# 设置标题行
with open(output_file, 'w') as f:f.write("命名空间,名称,副本数,容器镜像,资源请求,资源限制,就绪探针,存活探针,环境变量\n")# 执行 kubectl 命令并处理结果
command = f"kubectl get deployments.apps -n {namespace} -o json".split()
result = subprocess.check_output(command).decode()import json
data = json.loads(result)def format_probe(probe):if not probe:return "N/A"probe_type = ""details = ""if "httpGet" in probe:probe_type = "HTTP GET"http_get = probe["httpGet"]details = f"{probe_type}: Path: {http_get.get('path', 'N/A')}, Port: {http_get.get('port', 'N/A')}"elif "tcpSocket" in probe:probe_type = "TCP Socket"tcp_socket = probe["tcpSocket"]details = f"{probe_type}: Port: {tcp_socket.get('port', 'N/A')}"elif "exec" in probe:probe_type = "Exec"exec_command = probe["exec"]details = f"{probe_type}: Command: {' '.join(exec_command)}"elif "cpu" in probe and "memory" in probe:details = f"Resource: CPU {probe['cpu']}, Memory {probe['memory']}"else:return "N/A"return detailsdef format_resources(resources):if not resources:return "N/A"cpu = resources.get('cpu', 'N/A')memory = resources.get('memory', 'N/A')return f"CPU: {cpu}, Memory: {memory}"with open(output_file, 'a') as f:for item in data.get('items', []):namespace_name = item['metadata']['namespace']deployment_name = item['metadata']['name']replicas = item['spec']['replicas']images = [container['image'] for container in item['spec']['template']['spec']['containers']]image = images[0] if images else "N/A"request = format_resources(item['spec']['template']['spec']['containers'][0]['resources']['requests'])limit = format_resources(item['spec']['template']['spec']['containers'][0]['resources']['limits'])readiness_probe = "N/A"if item['spec']['template']['spec']['containers'] and 0 < len(item['spec']['template']['spec']['containers']) and 'readinessProbe' in item['spec']['template']['spec']['containers'][0]:readiness_probe = format_probe(item['spec']['template']['spec']['containers'][0]['readinessProbe'])liveness_probe = "N/A"if item['spec']['template']['spec']['containers'] and 0 < len(item['spec']['template']['spec']['containers']) and 'livenessProbe' in item['spec']['template']['spec']['containers'][0]:liveness_probe = format_probe(item['spec']['template']['spec']['containers'][0]['livenessProbe'])env_var = "N/A"if item['spec']['template']['spec']['containers'] and 'env' in item['spec']['template']['spec']['containers'][0]:env_var = next((env['value'] for env in item['spec']['template']['spec']['containers'][0]['env'] if env['name'] == "SW_AGENT_NAMESPACE"), "N/A")f.write(f"{namespace_name},{deployment_name},{replicas},{image},{request},{limit},{readiness_probe},{liveness_probe},{env_var}\n")print(f"查询结果已写入文件：{output_file}")

定制版-py脚本：

import subprocess
import datetime# 获取用户输入的命名空间
namespace = input("请输入当前需要查询的命名空间： ")# 获取当前时间并生成文件名
timestamp = datetime.datetime.now().strftime("%Y%m%d%H")
output_file = f"{timestamp}.csv"# 设置标题行
with open(output_file, 'w') as f:f.write("命名空间,名称,副本数,容器镜像,资源预留 CPU,资源预留 Memory,资源限制 CPU,资源限制 Memory,环境变量\n")# 执行 kubectl 命令并处理结果
command = f"kubectl get deployments.apps -n {namespace} -o json".split()
result = subprocess.check_output(command).decode()import json
data = json.loads(result)def format_probe(probe):if not probe:return "N/A"probe_type = ""details = ""if "httpGet" in probe:probe_type = "HTTP GET"http_get = probe["httpGet"]details = f"{probe_type}: Path: {http_get.get('path', 'N/A')}, Port: {http_get.get('port', 'N/A')}"elif "tcpSocket" in probe:probe_type = "TCP Socket"tcp_socket = probe["tcpSocket"]details = f"{probe_type}: Port: {tcp_socket.get('port', 'N/A')}"elif "exec" in probe:probe_type = "Exec"exec_command = probe["exec"]details = f"{probe_type}: Command: {' '.join(exec_command)}"elif "cpu" in probe and "memory" in probe:details = f"Resource: CPU {probe['cpu']}, Memory {probe['memory']}"else:return "N/A"return detailsdef format_resources(resources):if not resources:return "N/A"cpu = resources.get('cpu', 'N/A')memory = resources.get('memory', 'N/A')return f"CPU: {cpu}, Memory: {memory}"with open(output_file, 'a') as f:for item in data.get('items', []):namespace_name = item['metadata']['namespace']deployment_name = item['metadata']['name']replicas = item['spec']['replicas']images = [container['image'] for container in item['spec']['template']['spec']['containers']]image = images[0] if images else "N/A"request = format_resources(item['spec']['template']['spec']['containers'][0]['resources']['requests'])limit = format_resources(item['spec']['template']['spec']['containers'][0]['resources']['limits'])readiness_probe = "N/A"if item['spec']['template']['spec']['containers'] and 0 < len(item['spec']['template']['spec']['containers']) and 'readinessProbe' in item['spec']['template']['spec']['containers'][0]:readiness_probe = format_probe(item['spec']['template']['spec']['containers'][0]['readinessProbe'])liveness_probe = "N/A"if item['spec']['template']['spec']['containers'] and 0 < len(item['spec']['template']['spec']['containers']) and 'livenessProbe' in item['spec']['template']['spec']['containers'][0]:liveness_probe = format_probe(item['spec']['template']['spec']['containers'][0]['livenessProbe'])env_var = "N/A"if item['spec']['template']['spec']['containers'] and 'env' in item['spec']['template']['spec']['containers'][0]:env_var = next((env['value'] for env in item['spec']['template']['spec']['containers'][0]['env'] if env['name'] == "SW_AGENT_NAMESPACE"), "N/A")# 按照新的标题格式写入数据parts_request = request.split(", ") if request!= "N/A" else ["N/A", "N/A"]parts_limit = limit.split(", ") if limit!= "N/A" else ["N/A", "N/A"]f.write(f"{namespace_name},{deployment_name},{replicas},{image},{parts_request[0].replace('CPU: ', '')},{parts_request[1].replace('Memory: ', '')},{parts_limit[0].replace('CPU: ', '')},{parts_limit[1].replace('Memory: ', '')},{env_var}\n")print(f"查询结果已写入文件：{output_file}")

执行：

执行结果：

自动化脚本和输出结果递交上去后，受到公司 高级系统架构师（兼技术专家组云原生高级架构师）、云原生k8s二开工程师以及其他同事和领导的点赞。