CreateIndex API执行流程源码解析2

milvus版本:v2.3.2

上一篇介绍了CreateIndex对etcd元数据的操作，这里介绍另一个操作。

整体架构:

CreateIndex 的数据流向:

1.dataCoord执行CreateIndex。

func (s *Server) CreateIndex(ctx context.Context, req *indexpb.CreateIndexRequest) (*commonpb.Status, error) {
	......
	// 分配indexID,indexID=0
	indexID, err := s.meta.CanCreateIndex(req)
	......

	err = s.meta.CreateIndex(index)
	......

	// 将collectionID发送到channel,其它的goroutine进行消费。
	select {
	case s.notifyIndexChan <- req.GetCollectionID():
	default:
	}

	......
}

上一篇已经分析过s.meta.CreateIndex()，这里重点分析s.notifyIndexChan。将collectionID发送到channel,其它的goroutine进行消费。

2.消费notifyIndexChan也在dataCoord。

代码路径:internal\datacoord\index_service.go

启动dataCoord服务的时候会调用此方法。

func (s *Server) createIndexForSegmentLoop(ctx context.Context) {
	log.Info("start create index for segment loop...")
	defer s.serverLoopWg.Done()

	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			log.Warn("DataCoord context done, exit...")
			return
		case <-ticker.C:
			segments := s.meta.GetHasUnindexTaskSegments()
			for _, segment := range segments {
				if err := s.createIndexesForSegment(segment); err != nil {
					log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
					continue
				}
			}
		case collectionID := <-s.notifyIndexChan:
			log.Info("receive create index notify", zap.Int64("collectionID", collectionID))
			// 获取collection的segment信息
			segments := s.meta.SelectSegments(func(info *SegmentInfo) bool {
				return isFlush(info) && collectionID == info.CollectionID
			})
			for _, segment := range segments {
				if err := s.createIndexesForSegment(segment); err != nil {
					log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
					continue
				}
			}
		case segID := <-s.buildIndexCh:
			log.Info("receive new flushed segment", zap.Int64("segmentID", segID))
			segment := s.meta.GetSegment(segID)
			if segment == nil {
				log.Warn("segment is not exist, no need to build index", zap.Int64("segmentID", segID))
				continue
			}
			if err := s.createIndexesForSegment(segment); err != nil {
				log.Warn("create index for segment fail, wait for retry", zap.Int64("segmentID", segment.ID))
				continue
			}
		}
	}
}

走case collectionID := <-s.notifyIndexChan这个分支。

3.进入s.createIndexesForSegment()

func (s *Server) createIndexesForSegment(segment *SegmentInfo) error {
	// 要创建的索引的信息：索引名称，维度等信息
	indexes := s.meta.GetIndexesForCollection(segment.CollectionID, "")
	for _, index := range indexes {
		if _, ok := segment.segmentIndexes[index.IndexID]; !ok {
			if err := s.createIndexForSegment(segment, index.IndexID); err != nil {
				log.Warn("create index for segment fail", zap.Int64("segmentID", segment.ID),
					zap.Int64("indexID", index.IndexID))
				return err
			}
		}
	}
	return nil
}

4.进入s.createIndexForSegment()

在segment上建立索引。

func (s *Server) createIndexForSegment(segment *SegmentInfo, indexID UniqueID) error {
	log.Info("create index for segment", zap.Int64("segmentID", segment.ID), zap.Int64("indexID", indexID))
	buildID, err := s.allocator.allocID(context.Background())
	if err != nil {
		return err
	}
	segIndex := &model.SegmentIndex{
		SegmentID:    segment.ID,
		CollectionID: segment.CollectionID,
		PartitionID:  segment.PartitionID,
		NumRows:      segment.NumOfRows,
		IndexID:      indexID,
		BuildID:      buildID,
		CreateTime:   uint64(segment.ID),
		WriteHandoff: false,
	}
	// 添加segment-index
	// 前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}
	if err = s.meta.AddSegmentIndex(segIndex); err != nil {
		return err
	}
	// 入队，通知IndexNode执行索引创建任务
	s.indexBuilder.enqueue(buildID)
	return nil
}

segIndex变量:

s.meta.AddSegmentIndex(segIndex)在etcd添加segment-index。

==前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}==

func (kc *Catalog) CreateSegmentIndex(ctx context.Context, segIdx *model.SegmentIndex) error {
    // key规则
	key := BuildSegmentIndexKey(segIdx.CollectionID, segIdx.PartitionID, segIdx.SegmentID, segIdx.BuildID)

	value, err := proto.Marshal(model.MarshalSegmentIndexModel(segIdx))
	if err != nil {
		return err
	}
    // 写入etcd
	err = kc.MetaKv.Save(key, string(value))
	if err != nil {
		log.Error("failed to save segment index meta in etcd", zap.Int64("buildID", segIdx.BuildID),
			zap.Int64("segmentID", segIdx.SegmentID), zap.Error(err))
		return err
	}
	return nil
}

