Gke Log Server

GKE记录MCP服务器

一个生产就绪的模型上下文协议（MCP）服务器，将谷歌Kubernetes引擎日志暴露给克劳德等人工智能助手。

特性

• 查询GKE日志 按集群、命名空间、pod、容器
• 灵活的过滤 按严重程度、时间范围和搜索文本
• 准确的集群发现 使用GKE容器API（不是日志采样）
• 命名空间发现 来自带缓存的云日志
• 生产准备就绪 通过健康检查、安全强化和Kubernetes清单
• 工作负载标识 支持安全的GCP身份验证
• 内置缓存 用于具有可配置TTL的集群和命名空间列表
• 超时处理 具有可配置的请求超时
• 重试逻辑 API瞬时故障指数退避
• 输入验证 防止过滤器注入攻击

暴露的工具

快速开始

先决条件

• Python 3.11+
• 谷歌云SDK(gcloud)
• Docker（用于容器化部署）
• 具有GKE集群的GCP项目

本地开发

1. 克隆和安装依赖关系：

   cd gke-logs-mcp
   python -m venv .venv
   source .venv/bin/activate
   pip install -r requirements.txt

1. 使用GCP进行身份验证：

   gcloud auth application-default login
   export GCP_PROJECT_ID="your-project-id"

1. 运行服务器：

   python -m gke_logs_mcp.server

Claude桌面配置

添加到您的 claude_desktop_config.json:

{
  "mcpServers": {
    "gke-logs": {
      "command": "python",
      "args": ["-m", "gke_logs_mcp.server"],
      "env": {
        "GCP_PROJECT_ID": "your-project-id"
      }
    }
  }
}

测试

该项目包括一个全面的测试套件，涵盖服务器、客户端和健康检查组件。

运行测试

# Install dev dependencies
pip install -e ".[dev]"

# Run all tests
pytest

# Run with verbose output
pytest -v

# Run with coverage report
pytest --cov=gke_logs_mcp --cov=healthcheck --cov-report=term-missing

# Run specific test file
pytest tests/test_server.py

# Run specific test class
pytest tests/test_server.py::TestGKELogsClient

# Run specific test
pytest tests/test_server.py::TestValidateResourceName::test_valid_simple_name

测试结构

tests/
├── __init__.py
├── conftest.py          # Shared fixtures and mocks
├── test_server.py       # Server and client tests
└── test_healthcheck.py  # Health check endpoint tests

测试类别

• 单元测试：测试单个功能，如 validate_resource_name, escape_filter_string
• 客户端测试：测试 GKELogsClient 过滤器构建、条目格式化、缓存
• 异步测试：测试异步方法，如 get_logs, list_clusters, list_namespaces
• MCP处理器测试：测试工具注册和工具调用处理
• 健康检查测试：测试HTTP端点和子流程生命周期

写作测试

测试使用pytest和以下插件：

• pytest-asyncio：用于测试异步代码
• pytest-mock：用于模拟依赖关系
• pytest-cov：用于报道

示例测试：

import pytest
from gke_logs_mcp.server import validate_resource_name, ValidationError

def test_valid_cluster_name():
    assert validate_resource_name("my-cluster", "cluster_name") == "my-cluster"

def test_invalid_cluster_name():
    with pytest.raises(ValidationError):
        validate_resource_name('invalid"name', "cluster_name")

@pytest.mark.asyncio
async def test_list_clusters(logs_client, mock_cloud_logging_client):
    clusters = await logs_client.list_clusters()
    assert "test-cluster" in clusters

代码质量

# Install dev dependencies
pip install -e ".[dev]"

# Format code
black gke_logs_mcp/ tests/

# Lint code
ruff check gke_logs_mcp/ tests/

# Type checking
mypy gke_logs_mcp/

生产部署

1.设置GCP资源

export GCP_PROJECT_ID="your-project-id"
export GKE_CLUSTER_NAME="your-cluster"

chmod +x scripts/setup-gcp.sh
./scripts/setup-gcp.sh

2.构建和推送容器

docker build -t gcr.io/$GCP_PROJECT_ID/gke-logs-mcp:latest .
docker push gcr.io/$GCP_PROJECT_ID/gke-logs-mcp:latest

3.部署到Kubernetes

sed -i "s/YOUR_PROJECT_ID/$GCP_PROJECT_ID/g" k8s/deployment.yaml
kubectl apply -f k8s/deployment.yaml

环境变量

使用示例

一旦连接到克劳德：

“我们有什么集群？”

“显示最近一小时生产集群默认命名空间中的错误”

在api网关Pod中搜索“连接被拒绝”

建筑

服务器组件

• GKELogsConfig：用于配置和验证的Pydantic模型
• GKELogsClient：通过缓存和超时处理来包装云日志和GKE容器API
• create_server()：用于依赖注入和可测试性的工厂函数
• MCP工具处理器：通过装修商注册 list_tools 和 call_tool

主要特点

• 输入验证：根据K8s命名规则验证所有资源名称
• 过滤器逃逸：过滤字符串中引号和反斜杠的正确转义
• 缓存：用于集群和命名空间列表的基于TTL的缓存
• 流媒体：日志条目的内存高效迭代
• 超时：可配置超时 asyncio.wait_for
• 重试：瞬态的指数回退重试 ServiceUnavailable 错误
• 吉凯恩集装箱API：通过以下方式准确列出集群 container_v1.ClusterManagerClient

安全

• 使用工作负载标识进行GCP身份验证（生产中没有密钥）
• 最低IAM权限（只读）：

- roles/logging.viewer -从Cloud Logging读取日志 - roles/container.clusterViewer -列出GKE集群

• 以非root用户身份运行
• 包括出口/入口控制的网络策略
• 输入验证可防止过滤器注入攻击

许可证

麻省理工学院