zhanggangyong
33955b68dc
- 修改 template_manager.py、k8s_manager.py 中的端口映射 - 更新 jina_search_agent、azure_blob_agent 系列、a2a_litellm_agent 的代码和 Dockerfile 为 8000 - 添加端口修改脚本和测试脚本 Made-with: Cursor
1307 lines
53 KiB
Python
1307 lines
53 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
|
||
import re
|
||
import requests
|
||
import time
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def sanitize_k8s_name(name: str, max_length: int = 63) -> str:
|
||
"""将名称转换为 DNS-1035 合规格式
|
||
|
||
Kubernetes Service、Deployment 等资源名称必须符合 DNS-1035 标准：
|
||
- 只能包含小写字母、数字和连字符 '-'
|
||
- 必须以字母开头
|
||
- 必须以字母或数字结尾
|
||
- 最长 63 个字符
|
||
|
||
Args:
|
||
name: 原始名称
|
||
max_length: 最大长度（默认 63）
|
||
|
||
Returns:
|
||
合规的 K8s 资源名称
|
||
"""
|
||
# 转小写
|
||
name = name.lower()
|
||
# 将下划线和空格替换为连字符
|
||
name = name.replace("_", "-").replace(" ", "-")
|
||
# 移除非字母、数字、连字符的字符
|
||
name = re.sub(r'[^a-z0-9-]', '', name)
|
||
# 合并连续的连字符
|
||
name = re.sub(r'-+', '-', name)
|
||
# 如果以数字开头，添加 'a' 前缀
|
||
if name and name[0].isdigit():
|
||
name = 'a' + name
|
||
# 如果以连字符开头，去掉
|
||
name = name.lstrip('-')
|
||
# 截断到最大长度
|
||
name = name[:max_length]
|
||
# 去掉末尾的连字符
|
||
name = name.rstrip('-')
|
||
# 最终兜底：如果名称为空
|
||
if not name:
|
||
name = 'agent'
|
||
return name
|
||
|
||
|
||
class K8sManager:
|
||
"""Kubernetes资源管理器"""
|
||
|
||
# Azure DNS 配置（从环境变量读取）
|
||
AZURE_TENANT_ID = os.getenv("AZURE_TENANT_ID", "263c3ff6-1be5-4141-8308-b188464fb297")
|
||
AZURE_CLIENT_ID = os.getenv("AZURE_CLIENT_ID", "c5ba26db-f180-425f-bac3-93708d853988")
|
||
AZURE_CLIENT_SECRET = os.getenv("AZURE_CLIENT_SECRET", "ydt8Q~DIOfKXDUOJkzfgaNVNh2d1VrbeIaYuKbpD")
|
||
AZURE_SUBSCRIPTION_ID = os.getenv("AZURE_SUBSCRIPTION_ID", "c6c47e4c-f5f4-49f8-b26f-7728862c17d6")
|
||
AZURE_RESOURCE_GROUP = os.getenv("AZURE_RESOURCE_GROUP", "taiji-ai-v0")
|
||
AZURE_DNS_ZONE = os.getenv("AZURE_DNS_ZONE", "taijiagnet.com")
|
||
|
||
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.networking_v1 = client.NetworkingV1Api()
|
||
|
||
# 确保命名空间存在
|
||
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
|
||
|
||
def create_agent_namespace(self, agent_name: str, owner_id: str = None) -> str:
|
||
"""为 Agent 创建独立的命名空间
|
||
|
||
Args:
|
||
agent_name: Agent 名称
|
||
owner_id: 所有者 ID
|
||
|
||
Returns:
|
||
创建的命名空间名称
|
||
"""
|
||
# 生成命名空间名称（使用 agent-{sanitized_name} 格式，确保 DNS 合规）
|
||
sanitized_name = sanitize_k8s_name(agent_name)
|
||
namespace_name = f"agent-{sanitized_name}"[:63].rstrip('-')
|
||
|
||
try:
|
||
# 检查命名空间是否已存在
|
||
self.v1.read_namespace(name=namespace_name)
|
||
logger.info(f"命名空间 {namespace_name} 已存在")
|
||
return namespace_name
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
# 创建命名空间
|
||
labels = {
|
||
"managed-by": "agent-manager",
|
||
"agent-name": agent_name
|
||
}
|
||
if owner_id:
|
||
labels["owner-id"] = owner_id
|
||
|
||
namespace_manifest = client.V1Namespace(
|
||
metadata=client.V1ObjectMeta(
|
||
name=namespace_name,
|
||
labels=labels
|
||
)
|
||
)
|
||
self.v1.create_namespace(body=namespace_manifest)
|
||
logger.info(f"✅ 创建命名空间 {namespace_name}")
|
||
|
||
# 复制 ACR secret 到新命名空间
|
||
self._copy_acr_secret_to_namespace(namespace_name)
|
||
|
||
return namespace_name
|
||
else:
|
||
raise
|
||
|
||
def _copy_acr_secret_to_namespace(self, target_namespace: str):
|
||
"""复制 ACR secret 到目标命名空间
|
||
|
||
Args:
|
||
target_namespace: 目标命名空间
|
||
"""
|
||
try:
|
||
# 从 agent-manager 命名空间读取 acr-secret
|
||
source_secret = self.v1.read_namespaced_secret(
|
||
name="acr-secret",
|
||
namespace="agent-manager"
|
||
)
|
||
|
||
# 创建新的 secret（去除自动生成的字段）
|
||
new_secret = client.V1Secret(
|
||
metadata=client.V1ObjectMeta(
|
||
name="acr-secret",
|
||
namespace=target_namespace
|
||
),
|
||
data=source_secret.data,
|
||
type=source_secret.type
|
||
)
|
||
|
||
# 在目标命名空间创建 secret
|
||
self.v1.create_namespaced_secret(
|
||
namespace=target_namespace,
|
||
body=new_secret
|
||
)
|
||
logger.info(f"✅ 已复制 ACR secret 到命名空间 {target_namespace}")
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
logger.warning(f"⚠️ 源 ACR secret 不存在，跳过复制")
|
||
elif e.status == 409:
|
||
logger.info(f"ACR secret 已存在于命名空间 {target_namespace}")
|
||
else:
|
||
logger.error(f"❌ 复制 ACR secret 失败: {e}")
|
||
|
||
def create_service(self, service_name: str, namespace: str, pod_selector: Dict[str, str],
|
||
service_port: int, target_port: int) -> Dict:
|
||
"""为 Agent Pod 创建 LoadBalancer Service
|
||
|
||
Args:
|
||
service_name: Service 名称
|
||
namespace: 命名空间
|
||
pod_selector: Pod 选择器标签
|
||
service_port: Service 端口
|
||
target_port: Pod 目标端口
|
||
|
||
Returns:
|
||
创建的 Service 信息（包含外网 IP）
|
||
"""
|
||
try:
|
||
service_manifest = client.V1Service(
|
||
metadata=client.V1ObjectMeta(
|
||
name=service_name,
|
||
namespace=namespace,
|
||
labels={"managed-by": "agent-manager"}
|
||
),
|
||
spec=client.V1ServiceSpec(
|
||
selector=pod_selector,
|
||
ports=[
|
||
client.V1ServicePort(
|
||
name="http",
|
||
protocol="TCP",
|
||
port=service_port,
|
||
target_port=target_port
|
||
)
|
||
],
|
||
type="LoadBalancer"
|
||
)
|
||
)
|
||
|
||
response = self.v1.create_namespaced_service(
|
||
namespace=namespace,
|
||
body=service_manifest
|
||
)
|
||
|
||
logger.info(f"✅ LoadBalancer Service {service_name} 创建成功 (namespace: {namespace})")
|
||
|
||
# 获取外网 IP（可能需要等待分配）
|
||
external_ip = None
|
||
if response.status.load_balancer.ingress:
|
||
lb_ingress = response.status.load_balancer.ingress[0]
|
||
external_ip = lb_ingress.ip or lb_ingress.hostname
|
||
|
||
return {
|
||
"name": response.metadata.name,
|
||
"namespace": response.metadata.namespace,
|
||
"cluster_ip": response.spec.cluster_ip,
|
||
"port": service_port,
|
||
"external_ip": external_ip,
|
||
"type": "LoadBalancer"
|
||
}
|
||
except ApiException as e:
|
||
logger.error(f"创建 Service 失败: {e}")
|
||
raise Exception(f"创建 Service 失败: {e.reason}")
|
||
|
||
def create_ingress(self, ingress_name: str, namespace: str, service_name: str,
|
||
service_port: int, host: str = None, path: str = "/") -> Dict:
|
||
"""为 Agent Service 创建 Ingress
|
||
|
||
Args:
|
||
ingress_name: Ingress 名称
|
||
namespace: 命名空间
|
||
service_name: 后端 Service 名称
|
||
service_port: Service 端口
|
||
host: 域名（可选，如果为 None 则使用默认）
|
||
path: 路径前缀
|
||
|
||
Returns:
|
||
创建的 Ingress 信息
|
||
"""
|
||
try:
|
||
# 构建路径规则
|
||
http_ingress_path = client.V1HTTPIngressPath(
|
||
path=path,
|
||
path_type="Prefix",
|
||
backend=client.V1IngressBackend(
|
||
service=client.V1IngressServiceBackend(
|
||
name=service_name,
|
||
port=client.V1ServiceBackendPort(number=service_port)
|
||
)
|
||
)
|
||
)
|
||
|
||
# 构建规则
|
||
ingress_rule = client.V1IngressRule(
|
||
http=client.V1HTTPIngressRuleValue(paths=[http_ingress_path])
|
||
)
|
||
|
||
# 如果指定了 host，添加到规则中
|
||
if host:
|
||
ingress_rule.host = host
|
||
|
||
# 创建 Ingress manifest
|
||
ingress_manifest = client.V1Ingress(
|
||
metadata=client.V1ObjectMeta(
|
||
name=ingress_name,
|
||
namespace=namespace,
|
||
labels={"managed-by": "agent-manager"},
|
||
annotations={
|
||
"nginx.ingress.kubernetes.io/rewrite-target": "/",
|
||
"nginx.ingress.kubernetes.io/ssl-redirect": "false"
|
||
}
|
||
),
|
||
spec=client.V1IngressSpec(
|
||
ingress_class_name="nginx", # 使用 nginx ingress controller
|
||
rules=[ingress_rule]
|
||
)
|
||
)
|
||
|
||
response = self.networking_v1.create_namespaced_ingress(
|
||
namespace=namespace,
|
||
body=ingress_manifest
|
||
)
|
||
|
||
logger.info(f"✅ Ingress {ingress_name} 创建成功 (namespace: {namespace})")
|
||
|
||
# 获取 Ingress IP/域名
|
||
ingress_url = None
|
||
if response.status.load_balancer.ingress:
|
||
lb_ingress = response.status.load_balancer.ingress[0]
|
||
if lb_ingress.ip:
|
||
ingress_url = f"http://{lb_ingress.ip}{path}"
|
||
elif lb_ingress.hostname:
|
||
ingress_url = f"http://{lb_ingress.hostname}{path}"
|
||
|
||
return {
|
||
"name": response.metadata.name,
|
||
"namespace": response.metadata.namespace,
|
||
"host": host,
|
||
"path": path,
|
||
"url": ingress_url,
|
||
"note": "Ingress URL 将在负载均衡器配置完成后可用"
|
||
}
|
||
except ApiException as e:
|
||
logger.error(f"创建 Ingress 失败: {e}")
|
||
raise Exception(f"创建 Ingress 失败: {e.reason}")
|
||
|
||
# 模板端口映射
|
||
TEMPLATE_PORTS = {
|
||
"echo_agent": 8000,
|
||
"search_agent": 8080,
|
||
"search_agent_a2a": 8080,
|
||
"search_agent_mcp": 8080,
|
||
"mysql_agent": 8000,
|
||
"postgresql_agent": 8000,
|
||
"jina_search_agent": 8000,
|
||
"azure_blob_agent": 8000,
|
||
"azure_blob_agent_mcp": 8000,
|
||
"azure_blob_agent_a2a": 8000,
|
||
"a2a_litellm_agent": 8000,
|
||
"code_ai_agent": 8000,
|
||
"facebook_agent": 8000,
|
||
"media_downloader": 8000,
|
||
"content_analyzer": 8000,
|
||
"huoke": 8000,
|
||
"microsoft_learn_agent": 8000,
|
||
"aws_docs_mcp": 8000,
|
||
"google_mcp": 8000,
|
||
}
|
||
|
||
# 模板所需环境变量说明
|
||
TEMPLATE_ENV_INFO = {
|
||
"jina_search_agent": {
|
||
"required": {
|
||
"JINA_API_KEY": "Jina API密钥，从 https://jina.ai/ 获取"
|
||
},
|
||
"optional": {
|
||
"SERVICE_PORT": "HTTP服务端口，默认8000",
|
||
"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"
|
||
}
|
||
},
|
||
"azure_blob_agent": {
|
||
"required": {
|
||
"LITELLM_API_BASE": "LiteLLM服务地址，如 http://litellm-service:4000",
|
||
"LITELLM_MODEL": "使用的LLM模型，如 gpt-3.5-turbo",
|
||
"LITELLM_API_KEY": "LiteLLM API密钥"
|
||
},
|
||
"optional": {
|
||
"AZURE_STORAGE_CONNECTION_STRING": "Azure Storage连接字符串（可选，也可通过 /connect API 动态传入）",
|
||
"SERVICE_PORT": "HTTP服务端口，默认8000",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认0.0.0.0"
|
||
}
|
||
},
|
||
"azure_blob_agent_mcp": {
|
||
"required": {
|
||
"MODEL_PROVIDER": "模型提供商，如 openai, azure-openai",
|
||
"MODEL_NAME": "使用的模型名称，如 gpt-4",
|
||
"MODEL_API_KEY": "模型 API 密钥"
|
||
},
|
||
"optional": {
|
||
"AZURE_STORAGE_CONNECTION_STRING": "Azure Storage连接字符串",
|
||
"TOOLS_CONFIG": "工具配置 JSON",
|
||
"TOOL_ENDPOINT": "外部工具端点",
|
||
"TOOL_API_KEY": "工具 API 密钥",
|
||
"MODEL_ENDPOINT": "模型 API 端点",
|
||
"STORAGE_ACCOUNT_NAME": "存储账户名称",
|
||
"USER_ID": "用户标识",
|
||
"TENANT_ID": "租户标识",
|
||
"NAMESPACE": "Kubernetes 命名空间"
|
||
}
|
||
},
|
||
"azure_blob_agent_a2a": {
|
||
"required": {
|
||
"MODEL_PROVIDER": "模型提供商，如 openai, azure-openai",
|
||
"MODEL_NAME": "使用的模型名称，如 gpt-4",
|
||
"MODEL_API_KEY": "模型 API 密钥",
|
||
"AGENT_ID": "Agent 唯一标识",
|
||
"AGENT_ROLE": "Agent 角色"
|
||
},
|
||
"optional": {
|
||
"AZURE_STORAGE_CONNECTION_STRING": "Azure Storage连接字符串",
|
||
"TOOLS_CONFIG": "工具配置 JSON",
|
||
"TOOL_ENDPOINT": "外部工具端点",
|
||
"TOOL_API_KEY": "工具 API 密钥",
|
||
"MODEL_ENDPOINT": "模型 API 端点",
|
||
"STORAGE_ACCOUNT_NAME": "存储账户名称",
|
||
"AGENT_CAPABILITIES": "Agent 能力列表 JSON",
|
||
"USER_ID": "用户标识",
|
||
"TENANT_ID": "租户标识",
|
||
"NAMESPACE": "Kubernetes 命名空间"
|
||
}
|
||
},
|
||
"search_agent": {
|
||
"required": {
|
||
"LLM_BASE_URL": "LLM 服务地址，如 https://api.openai.com/v1",
|
||
"SERPER_API_KEY": "Serper 搜索 API 密钥，从 https://serper.dev 获取",
|
||
"JINA_API_KEY": "Jina Reader API 密钥，从 https://jina.ai 获取"
|
||
},
|
||
"optional": {
|
||
"LLM_API_KEY": "LLM API 密钥（可在搜索请求中传入）",
|
||
"MODEL_NAME": "LLM 模型名称，默认 gpt-4o-mini",
|
||
"MAX_ITERATIONS": "最大搜索迭代次数，默认 3",
|
||
"MAX_RESULTS_PER_QUERY": "每次搜索最大结果数，默认 10",
|
||
"CONTENT_MAX_LENGTH": "内容最大长度，默认 5000",
|
||
"TIMEOUT": "超时时间（秒），默认 30",
|
||
"SERVICE_PORT": "HTTP服务端口，默认 8080",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认 0.0.0.0"
|
||
}
|
||
},
|
||
"search_agent_a2a": {
|
||
"required": {
|
||
"LLM_BASE_URL": "LLM 服务地址，如 https://api.openai.com/v1",
|
||
},
|
||
"optional": {
|
||
"LITELLM_API_KEY": "LiteLLM API 密钥（可在请求中传入）",
|
||
"LLM_API_KEY": "LLM API 密钥（备选，可在请求中传入）",
|
||
"MODEL_NAME": "LLM 模型名称（优先）",
|
||
"LLM_MODEL": "LLM 模型名称（备选）",
|
||
"LITELLM_MODEL": "LiteLLM 模型名称（备选）",
|
||
"SERPER_API_KEY": "Serper 搜索 API 密钥（已内置默认值）",
|
||
"JINA_API_KEY": "Jina Reader API 密钥（已内置默认值）",
|
||
"SERVICE_PORT": "HTTP服务端口，默认 8080",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认 0.0.0.0"
|
||
},
|
||
"description": "A2A 协议搜索 Agent，支持通过请求动态传入 API key"
|
||
},
|
||
"search_agent_mcp": {
|
||
"required": {
|
||
"LLM_BASE_URL": "LLM 服务地址，如 https://api.openai.com/v1",
|
||
},
|
||
"optional": {
|
||
"LITELLM_API_KEY": "LiteLLM API 密钥（可在请求中传入）",
|
||
"LLM_API_KEY": "LLM API 密钥（备选，可在请求中传入）",
|
||
"MODEL_NAME": "LLM 模型名称（优先）",
|
||
"LLM_MODEL": "LLM 模型名称（备选）",
|
||
"LITELLM_MODEL": "LiteLLM 模型名称（备选）",
|
||
"SERPER_API_KEY": "Serper 搜索 API 密钥（已内置默认值）",
|
||
"JINA_API_KEY": "Jina Reader API 密钥（已内置默认值）",
|
||
"SERVICE_PORT": "HTTP服务端口，默认 8080",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认 0.0.0.0"
|
||
},
|
||
"description": "MCP 协议搜索 Agent，支持通过请求动态传入 API key"
|
||
},
|
||
"a2a_litellm_agent": {
|
||
"required": {
|
||
"LITELLM_API_BASE": "LiteLLM 服务地址",
|
||
"LITELLM_MODEL": "LiteLLM 模型名称"
|
||
},
|
||
"optional": {
|
||
"LITELLM_API_KEY": "LiteLLM API 密钥（可在请求中传入）",
|
||
"AGENT_NAME": "Agent 名称",
|
||
"AGENT_DESCRIPTION": "Agent 描述",
|
||
"SERVICE_PORT": "HTTP服务端口，默认 8000",
|
||
"SERVICE_HOST": "HTTP服务监听地址，默认 0.0.0.0"
|
||
}
|
||
},
|
||
"code_ai_agent": {
|
||
"required": {
|
||
"LLM_BASE_URL": "LLM 服务地址，如 https://api.openai.com/v1",
|
||
},
|
||
"optional": {
|
||
"LLM_API_KEY": "LLM API 密钥（可在请求中传入）",
|
||
"MODEL_NAME": "LLM 模型名称",
|
||
"API_HOST": "API 服务监听地址，默认 0.0.0.0",
|
||
"API_PORT": "API 服务端口，默认 8000",
|
||
"PROJECTS_DIR": "项目存储目录，默认 /tmp/projects"
|
||
},
|
||
"description": "代码助手 Agent，支持代码生成、分析和执行"
|
||
},
|
||
"facebook_agent": {
|
||
"required": {
|
||
"LITELLM_GATEWAY_URL": "LiteLLM Gateway 服务地址，如 https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/",
|
||
"FACEBOOK_API_KEY": "Facebook RapidAPI 密钥"
|
||
},
|
||
"optional": {
|
||
"LITELLM_API_KEY": "LiteLLM API 密钥（可在请求中传入）",
|
||
"LITELLM_MODEL": "LiteLLM 模型名称，默认 taiji/gpt-4o-mini",
|
||
"FACEBOOK_API_HOST": "Facebook API 主机，默认 facebook-scraper3.p.rapidapi.com",
|
||
"FACEBOOK_MCP_URL": "Facebook MCP URL，默认 https://mcp.rapidapi.com",
|
||
"MAX_RESULTS": "最大搜索结果数，默认 10",
|
||
"TIMEOUT": "超时时间（秒），默认 30",
|
||
"LOG_LEVEL": "日志级别，默认 INFO"
|
||
},
|
||
"description": "Facebook 搜索智能 Agent，支持 Facebook 内容搜索和 MCP 协议"
|
||
}
|
||
}
|
||
|
||
def get_template_info(self, template: str) -> Dict:
|
||
"""
|
||
获取模板信息（从数据库动态获取）
|
||
|
||
Args:
|
||
template: 模板类型
|
||
|
||
Returns:
|
||
模板信息（端口、所需环境变量等）
|
||
"""
|
||
from template_manager import template_manager as tm
|
||
template_data = tm.get_template(template)
|
||
|
||
if template_data:
|
||
return {
|
||
"template": template,
|
||
"name": template_data.get("name"),
|
||
"display_name": template_data.get("display_name"),
|
||
"description": template_data.get("description"),
|
||
"port": template_data.get("port", 8000),
|
||
"image": template_data.get("image"),
|
||
"agent_framework": template_data.get("agent_framework"),
|
||
"env_requirements": template_data.get("env_requirements", {}),
|
||
# 回退到硬编码的环境变量信息（兼容旧模板）
|
||
"env_info": self.TEMPLATE_ENV_INFO.get(template, template_data.get("env_requirements", {}))
|
||
}
|
||
|
||
# 回退到硬编码配置（兼容性）
|
||
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} 创建成功")
|
||
|
||
# 获取服务端口（从数据库动态获取）
|
||
from template_manager import template_manager as tm
|
||
service_port = tm.get_port(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")
|
||
|
||
# 从数据库动态获取模板镜像（延迟导入避免循环依赖）
|
||
from template_manager import template_manager as tm
|
||
image = tm.get_image(template)
|
||
|
||
# 构建环境变量列表
|
||
env_vars = [
|
||
client.V1EnvVar(name="POD_NAME", value=pod_name),
|
||
client.V1EnvVar(name="TEMPLATE_TYPE", value=template)
|
||
]
|
||
|
||
# NEW: 添加 Agent 框架配置
|
||
agent_framework = config_data.get("agent_framework", "langchain")
|
||
env_vars.append(client.V1EnvVar(name="AGENT_FRAMEWORK", value=agent_framework))
|
||
|
||
# NEW: 添加工具配置
|
||
if "tools_config" in config_data:
|
||
import json
|
||
env_vars.append(client.V1EnvVar(
|
||
name="TOOLS_CONFIG",
|
||
value=json.dumps(config_data["tools_config"])
|
||
))
|
||
|
||
if "tool_endpoint" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="TOOL_ENDPOINT", value=config_data["tool_endpoint"]))
|
||
|
||
if "tool_api_key" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="TOOL_API_KEY", value=config_data["tool_api_key"]))
|
||
|
||
# NEW: 添加模型配置
|
||
if "model_provider" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="MODEL_PROVIDER", value=config_data["model_provider"]))
|
||
|
||
if "model_name" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="MODEL_NAME", value=config_data["model_name"]))
|
||
|
||
if "model_endpoint" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="MODEL_ENDPOINT", value=config_data["model_endpoint"]))
|
||
|
||
if "model_api_key" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="MODEL_API_KEY", value=config_data["model_api_key"]))
|
||
|
||
# NEW: 添加存储配置
|
||
if "storage_connection_string" in config_data:
|
||
env_vars.append(client.V1EnvVar(
|
||
name="AZURE_STORAGE_CONNECTION_STRING",
|
||
value=config_data["storage_connection_string"]
|
||
))
|
||
|
||
if "storage_account_name" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="STORAGE_ACCOUNT_NAME", value=config_data["storage_account_name"]))
|
||
|
||
# NEW: 添加用户标识
|
||
if "user_id" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="USER_ID", value=config_data["user_id"]))
|
||
|
||
if "tenant_id" in config_data:
|
||
env_vars.append(client.V1EnvVar(name="TENANT_ID", value=config_data["tenant_id"]))
|
||
|
||
# NEW: 添加命名空间信息
|
||
namespace = config_data.get("namespace", self.namespace)
|
||
env_vars.append(client.V1EnvVar(name="NAMESPACE", value=namespace))
|
||
|
||
# 为特定模板添加默认环境变量
|
||
template_defaults = {
|
||
"facebook_agent": {
|
||
"LITELLM_GATEWAY_URL": "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io",
|
||
"LITELLM_MODEL": "taiji/gpt-4o-mini",
|
||
"FACEBOOK_API_KEY": "34225c5924msh453fa7aff7a52a9p1d7adfjsn260ba0796c00"
|
||
},
|
||
"code_ai_agent": {
|
||
"OPENAI_BASE_URL": "https://litellm.graystone-fb459c5d.southeastasia.azurecontainerapps.io/v1",
|
||
"LITELLM_MODEL": "taiji/gpt-4o-mini"
|
||
}
|
||
}
|
||
|
||
# 获取用户自定义环境变量
|
||
custom_env = config_data.get("env", {})
|
||
|
||
# 如果模板有默认值，先添加默认值（用户自定义值会覆盖）
|
||
if template in template_defaults:
|
||
defaults = template_defaults[template]
|
||
for key, value in defaults.items():
|
||
# 只有当用户没有提供该环境变量时才使用默认值
|
||
if key not in custom_env:
|
||
env_vars.append(client.V1EnvVar(name=key, value=str(value)))
|
||
logger.info(f"使用默认环境变量: {key}")
|
||
|
||
# 添加用户自定义环境变量（会覆盖默认值）
|
||
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 ["search_agent", "search_agent_a2a", "search_agent_mcp"]:
|
||
container_ports = [client.V1ContainerPort(container_port=8080)]
|
||
elif template in ["jina_search_agent", "azure_blob_agent", "azure_blob_agent_mcp", "azure_blob_agent_a2a", "a2a_litellm_agent", "code_ai_agent", "facebook_agent", "echo_agent", "mysql_agent", "postgresql_agent"]:
|
||
container_ports = [client.V1ContainerPort(container_port=8000)]
|
||
|
||
# 创建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",
|
||
"framework": agent_framework # NEW: 添加框架标签
|
||
}
|
||
# 添加用户自定义标签
|
||
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 delete_agent_namespace(self, agent_name: str) -> Dict:
|
||
"""删除Agent的命名空间及其所有资源
|
||
|
||
Args:
|
||
agent_name: Agent 名称
|
||
|
||
Returns:
|
||
删除结果
|
||
"""
|
||
sanitized_name = sanitize_k8s_name(agent_name)
|
||
namespace_name = f"agent-{sanitized_name}"[:63].rstrip('-')
|
||
# region agent log
|
||
try:
|
||
import json, time
|
||
with open("/home/taiji/tools/agent-manager/.cursor/debug.log", "a") as _f:
|
||
_f.write(json.dumps({
|
||
"sessionId": "debug-session",
|
||
"runId": "pre-fix",
|
||
"hypothesisId": "H1",
|
||
"location": "k8s_manager.py:delete_agent_namespace:entry",
|
||
"message": "delete_agent_namespace computed namespace",
|
||
"data": {
|
||
"agent_name": agent_name,
|
||
"namespace_name": namespace_name,
|
||
"is_manager_namespace": namespace_name == "agent-manager"
|
||
},
|
||
"timestamp": int(time.time() * 1000)
|
||
}) + "\n")
|
||
except Exception:
|
||
pass
|
||
# endregion
|
||
|
||
# 保护机制：防止删除 agent-manager 命名空间
|
||
if namespace_name == "agent-manager":
|
||
logger.error(f"❌ 禁止删除 agent-manager 命名空间！agent_name={agent_name}, namespace_name={namespace_name}")
|
||
raise Exception(f"禁止删除 agent-manager 命名空间。这是系统保护命名空间，不能被删除。")
|
||
|
||
try:
|
||
# 删除命名空间会自动删除其中的所有资源（Pod、Service、Ingress等）
|
||
self.v1.delete_namespace(
|
||
name=namespace_name,
|
||
body=client.V1DeleteOptions()
|
||
)
|
||
logger.info(f"✅ 命名空间 {namespace_name} 及其所有资源删除成功")
|
||
return {
|
||
"status": "success",
|
||
"message": f"Agent {agent_name} 的命名空间 {namespace_name} 及所有相关资源已删除",
|
||
"namespace": namespace_name
|
||
}
|
||
except ApiException as e:
|
||
if e.status == 404:
|
||
return {
|
||
"status": "not_found",
|
||
"message": f"命名空间 {namespace_name} 不存在"
|
||
}
|
||
logger.error(f"删除命名空间失败: {e}")
|
||
raise Exception(f"删除命名空间失败: {e.reason}")
|
||
|
||
def _get_azure_access_token(self) -> str:
|
||
"""获取 Azure Access Token
|
||
|
||
Returns:
|
||
Azure 访问令牌
|
||
"""
|
||
try:
|
||
token_url = f"https://login.microsoftonline.com/{self.AZURE_TENANT_ID}/oauth2/v2.0/token"
|
||
data = {
|
||
"client_id": self.AZURE_CLIENT_ID,
|
||
"client_secret": self.AZURE_CLIENT_SECRET,
|
||
"grant_type": "client_credentials",
|
||
"scope": "https://management.azure.com/.default"
|
||
}
|
||
|
||
response = requests.post(token_url, data=data, timeout=10)
|
||
response.raise_for_status()
|
||
|
||
token = response.json().get('access_token')
|
||
if not token:
|
||
raise Exception("未能获取访问令牌")
|
||
|
||
return token
|
||
except Exception as e:
|
||
logger.error(f"获取 Azure 访问令牌失败: {e}")
|
||
raise
|
||
|
||
def create_dns_record(self, subdomain: str, ip_address: str, ttl: int = 300) -> Dict:
|
||
"""为 IP 地址创建 Azure DNS A 记录
|
||
|
||
Args:
|
||
subdomain: 子域名（不包括主域名）
|
||
ip_address: 要指向的 IP 地址
|
||
ttl: DNS TTL（秒），默认 300
|
||
|
||
Returns:
|
||
DNS 记录信息
|
||
"""
|
||
try:
|
||
# 获取访问令牌
|
||
token = self._get_azure_access_token()
|
||
|
||
# 构建 DNS API URL
|
||
dns_url = (
|
||
f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/"
|
||
f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/"
|
||
f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01"
|
||
)
|
||
|
||
# DNS 记录数据
|
||
dns_data = {
|
||
"properties": {
|
||
"TTL": ttl,
|
||
"ARecords": [{"ipv4Address": ip_address}]
|
||
}
|
||
}
|
||
|
||
# 创建 DNS 记录
|
||
headers = {
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
response = requests.put(dns_url, json=dns_data, headers=headers, timeout=30)
|
||
response.raise_for_status()
|
||
|
||
full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}"
|
||
logger.info(f"✅ DNS 记录创建成功: {full_domain} -> {ip_address}")
|
||
|
||
return {
|
||
"subdomain": subdomain,
|
||
"domain": full_domain,
|
||
"ip_address": ip_address,
|
||
"ttl": ttl,
|
||
"status": "created"
|
||
}
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"创建 DNS 记录失败: {e}")
|
||
if hasattr(e.response, 'text'):
|
||
logger.error(f"错误详情: {e.response.text}")
|
||
raise Exception(f"创建 DNS 记录失败: {str(e)}")
|
||
|
||
def delete_dns_record(self, subdomain: str) -> Dict:
|
||
"""删除 Azure DNS A 记录
|
||
|
||
Args:
|
||
subdomain: 要删除的子域名
|
||
|
||
Returns:
|
||
删除结果
|
||
"""
|
||
try:
|
||
# 获取访问令牌
|
||
token = self._get_azure_access_token()
|
||
|
||
# 构建 DNS API URL
|
||
dns_url = (
|
||
f"https://management.azure.com/subscriptions/{self.AZURE_SUBSCRIPTION_ID}/"
|
||
f"resourceGroups/{self.AZURE_RESOURCE_GROUP}/providers/Microsoft.Network/"
|
||
f"dnsZones/{self.AZURE_DNS_ZONE}/A/{subdomain}?api-version=2018-05-01"
|
||
)
|
||
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
|
||
response = requests.delete(dns_url, headers=headers, timeout=30)
|
||
|
||
if response.status_code == 404:
|
||
logger.info(f"DNS 记录 {subdomain} 不存在，跳过删除")
|
||
return {"status": "not_found", "subdomain": subdomain}
|
||
|
||
response.raise_for_status()
|
||
|
||
full_domain = f"{subdomain}.{self.AZURE_DNS_ZONE}"
|
||
logger.info(f"✅ DNS 记录删除成功: {full_domain}")
|
||
|
||
return {
|
||
"subdomain": subdomain,
|
||
"domain": full_domain,
|
||
"status": "deleted"
|
||
}
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"删除 DNS 记录失败: {e}")
|
||
raise Exception(f"删除 DNS 记录失败: {str(e)}")
|
||
|
||
def wait_for_loadbalancer_ip(self, service_name: str, namespace: str,
|
||
max_wait: int = 300, interval: int = 5) -> str:
|
||
"""等待 LoadBalancer 分配外网 IP
|
||
|
||
Args:
|
||
service_name: Service 名称
|
||
namespace: 命名空间
|
||
max_wait: 最大等待时间（秒）
|
||
interval: 检查间隔（秒）
|
||
|
||
Returns:
|
||
分配的外网 IP 地址
|
||
"""
|
||
logger.info(f"等待 LoadBalancer IP 分配... (最多等待 {max_wait} 秒)")
|
||
|
||
elapsed = 0
|
||
while elapsed < max_wait:
|
||
try:
|
||
service = self.v1.read_namespaced_service(service_name, namespace)
|
||
if service.status.load_balancer.ingress:
|
||
ip = service.status.load_balancer.ingress[0].ip
|
||
if ip:
|
||
logger.info(f"✅ LoadBalancer IP 已分配: {ip}")
|
||
return ip
|
||
except Exception as e:
|
||
logger.warning(f"查询 Service 状态失败: {e}")
|
||
|
||
time.sleep(interval)
|
||
elapsed += interval
|
||
logger.info(f"等待中... ({elapsed}/{max_wait} 秒)")
|
||
|
||
raise Exception(f"LoadBalancer IP 分配超时（{max_wait} 秒）")
|
||
|
||
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}")
|