使用 Django 构建支持 Kubernetes API 测试连接的 POST 接口

---

使用 Django 构建支持 Kubernetes API 测试连接的 POST 接口

在 Kubernetes 集群运维和开发中，验证集群的连通性和资源访问权限是常见需求。本文将以 Django 框架为基础，构建一个只支持 POST 请求的接口，用于测试 Kubernetes 集群连接并获取默认命名空间中的 Pod 列表。

---

功能需求

• 用户通过 HTTP 接口传递 Kubernetes API 地址、Token 和端口。
• 系统验证请求参数的完整性并测试 Kubernetes 集群的连通性。
• 返回默认命名空间的 Pod 列表或具体的错误信息。
• 限制接口仅支持 POST 请求。

---

使用 kubectl 获取 Token

在 Kubernetes 中，Token 通常存储在 Secret 资源中，并且用于访问集群的 API。以下命令可以帮助您获取 Kubernetes 系统命名空间中的 Token：

kubectl -n kube-system describe $(kubectl -n kube-system get secret -n kube-system -o name | grep namespace) | grep token

命令解析

1. kubectl -n kube-system get secret -n kube-system -o name: 该命令列出 kube-system 命名空间中的所有 Secret 资源，并只输出 Secret 的名字。
2. grep namespace: 过滤出包含 namespace 字符串的 Secret 名称。通常，集群的服务账户 Token 存储在名为 namespace-xxxxx 的 Secret 中。
3. kubectl -n kube-system describe <secret_name>: 通过 describe 命令获取该 Secret 的详细信息。
4. grep token: 从 Secret 的详细信息中提取包含 Token 的部分。

输出示例

该命令会输出类似于以下内容的 Token：

token: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

完整代码实现

以下代码基于 Python 和 Django 框架，分为 Kubernetes API 客户端类、接口视图函数以及路由配置三部分。

Kubernetes API 客户端类

我们定义了一个 KubernetesClient 类，封装与 Kubernetes API 的交互逻辑。

from kubernetes import clientclass KubernetesClient:def __init__(self, host, token):"""初始化 Kubernetes API 客户端"""# 配置 Kubernetes API 客户端k8s_config = client.Configuration()k8s_config.host = hostk8s_config.verify_ssl = False  # 禁用 SSL 验证（仅测试环境使用）k8s_config.api_key = {"authorization": "Bearer " + token}client.Configuration.set_default(k8s_config)# 初始化核心和应用 APIself.k8s_core_v1 = client.CoreV1Api()def get_pods(self, namespace="default"):"""获取指定命名空间中的 Pod 列表"""try:pods = self.k8s_core_v1.list_namespaced_pod(namespace=namespace)return [pod.metadata.name for pod in pods.items]except client.ApiException as e:raise Exception(f"获取 Pod 列表失败：{e}")

功能说明

• 初始化时配置 Kubernetes API 地址和认证 Token。
• get_pods 方法通过 list_namespaced_pod 获取 Pod 列表并返回 Pod 名称。

---

Django 接口视图

在 Django 中，我们通过视图函数实现接口逻辑，并限制只支持 POST 请求。

from django.http import JsonResponse
import jsondef test_join(request):"""测试 Kubernetes 集群的连接"""# 检查请求方法是否为 POSTif request.method != "POST":return JsonResponse({"code": 405, "message": "仅支持 POST 请求"})# 提取请求参数try:body = json.loads(request.body)ipaddr = body.get("ipaddr")token = body.get("secret_key")port = body.get("port")if not all([ipaddr, token, port]):return JsonResponse({"code": 400, "message": "请输入正确的参数"})except json.JSONDecodeError:return JsonResponse({"code": 400, "message": "请求数据格式错误"})# 构造 Kubernetes API 地址host = f"https://{ipaddr}:{port}"try:# 初始化 Kubernetes 客户端kube_client = KubernetesClient(host, token)pods = kube_client.get_pods()return JsonResponse({"code": 200, "message": "连接成功", "data": pods})except Exception as e:return JsonResponse({"code": 400, "message": f"连接失败: {str(e)}"})

关键点解析

1. 请求方法限制:
   • 使用 request.method 检查请求类型，如果不是 POST 返回 HTTP 405 状态码。
2. 参数验证:
   • 从请求体中解析 ipaddr、token 和 port，确保参数完整且格式正确。
3. 连接测试:
   • 通过 KubernetesClient 测试连接并获取默认命名空间中的 Pod 列表。
4. 错误处理:
   • 捕获参数错误、Kubernetes API 异常等，并返回详细的错误信息。

---

路由配置

将视图函数绑定到路由，使其可以通过 HTTP 调用：

