565 lines
22 KiB
Python
565 lines
22 KiB
Python
"""
|
||
Kubernetes管理模块 - 负责与Kubernetes集群交互
|
||
"""
|
||
from kubernetes import client, config
|
||
from kubernetes.client.rest import ApiException
|
||
from typing import Dict, List, Optional
|
||
import logging
|
||
import os
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class K8sManager:
|
||
"""Kubernetes资源管理器"""
|
||
|
||
def __init__(self, namespace: str = "ai-agents", kubeconfig_path: str = None):
|
||
"""
|
||
初始化Kubernetes管理器
|
||
|
||
Args:
|
||
namespace: AI Agent部署的命名空间
|
||
kubeconfig_path: kubeconfig文件路径（可选）
|
||
"""
|
||
self.namespace = namespace
|
||
self._load_kube_config(kubeconfig_path)
|
||
|
||
self.v1 = client.CoreV1Api()
|
||
self.apps_v1 = client.AppsV1Api()
|
||
|
||
# 确保命名空间存在
|
||
self._ensure_namespace()
|
||
|
||
def _load_kube_config(self, kubeconfig_path: str = None):
|
||
"""
|
||
加载Kubernetes配置
|
||
优先级：
|
||
1. 集群内ServiceAccount（推荐用于生产环境）
|
||
2. 指定的kubeconfig文件路径
|
||
3. 默认的kubeconfig路径 (~/.kube/config)
|
||
"""
|
||
try:
|
||
# 方式1: 尝试加载集群内配置（当服务运行在K8s中且有ServiceAccount时）
|
||
config.load_incluster_config()
|
||
logger.info("✅ 使用集群内ServiceAccount配置")
|
||
except config.ConfigException:
|
||
try:
|
||
if kubeconfig_path and os.path.exists(kubeconfig_path):
|
||
# 方式2: 使用指定的kubeconfig文件
|
||
config.load_kube_config(config_file=kubeconfig_path)
|
||
logger.info(f"✅ 使用指定的kubeconfig: {kubeconfig_path}")
|
||
else:
|
||
# 方式3: 使用默认kubeconfig（开发环境）
|
||
config.load_kube_config()
|
||
logger.info("✅ 使用默认kubeconfig (~/.kube/config)")
|
||
except Exception as e:
|
||
logger.error(f"❌ 无法加载Kubernetes配置: {e}")
|
||
raise Exception(f"Kubernetes配置加载失败: {e}")
|
||
|
||
def _ensure_namespace(self):
|
||
"""确保AI Agent命名空间存在"""
|
||
try:
|
||
self.v1.read_namespace(name=self.namespace)
|
||
logger.info(f"命名空间 {self.namespace} 已存在")
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
# 创建命名空间
|
||
namespace_manifest = client.V1Namespace(
|
||
metadata=client.V1ObjectMeta(name=self.namespace)
|
||
)
|
||
self.v1.create_namespace(body=namespace_manifest)
|
||
logger.info(f"创建命名空间 {self.namespace}")
|
||
else:
|
||
raise
|
||
|
||
# 模板端口映射
|
||
TEMPLATE_PORTS = {
|
||
"jina_search_agent": 8080,
|
||
}
|
||
|
||
# 模板所需环境变量说明
|
||
TEMPLATE_ENV_INFO = {
|
||
"jina_search_agent": {
|
||
"required": {
|
||
"JINA_API_KEY": "Jina API密钥，从 https://jina.ai/ 获取"
|
||
},
|
||
"optional": {
|
||
"SERVICE_PORT": "HTTP服务端口，默认8080",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认0.0.0.0"
|
||
}
|
||
},
|
||
"mysql_agent": {
|
||
"required": {
|
||
"MYSQL_HOST": "MySQL数据库主机地址",
|
||
"MYSQL_USER": "MySQL用户名",
|
||
"MYSQL_PASSWORD": "MySQL密码",
|
||
"MYSQL_DATABASE": "MySQL数据库名",
|
||
"OPENAI_API_KEY": "OpenAI API密钥"
|
||
},
|
||
"optional": {
|
||
"MYSQL_PORT": "MySQL端口，默认3306"
|
||
}
|
||
},
|
||
"postgresql_agent": {
|
||
"required": {
|
||
"POSTGRES_HOST": "PostgreSQL数据库主机地址",
|
||
"POSTGRES_USER": "PostgreSQL用户名",
|
||
"POSTGRES_PASSWORD": "PostgreSQL密码",
|
||
"POSTGRES_DATABASE": "PostgreSQL数据库名",
|
||
"OPENAI_API_KEY": "OpenAI API密钥"
|
||
},
|
||
"optional": {
|
||
"POSTGRES_PORT": "PostgreSQL端口，默认5432"
|
||
}
|
||
}
|
||
}
|
||
|
||
def get_template_info(self, template: str) -> Dict:
|
||
"""
|
||
获取模板信息
|
||
|
||
Args:
|
||
template: 模板类型
|
||
|
||
Returns:
|
||
模板信息（端口、所需环境变量等）
|
||
"""
|
||
return {
|
||
"template": template,
|
||
"port": self.TEMPLATE_PORTS.get(template),
|
||
"env_info": self.TEMPLATE_ENV_INFO.get(template, {})
|
||
}
|
||
|
||
def create_pod(self, pod_name: str, template: str, config_data: Dict) -> Dict:
|
||
"""
|
||
创建Pod
|
||
|
||
Args:
|
||
pod_name: Pod名称
|
||
template: 模板类型（echo_agent, chat_agent等）
|
||
config_data: 配置信息（replicas, resources等）
|
||
|
||
Returns:
|
||
创建的Pod信息，包含访问地址
|
||
"""
|
||
try:
|
||
# 生成Pod规格
|
||
pod_manifest = self._generate_pod_manifest(pod_name, template, config_data)
|
||
|
||
# 创建Pod
|
||
response = self.v1.create_namespaced_pod(
|
||
namespace=self.namespace,
|
||
body=pod_manifest
|
||
)
|
||
|
||
logger.info(f"Pod {pod_name} 创建成功")
|
||
|
||
# 获取服务端口
|
||
service_port = self.TEMPLATE_PORTS.get(template)
|
||
|
||
result = {
|
||
"name": response.metadata.name,
|
||
"namespace": response.metadata.namespace,
|
||
"status": response.status.phase,
|
||
"created_at": response.metadata.creation_timestamp.isoformat() if response.metadata.creation_timestamp else None,
|
||
"template": template
|
||
}
|
||
|
||
# 如果是HTTP服务类型的agent，添加访问信息
|
||
if service_port:
|
||
result["service_port"] = service_port
|
||
result["access_info"] = {
|
||
"note": "Pod IP将在Pod运行后可用，请通过 /agents/{name}/status 获取",
|
||
"port": service_port,
|
||
"endpoints": {
|
||
"root": f"http://<pod_ip>:{service_port}/",
|
||
"health": f"http://<pod_ip>:{service_port}/health"
|
||
}
|
||
}
|
||
|
||
return result
|
||
except ApiException as e:
|
||
logger.error(f"创建Pod失败: {e}")
|
||
raise Exception(f"创建Pod失败: {e.reason}")
|
||
|
||
def _generate_pod_manifest(self, pod_name: str, template: str, config_data: Dict) -> client.V1Pod:
|
||
"""生成Pod配置清单"""
|
||
|
||
# 默认资源配置
|
||
replicas = config_data.get("replicas", 1)
|
||
cpu_request = config_data.get("cpu_request", "100m")
|
||
cpu_limit = config_data.get("cpu_limit", "500m")
|
||
memory_request = config_data.get("memory_request", "128Mi")
|
||
memory_limit = config_data.get("memory_limit", "512Mi")
|
||
|
||
# 根据模板类型选择镜像
|
||
image_map = {
|
||
"echo_agent": "agnettaiji.azurecr.io/ai-agents/echo-agent:latest",
|
||
"chat_agent": "agnettaiji.azurecr.io/ai-agents/chat-agent:latest",
|
||
"code_agent": "agnettaiji.azurecr.io/ai-agents/code-agent:latest",
|
||
"search_agent": "agnettaiji.azurecr.io/ai-agents/search-agent:latest",
|
||
"mysql_agent": "agnettaiji.azurecr.io/ai-agents/mysql-agent:latest",
|
||
"postgresql_agent": "agnettaiji.azurecr.io/ai-agents/postgresql-agent:latest",
|
||
"jina_search_agent": "agnettaiji.azurecr.io/ai-agents/jina-search-agent:latest",
|
||
}
|
||
image = image_map.get(template, image_map["echo_agent"])
|
||
|
||
# 构建环境变量列表
|
||
env_vars = [
|
||
client.V1EnvVar(name="POD_NAME", value=pod_name),
|
||
client.V1EnvVar(name="TEMPLATE_TYPE", value=template)
|
||
]
|
||
|
||
# 添加用户自定义环境变量
|
||
custom_env = config_data.get("env", {})
|
||
for key, value in custom_env.items():
|
||
env_vars.append(client.V1EnvVar(name=key, value=str(value)))
|
||
|
||
logger.info(f"Pod {pod_name} 环境变量数量: {len(env_vars)}")
|
||
|
||
# 设置容器端口（如果是HTTP服务类型的agent）
|
||
container_ports = None
|
||
if template in ["jina_search_agent"]:
|
||
container_ports = [client.V1ContainerPort(container_port=8080)]
|
||
|
||
# 创建Pod规格
|
||
container = client.V1Container(
|
||
name=pod_name,
|
||
image=image,
|
||
resources=client.V1ResourceRequirements(
|
||
requests={"cpu": cpu_request, "memory": memory_request},
|
||
limits={"cpu": cpu_limit, "memory": memory_limit}
|
||
),
|
||
env=env_vars,
|
||
ports=container_ports
|
||
)
|
||
|
||
pod_spec = client.V1PodSpec(
|
||
containers=[container],
|
||
restart_policy="Always",
|
||
image_pull_secrets=[client.V1LocalObjectReference(name="acr-secret")]
|
||
)
|
||
|
||
# 构建标签（合并默认标签和用户自定义标签）
|
||
labels = {
|
||
"app": "ai-agent",
|
||
"template": template,
|
||
"managed-by": "agent-manager"
|
||
}
|
||
# 添加用户自定义标签
|
||
if "labels" in config_data:
|
||
labels.update(config_data["labels"])
|
||
|
||
pod_manifest = client.V1Pod(
|
||
api_version="v1",
|
||
kind="Pod",
|
||
metadata=client.V1ObjectMeta(
|
||
name=pod_name,
|
||
labels=labels
|
||
),
|
||
spec=pod_spec
|
||
)
|
||
|
||
return pod_manifest
|
||
|
||
def delete_pod(self, pod_name: str) -> Dict:
|
||
"""
|
||
删除Pod
|
||
|
||
Args:
|
||
pod_name: Pod名称
|
||
|
||
Returns:
|
||
删除结果
|
||
"""
|
||
try:
|
||
self.v1.delete_namespaced_pod(
|
||
name=pod_name,
|
||
namespace=self.namespace,
|
||
body=client.V1DeleteOptions()
|
||
)
|
||
logger.info(f"Pod {pod_name} 删除成功")
|
||
return {"status": "success", "message": f"Pod {pod_name} 已删除"}
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
return {"status": "not_found", "message": f"Pod {pod_name} 不存在"}
|
||
logger.error(f"删除Pod失败: {e}")
|
||
raise Exception(f"删除Pod失败: {e.reason}")
|
||
|
||
def get_pod_status(self, pod_name: str) -> Dict:
|
||
"""
|
||
获取Pod状态
|
||
|
||
Args:
|
||
pod_name: Pod名称
|
||
|
||
Returns:
|
||
Pod状态信息，包含访问URL和资源使用情况
|
||
"""
|
||
try:
|
||
pod = self.v1.read_namespaced_pod(
|
||
name=pod_name,
|
||
namespace=self.namespace
|
||
)
|
||
|
||
template = pod.metadata.labels.get("template", "unknown")
|
||
pod_ip = pod.status.pod_ip
|
||
service_port = self.TEMPLATE_PORTS.get(template)
|
||
|
||
# 获取资源配额信息
|
||
container = pod.spec.containers[0]
|
||
resources = container.resources
|
||
resource_requests = {
|
||
"cpu": resources.requests.get("cpu") if resources.requests else None,
|
||
"memory": resources.requests.get("memory") if resources.requests else None
|
||
}
|
||
resource_limits = {
|
||
"cpu": resources.limits.get("cpu") if resources.limits else None,
|
||
"memory": resources.limits.get("memory") if resources.limits else None
|
||
}
|
||
|
||
# 尝试获取实际资源使用情况（需要metrics-server）
|
||
resource_usage = self._get_pod_resource_usage(pod_name)
|
||
|
||
# 获取容器实际状态 - 检查是否崩溃或异常
|
||
container_statuses = pod.status.container_statuses or []
|
||
actual_status = pod.status.phase # 默认使用Pod阶段
|
||
health_status = "healthy"
|
||
container_info = []
|
||
|
||
for container_status in container_statuses:
|
||
container_state = {}
|
||
restart_count = container_status.restart_count
|
||
|
||
# 检查容器状态
|
||
if container_status.state.running:
|
||
container_state = {
|
||
"state": "running",
|
||
"started_at": container_status.state.running.started_at.isoformat() if container_status.state.running.started_at else None
|
||
}
|
||
elif container_status.state.waiting:
|
||
container_state = {
|
||
"state": "waiting",
|
||
"reason": container_status.state.waiting.reason,
|
||
"message": container_status.state.waiting.message
|
||
}
|
||
# 容器在等待状态，标记为不健康
|
||
health_status = "unhealthy"
|
||
actual_status = "Waiting"
|
||
elif container_status.state.terminated:
|
||
container_state = {
|
||
"state": "terminated",
|
||
"reason": container_status.state.terminated.reason,
|
||
"exit_code": container_status.state.terminated.exit_code,
|
||
"message": container_status.state.terminated.message,
|
||
"finished_at": container_status.state.terminated.finished_at.isoformat() if container_status.state.terminated.finished_at else None
|
||
}
|
||
# 容器已终止，标记为不健康
|
||
health_status = "unhealthy"
|
||
actual_status = "Terminated"
|
||
|
||
# 检查容器是否就绪
|
||
if not container_status.ready:
|
||
health_status = "unhealthy"
|
||
|
||
# 如果重启次数过多，也标记为不健康
|
||
if restart_count > 5:
|
||
health_status = "degraded"
|
||
|
||
container_info.append({
|
||
"name": container_status.name,
|
||
"ready": container_status.ready,
|
||
"restart_count": restart_count,
|
||
**container_state
|
||
})
|
||
|
||
result = {
|
||
"name": pod.metadata.name,
|
||
"namespace": pod.metadata.namespace,
|
||
"status": actual_status,
|
||
"health_status": health_status, # 新增：真实健康状态
|
||
"template": template,
|
||
"created_at": pod.metadata.creation_timestamp.isoformat() if pod.metadata.creation_timestamp else None,
|
||
"node": pod.spec.node_name,
|
||
"pod_ip": pod_ip,
|
||
"containers": container_info, # 新增：容器详细信息
|
||
"resources": {
|
||
"requests": resource_requests,
|
||
"limits": resource_limits,
|
||
"usage": resource_usage
|
||
},
|
||
"conditions": [
|
||
{
|
||
"type": condition.type,
|
||
"status": condition.status,
|
||
"reason": condition.reason
|
||
}
|
||
for condition in (pod.status.conditions or [])
|
||
]
|
||
}
|
||
|
||
# 如果是HTTP服务类型的agent且Pod已有IP，添加访问URL
|
||
if service_port and pod_ip:
|
||
result["service_port"] = service_port
|
||
result["access_url"] = f"http://{pod_ip}:{service_port}"
|
||
result["endpoints"] = {
|
||
"root": f"http://{pod_ip}:{service_port}/",
|
||
"health": f"http://{pod_ip}:{service_port}/health"
|
||
}
|
||
|
||
return result
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
return {"status": "not_found", "message": f"Pod {pod_name} 不存在"}
|
||
logger.error(f"获取Pod状态失败: {e}")
|
||
raise Exception(f"获取Pod状态失败: {e.reason}")
|
||
|
||
def _get_pod_resource_usage(self, pod_name: str) -> Dict:
|
||
"""
|
||
获取Pod实际资源使用情况（需要metrics-server）
|
||
|
||
Args:
|
||
pod_name: Pod名称
|
||
|
||
Returns:
|
||
资源使用信息（cpu、memory）
|
||
"""
|
||
try:
|
||
# 使用CustomObjectsApi调用metrics API
|
||
custom_api = client.CustomObjectsApi()
|
||
metrics = custom_api.get_namespaced_custom_object(
|
||
group="metrics.k8s.io",
|
||
version="v1beta1",
|
||
namespace=self.namespace,
|
||
plural="pods",
|
||
name=pod_name
|
||
)
|
||
|
||
# 解析容器资源使用情况
|
||
containers = metrics.get("containers", [])
|
||
if containers:
|
||
container = containers[0]
|
||
usage = container.get("usage", {})
|
||
return {
|
||
"cpu": usage.get("cpu"),
|
||
"memory": usage.get("memory"),
|
||
"available": True
|
||
}
|
||
|
||
return {"cpu": None, "memory": None, "available": False, "reason": "无容器数据"}
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
return {"cpu": None, "memory": None, "available": False, "reason": "metrics-server未安装或Pod不存在"}
|
||
logger.warning(f"获取Pod资源使用情况失败: {e.reason}")
|
||
return {"cpu": None, "memory": None, "available": False, "reason": f"获取失败: {e.reason}"}
|
||
except Exception as e:
|
||
logger.warning(f"获取Pod资源使用情况异常: {str(e)}")
|
||
return {"cpu": None, "memory": None, "available": False, "reason": f"异常: {str(e)}"}
|
||
|
||
def get_pod_metrics(self, pod_name: str) -> Dict:
|
||
"""
|
||
获取Pod资源使用情况（CPU、内存）
|
||
|
||
Args:
|
||
pod_name: Pod名称
|
||
|
||
Returns:
|
||
Pod实时资源使用信息和配额信息
|
||
"""
|
||
try:
|
||
# 获取 Pod 配额信息
|
||
pod = self.v1.read_namespaced_pod(
|
||
name=pod_name,
|
||
namespace=self.namespace
|
||
)
|
||
|
||
container = pod.spec.containers[0]
|
||
resources = container.resources
|
||
|
||
result = {
|
||
"name": pod_name,
|
||
"namespace": self.namespace,
|
||
"requests": {
|
||
"cpu": resources.requests.get("cpu") if resources.requests else None,
|
||
"memory": resources.requests.get("memory") if resources.requests else None
|
||
},
|
||
"limits": {
|
||
"cpu": resources.limits.get("cpu") if resources.limits else None,
|
||
"memory": resources.limits.get("memory") if resources.limits else None
|
||
}
|
||
}
|
||
|
||
# 尝试获取实时使用情况（需要 metrics-server）
|
||
try:
|
||
from kubernetes.client import CustomObjectsApi
|
||
custom_api = CustomObjectsApi()
|
||
|
||
# 调用 metrics.k8s.io API
|
||
metrics = custom_api.get_namespaced_custom_object(
|
||
group="metrics.k8s.io",
|
||
version="v1beta1",
|
||
namespace=self.namespace,
|
||
plural="pods",
|
||
name=pod_name
|
||
)
|
||
|
||
# 提取实时使用数据
|
||
if metrics and "containers" in metrics:
|
||
container_metrics = metrics["containers"][0]
|
||
usage = container_metrics.get("usage", {})
|
||
|
||
result["usage"] = {
|
||
"cpu": usage.get("cpu"),
|
||
"memory": usage.get("memory")
|
||
}
|
||
result["timestamp"] = metrics.get("timestamp")
|
||
logger.info(f"✅ 获取到实时资源使用: CPU={usage.get('cpu')}, Memory={usage.get('memory')}")
|
||
else:
|
||
result["usage"] = None
|
||
logger.warning(f"⚠️ Metrics 数据格式异常")
|
||
|
||
except Exception as metrics_error:
|
||
logger.warning(f"⚠️ 无法获取实时资源使用（可能未安装 metrics-server）: {str(metrics_error)}")
|
||
result["usage"] = None
|
||
result["metrics_available"] = False
|
||
|
||
return result
|
||
|
||
except ApiException as e:
|
||
logger.error(f"获取Pod资源信息失败: {e}")
|
||
raise Exception(f"获取Pod资源信息失败: {e.reason}")
|
||
|
||
def list_pods(self, label_selector: Optional[str] = None) -> List[Dict]:
|
||
"""
|
||
列出所有Pod
|
||
|
||
Args:
|
||
label_selector: 标签选择器（可选）
|
||
|
||
Returns:
|
||
Pod列表
|
||
"""
|
||
try:
|
||
if label_selector is None:
|
||
label_selector = "managed-by=agent-manager"
|
||
|
||
pods = self.v1.list_namespaced_pod(
|
||
namespace=self.namespace,
|
||
label_selector=label_selector
|
||
)
|
||
|
||
return [
|
||
{
|
||
"name": pod.metadata.name,
|
||
"status": pod.status.phase,
|
||
"template": pod.metadata.labels.get("template", "unknown"),
|
||
"created_at": pod.metadata.creation_timestamp.isoformat() if pod.metadata.creation_timestamp else None,
|
||
"pod_ip": pod.status.pod_ip
|
||
}
|
||
for pod in pods.items
|
||
]
|
||
except ApiException as e:
|
||
logger.error(f"列出Pod失败: {e}")
|
||
raise Exception(f"列出Pod失败: {e.reason}")
|