237 lines
9.3 KiB
Python
237 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Author: Logan.Li
|
||
Gitee: https://gitee.com/attacker
|
||
email: admin@attacker.club
|
||
Date: 2025-01-05 12:25:52
|
||
LastEditTime: 2025-04-09
|
||
Description: 导出Kubernetes Deployment+Service和Ingress资源
|
||
"""
|
||
|
||
import os
|
||
import yaml
|
||
import subprocess
|
||
import sys
|
||
|
||
# 尝试导入kubernetes模块，如果失败则尝试安装
|
||
try:
|
||
from kubernetes import client, config
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
except ImportError:
|
||
print("kubernetes模块未安装，尝试安装...")
|
||
subprocess.check_call([sys.executable, "-m", "pip", "install", "kubernetes"])
|
||
from kubernetes import client, config
|
||
import urllib3
|
||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||
|
||
def sanitize_resource(obj, resource_type):
|
||
"""清理资源对象，移除不必要的字段"""
|
||
# 获取序列化后的对象
|
||
obj_yaml = api_client.sanitize_for_serialization(obj)
|
||
|
||
# 确保正确的apiVersion和kind在最前面
|
||
if isinstance(obj, client.V1Service):
|
||
obj_yaml['apiVersion'] = 'v1'
|
||
obj_yaml['kind'] = 'Service'
|
||
elif isinstance(obj, client.V1Deployment):
|
||
obj_yaml['apiVersion'] = 'apps/v1'
|
||
obj_yaml['kind'] = 'Deployment'
|
||
elif isinstance(obj, client.V1Ingress):
|
||
obj_yaml['apiVersion'] = 'networking.k8s.io/v1'
|
||
obj_yaml['kind'] = 'Ingress'
|
||
|
||
# 重新组织字段顺序，确保apiVersion和kind在最前面
|
||
ordered_obj = {
|
||
'apiVersion': obj_yaml.pop('apiVersion'),
|
||
'kind': obj_yaml.pop('kind'),
|
||
'metadata': obj_yaml.pop('metadata', {}),
|
||
'spec': obj_yaml.pop('spec', {})
|
||
}
|
||
|
||
# 清理metadata中的不必要字段
|
||
metadata_fields = ['selfLink', 'generation', 'creationTimestamp',
|
||
'resourceVersion', 'uid', 'managedFields']
|
||
for field in metadata_fields:
|
||
ordered_obj['metadata'].pop(field, None)
|
||
|
||
# 清理annotations
|
||
annotations_to_remove = [
|
||
'deployment.kubernetes.io/revision',
|
||
'kubectl.kubernetes.io/restartedAt',
|
||
'kubectl.kubernetes.io/last-applied-configuration'
|
||
]
|
||
|
||
if 'annotations' in ordered_obj['metadata']:
|
||
for ann in annotations_to_remove:
|
||
ordered_obj['metadata']['annotations'].pop(ann, None)
|
||
|
||
# 如果annotations为空，则删除该字段
|
||
if not ordered_obj['metadata']['annotations']:
|
||
del ordered_obj['metadata']['annotations']
|
||
|
||
# 清理spec中的不必要字段
|
||
if resource_type == 'Service':
|
||
if 'clusterIPs' in ordered_obj['spec']:
|
||
del ordered_obj['spec']['clusterIPs']
|
||
if 'clusterIP' in ordered_obj['spec']:
|
||
del ordered_obj['spec']['clusterIP']
|
||
if 'ipFamilies' in ordered_obj['spec']:
|
||
del ordered_obj['spec']['ipFamilies']
|
||
if 'internalTrafficPolicy' in ordered_obj['spec']:
|
||
del ordered_obj['spec']['internalTrafficPolicy']
|
||
if 'ipFamilyPolicy' in ordered_obj['spec']:
|
||
del ordered_obj['spec']['ipFamilyPolicy']
|
||
|
||
elif resource_type == 'Deployment':
|
||
if 'strategy' in ordered_obj['spec']:
|
||
# 保留strategy但移除不需要的字段
|
||
strategy = ordered_obj['spec']['strategy']
|
||
if 'rollingUpdate' in strategy:
|
||
rolling_update = strategy['rollingUpdate']
|
||
if 'maxUnavailable' in rolling_update:
|
||
del rolling_update['maxUnavailable']
|
||
if 'maxSurge' in rolling_update:
|
||
del rolling_update['maxSurge']
|
||
# 如果rollingUpdate为空，则删除
|
||
if not rolling_update:
|
||
del strategy['rollingUpdate']
|
||
|
||
# 移除status字段和其他不需要的字段
|
||
if 'status' in obj_yaml:
|
||
del obj_yaml['status']
|
||
|
||
# 添加其他可能需要的字段
|
||
for key, value in obj_yaml.items():
|
||
if key not in ['apiVersion', 'kind', 'metadata', 'spec'] and value is not None:
|
||
ordered_obj[key] = value
|
||
|
||
return ordered_obj
|
||
|
||
def export_resources():
|
||
"""导出Kubernetes资源到不同目录"""
|
||
# 加载kubeconfig
|
||
config.load_kube_config()
|
||
global api_client
|
||
api_client = client.ApiClient()
|
||
|
||
# 创建API客户端
|
||
apps_api = client.AppsV1Api(api_client)
|
||
core_api = client.CoreV1Api(api_client)
|
||
networking_api = client.NetworkingV1Api(api_client)
|
||
|
||
# 获取所有资源
|
||
all_deployments = apps_api.list_deployment_for_all_namespaces().items
|
||
all_services = core_api.list_service_for_all_namespaces().items
|
||
all_ingresses = networking_api.list_ingress_for_all_namespaces().items
|
||
|
||
# 创建输出目录
|
||
deployments_dir = os.path.join(os.getcwd(), 'deployments')
|
||
ingress_dir = os.path.join(os.getcwd(), 'ingress')
|
||
os.makedirs(deployments_dir, exist_ok=True)
|
||
os.makedirs(ingress_dir, exist_ok=True)
|
||
|
||
# 按命名空间和组织资源
|
||
namespace_resources = {}
|
||
|
||
# 处理Deployment和Service
|
||
for dep in all_deployments:
|
||
namespace = dep.metadata.namespace
|
||
dep_name = dep.metadata.name
|
||
|
||
if namespace not in namespace_resources:
|
||
namespace_resources[namespace] = {'deployments': {}, 'services': {}, 'ingresses': {}}
|
||
|
||
# 清理并添加Deployment
|
||
cleaned_dep = sanitize_resource(dep, 'Deployment')
|
||
namespace_resources[namespace]['deployments'][dep_name] = cleaned_dep
|
||
|
||
# 查找匹配的Service
|
||
pod_labels = dep.spec.template.metadata.labels
|
||
matched_svcs = [
|
||
svc for svc in all_services
|
||
if svc.metadata.namespace == namespace
|
||
and svc.spec.selector
|
||
and all(pod_labels.get(k) == v for k, v in svc.spec.selector.items())
|
||
]
|
||
|
||
# 清理并添加匹配的Service
|
||
for svc in matched_svcs:
|
||
cleaned_svc = sanitize_resource(svc, 'Service')
|
||
namespace_resources[namespace]['services'][svc.metadata.name] = cleaned_svc
|
||
|
||
# 处理Ingress
|
||
for ingress in all_ingresses:
|
||
namespace = ingress.metadata.namespace
|
||
if namespace not in namespace_resources:
|
||
namespace_resources[namespace] = {'deployments': {}, 'services': {}, 'ingresses': {}}
|
||
|
||
# 清理并添加Ingress
|
||
cleaned_ingress = sanitize_resource(ingress, 'Ingress')
|
||
namespace_resources[namespace]['ingresses'][ingress.metadata.name] = cleaned_ingress
|
||
|
||
# 导出Deployment+Service到deployments目录
|
||
for namespace, resources in namespace_resources.items():
|
||
# 在deployments目录下创建命名空间子目录
|
||
ns_deployments_dir = os.path.join(deployments_dir, namespace)
|
||
os.makedirs(ns_deployments_dir, exist_ok=True)
|
||
|
||
# 导出每个Deployment及其相关Service
|
||
for dep_name, deployment in resources['deployments'].items():
|
||
# 查找与此Deployment相关的Service
|
||
related_services = []
|
||
for svc_name, service in resources['services'].items():
|
||
# 简单匹配：Service名称包含Deployment名称
|
||
if dep_name in svc_name:
|
||
related_services.append(service)
|
||
|
||
# 组合Deployment和Service
|
||
combined_resources = [deployment]
|
||
combined_resources.extend(related_services)
|
||
|
||
# 生成文件名
|
||
filename = f"{dep_name}.yaml"
|
||
filepath = os.path.join(ns_deployments_dir, filename)
|
||
|
||
# 写入文件，确保正确的YAML格式
|
||
with open(filepath, 'w') as f:
|
||
yaml.dump_all(combined_resources, f, default_flow_style=False, sort_keys=False, explicit_start=True)
|
||
|
||
print(f"Exported {len(combined_resources)} resources to {filepath}")
|
||
|
||
# 导出没有关联Deployment的独立Service
|
||
exported_services = set()
|
||
for deployment_name in resources['deployments']:
|
||
for svc_name in resources['services']:
|
||
if deployment_name in svc_name:
|
||
exported_services.add(svc_name)
|
||
|
||
for svc_name, service in resources['services'].items():
|
||
if svc_name not in exported_services:
|
||
filename = f"{svc_name}.yaml"
|
||
filepath = os.path.join(ns_deployments_dir, filename)
|
||
|
||
with open(filepath, 'w') as f:
|
||
yaml.dump(service, f, default_flow_style=False, sort_keys=False)
|
||
|
||
print(f"Exported standalone Service to {filepath}")
|
||
|
||
# 导出Ingress到ingress目录
|
||
for namespace, resources in namespace_resources.items():
|
||
if resources['ingresses']:
|
||
# 在ingress目录下创建命名空间子目录
|
||
ns_ingress_dir = os.path.join(ingress_dir, namespace)
|
||
os.makedirs(ns_ingress_dir, exist_ok=True)
|
||
|
||
# 导出所有Ingress
|
||
for ing_name, ingress in resources['ingresses'].items():
|
||
filename = f"{ing_name}.yaml"
|
||
filepath = os.path.join(ns_ingress_dir, filename)
|
||
|
||
with open(filepath, 'w') as f:
|
||
yaml.dump(ingress, f, default_flow_style=False, sort_keys=False)
|
||
|
||
print(f"Exported Ingress to {filepath}")
|
||
|
||
if __name__ == "__main__":
|
||
export_resources() |