前言：公司云原生k8s二開工程師發(fā)了一串通用性命令用來查詢以下數(shù)值，我想著能不能將這命令寫成一個自動化腳本。
起初設計的 版本一：開頭加一條環(huán)境變量，執(zhí)行腳本后，提示輸入：需要查詢的命名空間，輸出信息追加到以當前年月日時來命名自動生成的txt文件；
版本二：自動生成中文排頭標題，并生成csv文件，這樣就不用手動將txt轉化成excl表了；
版本三：發(fā)現(xiàn)生成csv文件，其中 副本數(shù) 和 容器鏡像信息區(qū)分不開，想著用在文本格式處理更有優(yōu)勢的python來寫，最終成功了。

需要注意的是我們這邊k8s容器平臺是二開版，然后由于我目前試驗的命名空間業(yè)務的特殊性，標題里通用的 就緒探針 和 存活探針，采集的參數(shù)為：資源限制 CPU 和 資源限制 Memory的，兒你使用時，具體的參數(shù)需要根據(jù)你的當前deployments.apps配置參數(shù)來判定。

查詢數(shù)值：

namespaceName：部署所在的命名空間
deploymentName：部署的名稱
replicas：部署的副本數(shù)量
image：容器的鏡像
resourcesRequest：容器請求的資源
resourcesLimits：容器資源的限制
readinessProbe：就緒探針的配置
livenessProbe：存活探針的配置
skyworkingNamespace：環(huán)境變量 SW_AGENT_NAMESPACE 的值
lopLogsApplog：環(huán)境變量 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# 設置環(huán)境變量
export ENV_VARIABLE="SomeValue"read -p "請輸入需要查詢的命名空間： " namespace# 獲取日期和時間并格式化作為文件名
current_date=$(date +%Y%m%d%H)
filename="${current_date}.txt"# 執(zhí)行 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 文件中。"

執(zhí)行結果：

版本二：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 文件中。"

執(zhí)行結果：
版本三：
改了很多版，這里就直接展示成功吧

通用版-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("命名空間,名稱,副本數(shù),容器鏡像,資源請求,資源限制,就緒探針,存活探針,環(huán)境變量\n")# 執(zhí)行 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("命名空間,名稱,副本數(shù),容器鏡像,資源預留 CPU,資源預留 Memory,資源限制 CPU,資源限制 Memory,環(huán)境變量\n")# 執(zhí)行 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")# 按照新的標題格式寫入數(shù)據(jù)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}")

執(zhí)行：

執(zhí)行結果：

自動化腳本和輸出結果遞交上去后，受到公司 高級系統(tǒng)架構師（兼技術專家組云原生高級架構師）、云原生k8s二開工程師以及其他同事和領導的點贊。