[MCP学习笔记]MCP服务网格化：Istio控制平面改造

// 导入 MCP 协议相关的库
import (
    mcp "github.com/envoyproxy/go-control-plane/envoy/service/mcp/v3"
    "github.com/golang/protobuf/proto"
    "google.golang.org/grpc"
)

// MCP 客户端结构体
type MCPClient struct {
    conn       *grpc.ClientConn
    stream     mcp.AggregatedDiscoveryService_StreamAggregatedResourcesClient
    node       *core.Node
    resources  map[string]Resource
    cancelFunc context.CancelFunc
}

// 创建 MCP 客户端
func NewMCPClient(addr string, node *core.Node) (*MCPClient, error) {
    conn, err := grpc.Dial(addr, grpc.WithInsecure())
    if err != nil {
        return nil, err
    }

    stream, err := mcp.NewAggregatedDiscoveryServiceClient(conn).StreamAggregatedResources(
        context.Background(),
        grpc.WaitForReady(true),
    )
    if err != nil {
        return nil, err
    }

    client := &MCPClient{
        conn:     conn,
        stream:   stream,
        node:     node,
        resources: make(map[string]Resource),
    }

    return client, nil
}

// 发送发现请求
func (c *MCPClient) sendDiscoveryRequest(resourceTypes []string, versionInfo string) error {
    request := &mcp.DiscoveryRequest{
        Node:          c.node,
        TypeUrl:       resourceTypes,
        VersionInfo:   versionInfo,
        ResponseNonce: "",
        ErrorDetail:   nil,
    }

    err := c.stream.Send(request)
    if err != nil {
        return err
    }

    return nil
}

// 接收发现响应
func (c *MCPClient) receiveDiscoveryResponse() (*mcp.DiscoveryResponse, error) {
    response, err := c.stream.Recv()
    if err != nil {
        return nil, err
    }

    return response, nil
}

// 修改 pilot 包中的 discovery 服务
func (s * DiscoveryService) StreamAggregatedResources(stream mcp.AggregatedDiscoveryService_StreamAggregatedResourcesServer) error {
    node, err := getNodeFromStream(stream)
    if err != nil {
        return err
    }

    client := NewMCPClient(s.configStore, node)
    if client == nil {
        return fmt.Errorf("failed to create MCP client")
    }

    for {
        req, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            return err
        }

        // 处理 MCP 请求
        err = client.handleRequest(req)
        if err != nil {
            return err
        }
    }

    return nil
}

// 处理 MCP 请求
func (c *MCPClient) handleRequest(req *mcp.DiscoveryRequest) error {
    // 根据请求类型获取资源
    resources, err := c.getResourceByType(req.TypeUrl)
    if err != nil {
        return err
    }

    // 构建响应
    response := &mcp.DiscoveryResponse{
        TypeUrl:       req.TypeUrl,
        VersionInfo:   c.getVersionInfo(req.TypeUrl),
        Resources:     resources,
        Nonce:         generateNonce(),
        CacheTime:     nil,
    }

    // 发送响应
    err = c.sendResponse(response)
    if err != nil {
        return err
    }

    return nil
}

# 定义命名空间资源隔离
apiVersion: v1
kind: Namespace
metadata:
  name: istio-mcp
  labels:
    istio-control-plane: enabled
    mcp-enabled: "true"
---
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: istio-mcp-role
  namespace: istio-mcp
rules:
- apiGroups: [""]
  resources: ["pods", "services"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["networking.istio.io"]
  resources: ["gateways", "virtualservices"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: istio-mcp-role-binding
  namespace: istio-mcp
subjects:
- kind: ServiceAccount
  name: istio-mcp-sa
  namespace: istio-mcp
roleRef:
  kind: Role
  name: istio-mcp-role
  apiGroup: rbac.authorization.k8s.io

// 缓存结构体
type Cache struct {
    sync.RWMutex
    cacheMap    map[string]cachedResource
    ttl         time.Duration
    cleaner     *cleaner
}

type cachedResource struct {
    resource    Resource
    timestamp   time.Time
    version     string
}

// 初始化缓存
func NewCache(ttl time.Duration) *Cache {
    c := &Cache{
        cacheMap: make(map[string]cachedResource),
        ttl:      ttl,
    }

    c.cleaner = newCleaner(c, ttl)
    c.cleaner.start()

    return c
}

// 获取缓存资源
func (c *Cache) Get(key string) (Resource, bool) {
    c.RLock()
    defer c.RUnlock()

    res, exists := c.cacheMap[key]
    if !exists {
        return nil, false
    }

    // 检查缓存是否过期
    if time.Since(res.timestamp) > c.ttl {
        delete(c.cacheMap, key)
        return nil, false
    }

    return res.resource, true
}

// 设置缓存资源
func (c *Cache) Set(key string, resource Resource, version string) {
    c.Lock()
    defer c.Unlock()

    c.cacheMap[key] = cachedResource{
        resource:   resource,
        timestamp:  time.Now(),
        version:    version,
    }
}

# 编译代码
go build -o istio-mcp ./cmd/istio-mcp

# 构建镜像
docker build -t istio-mcp:1.0 .

# 推送镜像
docker push istio-mcp:1.0

# 修改 istiod 部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: istiod
  namespace: istio-system
spec:
  replicas: 1
  selector:
    matchLabels:
      app: istiod
  template:
    metadata:
      labels:
        app: istiod
        version: mcp-1.0
    spec:
      serviceAccountName: istiod-service-account
      containers:
      - name: discovery
        image: istio-mcp:1.0
        args:
        - discovery
        - --meshConfig
        - /etc/istio/config/mesh
        ports:
        - containerPort: 15010
        - containerPort: 15012
        - containerPort: 8060
        - containerPort: 443
        - containerPort: 15014
        volumeMounts:
        - name: istio-config
          mountPath: /etc/istio/config
        - name: istio-certs
          mountPath: /etc/certs
          readOnly: true

# 安装改造后的 Istio 控制平面
istioctl install --set profile=mcp --set values.global.mcpEnabled=true -y

# 验证组件状态
kubectl get pods -n istio-system -l app=istiod

# 验证 MCP 协议通信
kubectl exec -it <data-plane-pod> -n <namespace> -- istioctl proxy-config mcp