5.进入s.indexBuilder.enqueue(buildID)

此函数作用:入队，通知IndexNode执行索引创建任务。

6.进入process()

func (ib *indexBuilder) process(buildID UniqueID) bool {
	......
	// 有一个定时器，默认1秒，配置项indexCoord.scheduler.interval
	switch state {
	case indexTaskInit:
		segment := ib.meta.GetSegment(meta.SegmentID)
		if !isSegmentHealthy(segment) || !ib.meta.IsIndexExist(meta.CollectionID, meta.IndexID) {
			log.Ctx(ib.ctx).Info("task is no need to build index, remove it", zap.Int64("buildID", buildID))
			if err := ib.meta.DeleteTask(buildID); err != nil {
				log.Ctx(ib.ctx).Warn("IndexCoord delete index failed", zap.Int64("buildID", buildID), zap.Error(err))
				return false
			}
			deleteFunc(buildID)
			return true
		}
		indexParams := ib.meta.GetIndexParams(meta.CollectionID, meta.IndexID)
		if isFlatIndex(getIndexType(indexParams)) || meta.NumRows < Params.DataCoordCfg.MinSegmentNumRowsToEnableIndex.GetAsInt64() {
			// 数量小于最低数量无需建立索引,默认为1024
			log.Ctx(ib.ctx).Info("segment does not need index really", zap.Int64("buildID", buildID),
				zap.Int64("segmentID", meta.SegmentID), zap.Int64("num rows", meta.NumRows))
			if err := ib.meta.FinishTask(&indexpb.IndexTaskInfo{
				BuildID:        buildID,
				State:          commonpb.IndexState_Finished,
				IndexFileKeys:  nil,
				SerializedSize: 0,
				FailReason:     "",
			}); err != nil {
				log.Ctx(ib.ctx).Warn("IndexCoord update index state fail", zap.Int64("buildID", buildID), zap.Error(err))
				return false
			}
			updateStateFunc(buildID, indexTaskDone)
			return true
		}
		// peek client
		// if all IndexNodes are executing task, wait for one of them to finish the task.
		nodeID, client := ib.nodeManager.PeekClient(meta)
		if client == nil {
			log.Ctx(ib.ctx).WithRateGroup("dc.indexBuilder", 1, 60).RatedInfo(5, "index builder peek client error, there is no available")
			return false
		}
		// update version and set nodeID
		if err := ib.meta.UpdateVersion(buildID, nodeID); err != nil {
			log.Ctx(ib.ctx).Warn("index builder update index version failed", zap.Int64("build", buildID), zap.Error(err))
			return false
		}
		// binLogs为向量数据文件的位置信息
		// files/insert_log/444517122896489678/444517122896489679/444517122896489694/102/444517122896236031
		// files/insert_log/{collectionID}/{partitionID}/{segmentID}/{fieldID}/444517122896236031
		binLogs := make([]string, 0)
		fieldID := ib.meta.GetFieldIDByIndexID(meta.CollectionID, meta.IndexID)
		for _, fieldBinLog := range segment.GetBinlogs() {
			if fieldBinLog.GetFieldID() == fieldID {
				for _, binLog := range fieldBinLog.GetBinlogs() {
					binLogs = append(binLogs, binLog.LogPath)
				}
				break
			}
		}

		typeParams := ib.meta.GetTypeParams(meta.CollectionID, meta.IndexID)

		var storageConfig *indexpb.StorageConfig
		// 获取元数据存储类型：minio
		if Params.CommonCfg.StorageType.GetValue() == "local" {
			storageConfig = &indexpb.StorageConfig{
				RootPath:    Params.LocalStorageCfg.Path.GetValue(),
				StorageType: Params.CommonCfg.StorageType.GetValue(),
			}
		} else {
			storageConfig = &indexpb.StorageConfig{
				Address:          Params.MinioCfg.Address.GetValue(),
				AccessKeyID:      Params.MinioCfg.AccessKeyID.GetValue(),
				SecretAccessKey:  Params.MinioCfg.SecretAccessKey.GetValue(),
				UseSSL:           Params.MinioCfg.UseSSL.GetAsBool(),
				BucketName:       Params.MinioCfg.BucketName.GetValue(),
				RootPath:         Params.MinioCfg.RootPath.GetValue(),
				UseIAM:           Params.MinioCfg.UseIAM.GetAsBool(),
				IAMEndpoint:      Params.MinioCfg.IAMEndpoint.GetValue(),
				StorageType:      Params.CommonCfg.StorageType.GetValue(),
				Region:           Params.MinioCfg.Region.GetValue(),
				UseVirtualHost:   Params.MinioCfg.UseVirtualHost.GetAsBool(),
				CloudProvider:    Params.MinioCfg.CloudProvider.GetValue(),
				RequestTimeoutMs: Params.MinioCfg.RequestTimeoutMs.GetAsInt64(),
			}
		}
		req := &indexpb.CreateJobRequest{
			ClusterID:           Params.CommonCfg.ClusterPrefix.GetValue(),
			IndexFilePrefix:     path.Join(ib.chunkManager.RootPath(), common.SegmentIndexPath),
			BuildID:             buildID,
			DataPaths:           binLogs,
			IndexVersion:        meta.IndexVersion + 1,
			StorageConfig:       storageConfig,
			IndexParams:         indexParams,
			TypeParams:          typeParams,
			NumRows:             meta.NumRows,
			CurrentIndexVersion: ib.indexEngineVersionManager.GetCurrentIndexEngineVersion(),
		}
		// 通知IndexNode创建索引
		if err := ib.assignTask(client, req); err != nil {
			......
		}
		......
		// update index meta state to InProgress
		// 更新索引状态
		if err := ib.meta.BuildIndex(buildID); err != nil {
			......
		}
		updateStateFunc(buildID, indexTaskInProgress)

	case indexTaskDone:
		......

	default:
		......
	}
	return true
}

