CreateCollectiondataSyncService执行流程源码解析

milvus版本:v2.3.2

CreateCollection这个API流程较长，也是milvus的核心API之一，涉及的内容比较复杂。这里介绍dataSyncService相关的流程。

这边文章基于【CreateCollection流程_addCollectionMetaStep_milvus源码解析】这篇文章。

整体架构:

CreateCollection 的数据流向:

1.客户端sdk发出CreateCollection API请求。

客户端SDK向proxy发送一个CreateCollection API请求，创建一个名为hello_milvus的collection。

dataCoord会向etcd写入2种类型的key：

• channelwatch/{nodeID}/{chName}
• datacoord-meta/channel-removal/{channelName}

2.dataNode启动时会watch etcd的key（channelwatch/）

这样当dataCoord写入此类型的key，就会触发dataNode相应的动作。

下面进行源码分析:

1.从dataNode启动开始

堆栈:

Run()
  |--d.svr.Run()(cmd\components\data_node.go)
   |--s.start()(internal\distributed\datanode\service.go)
     |--s.datanode.Start()(同上)
       |--Start()（grpc调用internal\datanode\data_node.go）
         |--go node.StartWatchChannels(node.ctx)(同上)

go node.StartWatchChannels()：开启一个goroutine进行watch。

2.进入node.StartWatchChannels()

代码路径:internal\datanode\event_manager.go

// StartWatchChannels start loop to watch channel allocation status via kv(etcd for now)
func (node *DataNode) StartWatchChannels(ctx context.Context) {
   
   
    defer node.stopWaiter.Done()
    defer logutil.LogPanic()
    // 构建key的规则:channelwatch/{nodeID}
    watchPrefix := path.Join(Params.CommonCfg.DataCoordWatchSubPath.GetValue(), fmt.Sprintf("%d", node.GetSession().ServerID))
    log.Info("Start watch channel", zap.String("prefix", watchPrefix))
    // 在etcd上watch key,例如:channelwatch/5
    // 创建collection会产生key:channelwatch/{nodeID}/{chName}
    evtChan := node.watchKv.WatchWithPrefix(watchPrefix)
    // after watch, first check all exists nodes first
    err := node.checkWatchedList()
    if err != nil {
   
   
        log.Warn("StartWatchChannels failed", zap.Error(err))
        return
    }
    // 处理watch事件
    for {
   
   
        select {
   
   
        case <-ctx.Done():
            log.Info("watch etcd loop quit")
            return
        case event, ok := <-evtChan:
            if !ok {
   
   
                ......
            }

            if err := event.Err(); err != nil {
   
   
                ......
            }
            // 处理watch的event
            for _, evt := range event.Events {
   
   
                // We need to stay in order until events enqueued
                node.handleChannelEvt(evt)
            }
        }
    }
}

node.handleChannelEvt(evt)用来处理具体的event，主要是PUT和DELETE事件。

创建collection会向etcd写入kv，属于PUT事件。

删除collection会删除etcd的kv，属于DELETE事件。

3.进入node.handleChannelEvt()

代码路径:internal\datanode\data_node.go

// handleChannelEvt handles event from kv watch event
func (node *DataNode) handleChannelEvt(evt *clientv3.Event) {
   
   
    var e *event
    // 根据type填充event
    switch evt.Type {
   
   
    case clientv3.EventTypePut: // datacoord shall put channels needs to be watched here
        e = &event{
   
   
            eventType: putEventType,
            version:   evt.Kv.Version,
        }

    case clientv3.EventTypeDelete:
        e = &event{
   
   
            eventType: deleteEventType,
            version:   evt.Kv.Version,
        }
    }
    node.handleWatchInfo(e, string(evt.Kv.Key), evt.Kv.Value)
}

evt.Kv.Key的值:

by-dev/meta/channelwatch/6/by-dev-rootcoord-dml_0_445698323354747022v0

4.进入node.handleWatchInfo()

代码路径:internal\datanode\event_manager.go

func (node *DataNode) handleWatchInfo(e *event, key string, data []byte) {
   
   
    switch e.eventType {
   
   
    case putEventType:
        // 反序列化得到ChannelWatchInfo
        watchInfo, err := parsePutEventData(data)
        if err != nil {
   
   
            log.Warn("fail to handle watchInfo", zap.Int("event type", e.eventType), zap.String("key", key), zap.Error(err))
            return
        }

        if isEndWatchState(watchInfo.State) {
   
   
            log.Info("DataNode received a PUT event with an end State", zap.String("state", watchInfo.State.String()))
            return
        }

        if watchInfo.Progress != 0 {
   
   
            log.Info("DataNode received a PUT event with tickler update progress", zap.String("channel", watchInfo.Vchan.ChannelName), zap.Int64("version", e.version))
            return
        }

        e.info = watchInfo
        // 填充虚拟channel名称
        e.vChanName = watchInfo.GetVchan().GetChannelName()
        log.Info("DataNode is handling watchInfo PUT event", zap.String("key", key), zap.Any("watch state", watchInfo.GetState().String()))
    case deleteEventType:
        e.vChanName = parseDeleteEventKey(key)
        log.Info("DataNode is handling watchInfo DELETE event", zap.String("key", key))
    }

    actualManager, loaded := node.eventManagerMap.GetOrInsert(e.vChanName, newChannelEventManager(
        node.handlePutEvent, node.handleDeleteEvent, retryWatchInterval,
    ))
    // loaded=false
    if !loaded {
   
   
        actualManager.Run()
    }

    actualManager.handleEvent(*e)

    // Whenever a delete event comes, this eventManager will be removed from map
    if e.eventType == deleteEventType {
   
   
        if m, loaded := node.eventManagerMap.GetAndRemove(e.vChanName); loaded {
   
   
            m.Close()
        }
    }
}