from django.urls import path
from .views import test_joinurlpatterns = [path('join_test/', test_join),
]

---

接口测试

以下是测试接口的示例。

请求示例

使用 curl 工具发送 POST 请求：

curl -X POST http://127.0.0.1:8000/join_test/ \
-H "Content-Type: application/json" \
-d '{"ipaddr": "192.168.1.10","secret_key": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...","port": 6443
}'

响应结果

成功

如果连接成功并获取到 Pod 列表，返回如下结果：

{"code": 200,"message": "连接成功","data": ["nginx-deployment-7c77f6df69-8wmdt", "redis-pod"]
}

错误

• 非 POST 请求：

  {"code": 405,"message": "仅支持 POST 请求"
  }
  

• 参数缺失：

  {"code": 400,"message": "请输入正确的参数"
  }
  

• 集群认证失败：

  {"code": 400,"message": "连接失败: Unauthorized"
  }
  

---

优化建议

1. 安全性

• 启用 SSL 验证：将 verify_ssl 设置为 True，并配置证书路径。
• 使用环境变量管理敏感信息（如 Token）。

2. 错误处理

• 针对不同异常返回精确的错误码（如认证失败返回 401，资源未找到返回 404）。

3. 扩展返回数据

在返回的 Pod 列表中附加更多信息，例如 Pod 的状态、节点信息等：

return [{"name": pod.metadata.name,"status": pod.status.phase,"node": pod.spec.node_name
} for pod in pods.items]

---

完整代码示例

以下是完整的代码示例，包括 Kubernetes API 客户端类、Django 视图函数和路由配置，确保能够实现测试 Kubernetes 集群连接并返回 Pod 列表的功能：

---

Kubernetes API 客户端类

from kubernetes import clientclass KubernetesClient:def __init__(self, host, token):"""初始化 Kubernetes API 客户端"""# 配置 Kubernetes API 客户端k8s_config = client.Configuration()k8s_config.host = hostk8s_config.verify_ssl = False  # 禁用 SSL 验证（仅测试环境使用）k8s_config.api_key = {"authorization": "Bearer " + token}client.Configuration.set_default(k8s_config)# 初始化核心和应用 APIself.k8s_core_v1 = client.CoreV1Api()def get_pods(self, namespace="default"):"""获取指定命名空间中的 Pod 列表"""try:pods = self.k8s_core_v1.list_namespaced_pod(namespace=namespace)return [pod.metadata.name for pod in pods.items]except client.ApiException as e:raise Exception(f"获取 Pod 列表失败：{e}")

---

Django 接口视图

from django.http import JsonResponse
import json
from .kubernetes_client import KubernetesClientdef test_join(request):"""测试 Kubernetes 集群的连接"""# 检查请求方法是否为 POSTif request.method != "POST":return JsonResponse({"code": 405, "message": "仅支持 POST 请求"}, status=405)# 提取请求参数try:body = json.loads(request.body)ipaddr = body.get("ipaddr")token = body.get("secret_key")port = body.get("port")if not all([ipaddr, token, port]):return JsonResponse({"code": 400, "message": "请输入正确的参数"})except json.JSONDecodeError:return JsonResponse({"code": 400, "message": "请求数据格式错误"})# 构造 Kubernetes API 地址host = f"https://{ipaddr}:{port}"try:# 初始化 Kubernetes 客户端kube_client = KubernetesClient(host, token)pods = kube_client.get_pods()return JsonResponse({"code": 200, "message": "连接成功", "data": pods})except Exception as e:return JsonResponse({"code": 400, "message": f"连接失败: {str(e)}"})

---

路由配置

from django.urls import path
from .views import test_joinurlpatterns = [path('join_test/', test_join),
]

---

完整代码说明

1. KubernetesClient 类:

   • KubernetesClient 类封装了与 Kubernetes 集群的连接和交互，特别是获取 Pod 列表的功能。
   • get_pods 方法通过 Kubernetes API 获取指定命名空间中的 Pod 列表。

2. test_join 视图函数:

   • 视图函数仅支持 POST 请求，使用 request.method 检查请求类型。
   • 提取请求体中的 ipaddr、secret_key 和 port 参数，确保这些参数完整且正确。
   • 使用 KubernetesClient 测试 Kubernetes 集群连接并返回 Pod 列表。

3. 路由配置:

   • 配置了路由，使得用户可以通过 POST 请求访问 /join_test/ 路径进行集群连接测试。

---

总结

本文详细介绍了如何通过 Django 构建一个只支持 POST 请求的接口，用于测试 Kubernetes 集群连接和获取 Pod 列表。通过请求方法限制、参数验证和异常处理，该接口在功能性和安全性上都有较好的保障。希望对您有所帮助！