mcp

介绍

MCP是基于订阅的配置分发API。配置使用者(即sink)从配置生产者(即source)请求更新资源集合.添加,更新或删除资源时,source会将资源更新推送到sink.sink积极确认资源更新,如果sink接受,则返回ACK,如果被拒绝则返回NACK,例如: 因为资源无效。一旦对先前的更新进行了ACK/NACK,则源可以推送其他更新.该源一次只能运行一个未完成的更新(每个集合).

MCP是一对双向流gRPC API服务(ResourceSource和ResourceSink)。

• 当source是服务器而sink是客户端时,将使用ResourceSource服务.默认情况下,Galley实现ResourceSource服务,并且Pilot/Mixer连接作为客户端。

• 当source是客户端,而接收器是服务器时,将使用ResourceSink服务.可以将Galley配置为可选地dial-out到远程配置sink,例如 Pilot位于另一个集群中,在该集群中,它不能作为客户端启动与Galley的连接.在这种情况下,Pilot将实现ResourceSink服务,而Galley将作为客户端进行连接。

就消息交换而言,ResourceSource和ResourceSink在语义上是等效的.唯一有意义的区别是谁启动连接并打开grpc流。

数据模型

MCP是一种传输机制,可以通过管理器组件配置先导和混合器.MCP定义了每种资源的通用元数据格式,而资源特定的内容则在其他位置定义(例如https://github.com/istio/api/tree/master/networking/v1alpha3).

Collections

相同类型的资源被组织到命名集合中.Istio API集合名称的格式为istio///,其中,和<api由API样式准则定义.例如,VirtualService的collection名称为istio/networking/v1alpha3/virtualservices。

元数据

建立连接

ResourceSource服务-客户端是reource sink.客户端dail服务器并建立新的gRPC流.客户端发送RequestResources并接收Resources消息。

ResourceSink服务-客户端是资源源.客户端拨打服务器并建立新的gRPC流.服务器发送RequestResources并接收Resources消息。

配置升级

以下概述适用于ResourceSink和ResourceSource服务,无论客户端/服务器角色如何。

资源更新协议由增量xDS协议派生。除了资源提示已被删除之外,协议交换几乎是相同的。下面的大多数文本和图表都是从增量xDS文档中复制并进行相应调整的。

在MCP中,资源首先按collection进行组织。在每个collection中,资源可以通过元数据名称唯一地标识。对各个资源进行版本控制,以区分同一命名资源的较新版本。

可以在两种情况下发送RequestResource消息:

• MCP双向更改流中的初始消息

• 作为对先前资源消息的ACK或NACK响应。在这种情况下,response_nonce设置为资源消息中的现时值.ACK/NACK由后续请求中是否存在error_detail决定。

初始的RequestResources消息包括与所订阅的资源集(例如VirtualService)相对应的集合,节点接收器标识符和nonce字段以及initial_resource_version(稍后会详细介绍)。当请求的资源可用时,source将发送资源消息。处理资源消息后,sink在流上发送新的RequestResources消息,指定成功应用的最后一个版本以及源提供的随机数。

随机数字段用于将每个集合的RequestResources和Resources消息配对。源一次只能发送一个未完成的资源消息(每个collection),并等待接收器进行ACK/NACK。接收到更新后,接收器将在解码,验证更新并将更新持久保存到其内部配置存储后,期望相对较快地发送ACK/NACK。

source应忽略具有过期和未知随机数的请求,这些请求与最近发送的Resource消息中的随机数不匹配。

成功示例

以下示例显示接收器接收到已成功ACK的一系列更改。

以下示例显示了与增量更新一起交付的所需资源.此示例假定source支持增量.当source不支持增量更新时,考虑到接收器是否请求增量更新,推送的资源将始终将增量设置为false.在任何时候,源都可以决定推送完整状态更新,而不必考虑接收器的请求.双方必须协商(即同意)在每个请求/响应的基础上使用增量,以增量发送更新。

错误示例

以下示例显示了无法应用更改时发生的情况

接收器仅在特殊情况下应为NACK。例如,如果一组资源无效,格式错误或无法解码。NACK的更新应发出警报,以供人类随后进行调查.源不应该重新发送先前NACK相同的资源集.在将金丝雀推送到更大数量的资源接收器之前,也可以将金丝雀推送到专用接收器,以验证正确性(非NACK)。

MCP中的随机数用于匹配RequestResources和Resource。在重新连接时,接收器可以通过为每个集合指定带有initial_resource_version的已知资源版本来尝试恢复与同一源的会话。

mcp实现探

接下来以官方的测试用例分析官方的mcp实现,代码地址:https://github.com/istio/istio/blob/master/pkg/mcp/testing/server.go 对于mcpsource通过

mcp.RegisterResourceSourceServer(gs, srcServer)

进行注册

Server.EstablishResourceStream

获取客户端信息,并进行鉴权,交给ProcessStream进行处理

func (s *Server) EstablishResourceStream(stream mcp.ResourceSource_EstablishResourceStreamServer) error {
    if s.rateLimiter != nil {
        if err := s.rateLimiter.Wait(stream.Context()); err != nil {
            return err
        }

    }
    var authInfo credentials.AuthInfo
    if peerInfo, ok := peer.FromContext(stream.Context()); ok { //获取客户端信息
        authInfo = peerInfo.AuthInfo
    } else {
        scope.Warnf("No peer info found on the incoming stream.")
    }

    if err := s.authCheck.Check(authInfo); err != nil { // 认证
        return status.Errorf(codes.Unauthenticated, "Authentication failure: %v", err)
    }

    if err := stream.SendHeader(s.metadata); err != nil {
        return err
    }
    err := s.src.ProcessStream(stream) 
    code := status.Code(err)
    if code == codes.OK || code == codes.Canceled || err == io.EOF {
        return nil
    }
    return err
}

Source.ProcessStream

• s.newConnection(stream)

初始化connection,

con := &connection{
    stream:   stream,
    peerAddr: peerAddr,
    requestC: make(chan *mcp.RequestResources),
    watches:  make(map[string]*watch),
    watcher:  s.watcher,
    id:       atomic.AddInt64(&s.nextStreamID, 1),
    reporter: s.reporter,
    limiter:  s.requestLimiter.Create(),
    queue:    internal.NewUniqueScheduledQueue(len(s.collections)),
}

watcher即为Server.src的watcher

func NewServer(srcOptions *Options, serverOptions *ServerOptions) *Server {
    s := &Server{
        src:         New(srcOptions),
        authCheck:   serverOptions.AuthChecker,
        rateLimiter: serverOptions.RateLimiter,
        metadata:    serverOptions.Metadata,
    }
    return s
}

• con.receive()

通过receive不断将数据写入requestC channel

• 从requestC 读取请求

case req, more := <-con.requestC:
    if !more {
        return con.reqError
    }
    if con.limiter != nil {
        if err := con.limiter.Wait(stream.Context()); err != nil {
            return err
        }

    }
    if err := con.processClientRequest(req); err != nil {
        return err
    }

• 响应入队

func (con *connection) queueResponse(resp *WatchResponse) {
    if resp == nil {
        con.queue.Close()
    } else {
        con.queue.Enqueue(resp.Collection, resp) //响应入队
    }
}

• 响应出队

从queue中获取需要返回的response,返回给sink

collection, item, ok := con.queue.Dequeue() // 从queue读取处理后的返回结果
if !ok {
    break
}

resp := item.(*WatchResponse)

w, ok := con.watches[collection]
if !ok {
    scope.Errorf("unknown collection in dequeued watch response: %v", collection)
    break // bug?
}

// the response may have been cleared before we got to it
if resp != nil {
    if err := con.pushServerResponse(w, resp); err != nil { //通过pushServerResponse 将响应发送给sink
        return err
    }
}

connection.processClientRequest

通过调用Watch方法处理请求

sr := &Request{
            SinkNode:    req.SinkNode,
            Collection:  collection,
            VersionInfo: versionInfo,
            incremental: req.Incremental,
        }
w.cancel = con.watcher.Watch(sr, con.queueResponse, con.peerAddr)

由此我们可以看出我们实现一个mcp的核心在于实现一个watcher,传递给mcp source server

type Watcher interface {
    Watch(*Request, PushResponseFunc, string) CancelWatchFunc
}

官方mcp watcher 实现

pkg/mcp/testing/server.go

cache := snapshot.New(groups.DefaultIndexFn)

重点关注cache.Watch

func (c *Cache) Watch(
    request *source.Request,
    pushResponse source.PushResponseFunc,
    peerAddr string) source.CancelWatchFunc {
    group := c.groupIndex(request.Collection, request.SinkNode) // 获取sink要获取的group

    c.mu.Lock()
    defer c.mu.Unlock()

    info := c.fillStatus(group, request, peerAddr) // 初始化对应group peer的同步状态信息

    collection := request.Collection

    // return an immediate response if a snapshot is available and the
    // requested version doesn't match.
    if snapshot, ok := c.snapshots[group]; ok {  // 获取对应组的snapshot

        version := snapshot.Version(request.Collection) // 计算版本信息
        scope.Debugf("Found snapshot for group: %q for %v @ version: %q",
            group, request.Collection, version)

        if version != request.VersionInfo {        // 如果sink的当前版本和source的版本不一致则推送response
            scope.Debugf("Responding to group %q snapshot:\n%v\n", group, snapshot)
            response := &source.WatchResponse{
                Collection: request.Collection,
                Version:    version,
                Resources:  snapshot.Resources(request.Collection),
                Request:    request,
            }
            pushResponse(response)
            return nil
        }
        info.synced[request.Collection][peerAddr] = true
    }

    // 如果版本一致则.返回一个cancel,同时记录对应的watchs,当SetSnapshot时,也就是有更新时会进行调用
    c.watchCount++
    watchID := c.watchCount

    scope.Infof("Watch(): created watch %d for %s from group %q, version %q",
        watchID, collection, group, request.VersionInfo)

    info.mu.Lock()
    info.watches[watchID] = &responseWatch{request: request, pushResponse: pushResponse}
    info.mu.Unlock()

    cancel := func() {
        c.mu.Lock()
        defer c.mu.Unlock()
        if s, ok := c.status[group]; ok {
            s.mu.Lock()
            delete(s.watches, watchID)
            s.mu.Unlock()
        }
    }
    return cancel
}