ib.assignTask()通知indexNode创建索引。

7.进入ib.assignTask()

// assignTask sends the index task to the IndexNode, it has a timeout interval, if the IndexNode doesn't respond within
// the interval, it is considered that the task sending failed.
func (ib *indexBuilder) assignTask(builderClient types.IndexNodeClient, req *indexpb.CreateJobRequest) error {
	......
    // rpc调用indexNode
	resp, err := builderClient.CreateJob(ctx, req)
	......
}

8.进入builderClient.CreateJob()

代码路径:internal\indexnode\indexnode_service.go

func (i *IndexNode) CreateJob(ctx context.Context, req *indexpb.CreateJobRequest) (*commonpb.Status, error) {
	......
	task := &indexBuildTask{
		ident:          fmt.Sprintf("%s/%d", req.ClusterID, req.BuildID),
		ctx:            taskCtx,
		cancel:         taskCancel,
		BuildID:        req.GetBuildID(),
		ClusterID:      req.GetClusterID(),
		node:           i,
		req:            req,
		cm:             cm,
		nodeID:         i.GetNodeID(),
		tr:             timerecord.NewTimeRecorder(fmt.Sprintf("IndexBuildID: %d, ClusterID: %s", req.BuildID, req.ClusterID)),
		serializedSize: 0,
	}
	ret := merr.Success()
	if err := i.sched.IndexBuildQueue.Enqueue(task); err != nil {
		......
	}
	......
}

task执行逻辑:

pipelines := []func(context.Context) error{t.Prepare, t.BuildIndex, t.SaveIndexFiles}

依次执行task的Prepare()、BuildIndex()、SaveIndexFiles()方法。

9.进入BuildIndex()

func (it *indexBuildTask) BuildIndex(ctx context.Context) error {
	......
	var buildIndexInfo *indexcgowrapper.BuildIndexInfo
	buildIndexInfo, err = indexcgowrapper.NewBuildIndexInfo(it.req.GetStorageConfig())
	......
	err = buildIndexInfo.AppendFieldMetaInfo(it.collectionID, it.partitionID, it.segmentID, it.fieldID, it.fieldType)
	......

	err = buildIndexInfo.AppendIndexMetaInfo(it.req.IndexID, it.req.BuildID, it.req.IndexVersion)
	......

	err = buildIndexInfo.AppendBuildIndexParam(it.newIndexParams)
	......

	jsonIndexParams, err := json.Marshal(it.newIndexParams)
	......

	err = buildIndexInfo.AppendBuildTypeParam(it.newTypeParams)
	......

	for _, path := range it.req.GetDataPaths() {
		err = buildIndexInfo.AppendInsertFile(path)
		if err != nil {
			log.Ctx(ctx).Warn("append insert binlog path failed", zap.Error(err))
			return err
		}
	}

	it.currentIndexVersion = getCurrentIndexVersion(it.req.GetCurrentIndexVersion())
	if err := buildIndexInfo.AppendIndexEngineVersion(it.currentIndexVersion); err != nil {
		log.Ctx(ctx).Warn("append index engine version failed", zap.Error(err))
		return err
	}
	// cgo层调用CreateIndex()
    // it.index有个Upload()方法写入s3(SaveIndexFiles方法里调用it.index.Upload()写入s3)
	it.index, err = indexcgowrapper.CreateIndex(ctx, buildIndexInfo)
	if err != nil {
		......
	}

	......
}

indexcgowrapper.CreateIndex()调用C层创建索引。

10.进入SaveIndexFiles()

func (it *indexBuildTask) SaveIndexFiles(ctx context.Context) error {
	......
	// c++层上传索引文件到s3
	indexFilePath2Size, err := it.index.UpLoad()
	......

}

总结:

• CreateIndex由proxy传递给协调器dataCoord操作etcd。
  • 前缀/field-index/{collectionID}/{IndexID}
  • 前缀/segment-index/{collectionID}/{partitionID}/{segmentID}/{buildID}
• dataCoord通知indexNode创建索引,indexNode通过cgo调用C++层创建索引，并上传至s3。