如何查询 TKE 集群列表¶

from tencentcloud.common import credential
from tencentcloud.tke.v20180525 import tke_client, models

# 初始化认证
cred = credential.Credential("SecretId", "SecretKey")
client = tke_client.TkeClient(cred, "ap-guangzhou")

# 方式1: 查询所有集群
req = models.DescribeClustersRequest()
resp = client.DescribeClusters(req)

print(f"集群总数: {resp.TotalCount}")
for cluster in resp.Clusters:
    print(f"集群名称: {cluster.ClusterName}")
    print(f"集群ID: {cluster.ClusterId}")
    print(f"集群状态: {cluster.ClusterStatus}")
    print(f"K8s版本: {cluster.ClusterVersion}")
    print("-" * 40)

# 方式2: 查询指定集群
req = models.DescribeClustersRequest()
req.ClusterIds = ["cls-xxxxxxxx"]
resp = client.DescribeClusters(req)

if resp.TotalCount > 0:
    cluster = resp.Clusters[0]
    print(f"集群详情:")
    print(f"  名称: {cluster.ClusterName}")
    print(f"  状态: {cluster.ClusterStatus}")
    print(f"  版本: {cluster.ClusterVersion}")
    print(f"  节点数: {cluster.ClusterNodeNum}")

# 方式3: 使用过滤条件
req = models.DescribeClustersRequest()
req.Filters = [
    models.Filter(Name="ClusterStatus", Values=["Running"]),
    models.Filter(Name="ClusterType", Values=["MANAGED_CLUSTER"])
]
resp = client.DescribeClusters(req)

print(f"找到 {resp.TotalCount} 个运行中的托管集群")

# 验证响应
assert resp.TotalCount > 0, "未找到任何集群"
assert len(resp.Clusters) > 0, "集群列表为空"

# 验证集群信息完整
cluster = resp.Clusters[0]
assert cluster.ClusterId is not None, "集群 ID 为空"
assert cluster.ClusterName is not None, "集群名称为空"
assert cluster.ClusterStatus in ["Running", "Creating", "Abnormal"], "集群状态异常"

def get_all_clusters(client, region):
    """获取所有集群（分页）"""
    all_clusters = []
    offset = 0
    limit = 100  # 每页最多 100 条

    while True:
        req = models.DescribeClustersRequest()
        req.Limit = limit
        req.Offset = offset

        resp = client.DescribeClusters(req)
        all_clusters.extend(resp.Clusters)

        if len(resp.Clusters) < limit:
            break

        offset += limit

    return all_clusters

# 使用示例
clusters = get_all_clusters(client, "ap-guangzhou")
print(f"总共找到 {len(clusters)} 个集群")

regions = ["ap-guangzhou", "ap-beijing", "ap-shanghai"]
all_clusters = {}

for region in regions:
    client = tke_client.TkeClient(cred, region)
    req = models.DescribeClustersRequest()
    resp = client.DescribeClusters(req)

    all_clusters[region] = resp.Clusters
    print(f"{region}: {resp.TotalCount} 个集群")

# 统计总数
total = sum(len(clusters) for clusters in all_clusters.values())
print(f"所有地域总共 {total} 个集群")

def filter_clusters(client, status=None, cluster_type=None, min_nodes=0):
    """按条件筛选集群"""
    req = models.DescribeClustersRequest()

    filters = []
    if status:
        filters.append(models.Filter(Name="ClusterStatus", Values=[status]))
    if cluster_type:
        filters.append(models.Filter(Name="ClusterType", Values=[cluster_type]))

    if filters:
        req.Filters = filters

    resp = client.DescribeClusters(req)

    # 按节点数过滤
    result = [c for c in resp.Clusters if c.ClusterNodeNum >= min_nodes]

    return result

# 查询运行中且节点数 ≥ 3 的托管集群
clusters = filter_clusters(
    client,
    status="Running",
    cluster_type="MANAGED_CLUSTER",
    min_nodes=3
)
print(f"找到 {len(clusters)} 个符合条件的集群")

import csv

def export_clusters_to_csv(clusters, filename="clusters.csv"):
    """导出集群信息到 CSV"""
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow([
            "集群ID", "集群名称", "状态", "版本", 
            "节点数", "规格", "VPC ID", "创建时间"
        ])

        for cluster in clusters:
            writer.writerow([
                cluster.ClusterId,
                cluster.ClusterName,
                cluster.ClusterStatus,
                cluster.ClusterVersion,
                cluster.ClusterNodeNum,
                cluster.ClusterLevel,
                cluster.ClusterNetworkSettings.VpcId,
                cluster.CreatedTime
            ])

    print(f"已导出 {len(clusters)} 个集群信息到 {filename}")

# 使用示例
req = models.DescribeClustersRequest()
resp = client.DescribeClusters(req)
export_clusters_to_csv(resp.Clusters)