变量e的值:

newChannelEventManager()返回一个channelEventManager结构体:

func newChannelEventManager(handlePut func(*datapb.ChannelWatchInfo, int64) error,
    handleDel func(string), retryInterval time.Duration,
) *channelEventManager {
   
   
    return &channelEventManager{
   
   
        eventChan:         make(chan event, 10),
        closeChan:         make(chan struct{
   
   }),
        handlePutEvent:    handlePut,// 设置PUT处理函数
        handleDeleteEvent: handleDel,// 设置DELETE处理函数
        retryInterval:     retryInterval,
    }
}

handlePutEvent设置为node.handlePutEvent()。

handleDeleteEvent设置为node.handleDeleteEvent()。

4.进入actualManager.Run()

代码路径:internal\datanode\event_manager.go

func (e *channelEventManager) Run() {
   
   
    e.wg.Add(1)
    go func() {
   
   
        defer e.wg.Done()
        for {
   
   
            select {
   
   
            case event := <-e.eventChan:
                switch event.eventType {
   
   
                case putEventType:
                    // 处理PUT事件:node.handlePutEvent()
                    err := e.handlePutEvent(event.info, event.version)
                    if err != nil {
   
   
                        // logging the error is convenient for follow-up investigation of problems
                        log.Warn("handle put event failed", zap.String("vChanName", event.vChanName), zap.Error(err))
                    }
                case deleteEventType:
                    // 处理DELETE事件:node.handleDeleteEvent()
                    e.handleDeleteEvent(event.vChanName)
                }
            case <-e.closeChan:
                return
            }
        }
    }()
}

event.version的值为1

event.info的值如下:

5.进入e.handlePutEvent()，其实就是进入node.handlePutEvent()

代码路径:internal\datanode\event_manager.go

func (node *DataNode) handlePutEvent(watchInfo *datapb.ChannelWatchInfo, version int64) (err error) {
   
   
    // 获取虚拟channel名称,例如:by-dev-rootcoord-dml_2_445698762473996462v0
    vChanName := watchInfo.GetVchan().GetChannelName()
    // 获取key,例如:channelwatch/7/by-dev-rootcoord-dml_2_445698762473996462v0
    key := path.Join(Params.CommonCfg.DataCoordWatchSubPath.GetValue(), fmt.Sprintf("%d", node.GetSession().ServerID), vChanName)
    tickler := newEtcdTickler(version, key, watchInfo, node.watchKv, Params.DataNodeCfg.WatchEventTicklerInterval.GetAsDuration(time.Second))

    switch watchInfo.State {
   
   
    case datapb.ChannelWatchState_Uncomplete, datapb.ChannelWatchState_ToWatch:
        // 走这条路径
        if err := node.flowgraphManager.addAndStartWithEtcdTickler(node, watchInfo.GetVchan(), watchInfo.GetSchema(), tickler); err != nil {
   
   
            ......
        } else {
   
   
            log.Info("handle put event: new data sync service success", zap.String("vChanName", vChanName))
            watchInfo.State = datapb.ChannelWatchState_WatchSuccess
        }
    case datapb.ChannelWatchState_ToRelease:
        ......
    }

    v, err := proto.Marshal(watchInfo)
    if err != nil {
   
   
        ......
    }

    success, err := node.watchKv.CompareVersionAndSwap(key, tickler.version, string(v))

    if err != nil {
   
   
        ......
    }
    log.Info("handle put event success", zap.String("key", key),
        zap.String("state", watchInfo.State.String()), zap.String("vChanName", vChanName))
    return nil
}

6.进入node.flowgraphManager.addAndStartWithEtcdTickler()

代码路径:internal\datanode\flow_graph_manager.go

func (fm *flowgraphManager) addAndStartWithEtcdTickler(dn *DataNode, vchan *datapb.VchannelInfo, schema *schemapb.CollectionSchema, tickler *etcdTickler) error {
   
   
    log := log.With(zap.String("channel", vchan.GetChannelName()))
    if fm.flowgraphs.Contain(vchan.GetChannelName()) {
   
   
        log.Warn("try to add an existed DataSyncService")
        return nil
    }
    // 构建dataSyncService结构体
    dataSyncService, err := newServiceWithEtcdTickler(context.TODO(), dn, &datapb.ChannelWatchInfo{
   
   
        Schema: schema,
        Vchan:  vchan,
    }, tickler)
    if err != nil {
   
   
        log.Warn("fail to create new DataSyncService", zap.Error(err))
        return err
    }
    // 启动dataSyncService
    dataSyncService.start()
    fm.flowgraphs.Insert(vchan.GetChannelName(), dataSyncService)

    metrics.DataNodeNumFlowGraphs.WithLabelValues(fmt.Sprint(paramtable.GetNodeID())).Inc()
    return nil
}

总结:

1.datanode对channelwatch/{nodeID}进行watch

2.当创建或者删除collection,对etcd进行PUT或者DELETE，触发datanode相应事件。

当PUT事件发生，触发dataSyncService的启动。

一个shard对应一个vchannel，同时启动一个dataSyncService。