CreateDatabase API执行流程源码解析

整体架构:

CreateDatabase 的数据流向:

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

from pymilvus import (
    connections,
    db,
)

_HOST = '192.168.230.71'
_PORT = '19530'

def connect_to_milvus(db_name="default"):
    print("connect to milvus")
    connections.connect(host=_HOST,
                        port=_PORT,
                        db_name=db_name,
                        )


if __name__ == '__main__':
    connect_to_milvus(db_name="default")

    if "db1" not in db.list_database():
        print("create database: db1")
        db.create_database(db_name="db1")

    print("list databases:")
    print(db.list_database())

客户端SDK向proxy发送一个CreateDatabase API请求，创建一个名为db1的数据库。

2.客户端接受API请求，将request封装为createDatabaseTask，并压入ddQueue队列。

代码路径:internal\proxy\impl.go

func (node *Proxy) CreateDatabase(ctx context.Context, request *milvuspb.CreateDatabaseRequest) (*commonpb.Status, error) {
   
   
    ......
    // request封装为task
    cct := &createDatabaseTask{
   
   
        ctx:                   ctx,
        Condition:             NewTaskCondition(ctx),
        CreateDatabaseRequest: request,
        rootCoord:             node.rootCoord,
        replicateMsgStream:    node.replicateMsgStream,
    }
    ......
    // 将task压入ddQueue队列
    if err := node.sched.ddQueue.Enqueue(cct); err != nil {
   
   

    }
    ......
    // 等待cct执行完
    if err := cct.WaitToFinish(); err != nil {
   
   

    }
}

3.执行createDatabaseTask的3个方法PreExecute、Execute、PostExecute。

PreExecute()一般为参数校验等工作。

Execute()一般为真正执行逻辑。

代码路径:internal\proxy\task_database.go

func (cdt *createDatabaseTask) Execute(ctx context.Context) error {
   
   
    var err error
    cdt.result, err = cdt.rootCoord.CreateDatabase(ctx, cdt.CreateDatabaseRequest)
    if cdt.result != nil && cdt.result.ErrorCode == commonpb.ErrorCode_Success {
   
   
        SendReplicateMessagePack(ctx, cdt.replicateMsgStream, cdt.CreateDatabaseRequest)
    }
    return err
}

从代码可以看出调用了rootCoord的CreateDatabase接口。

4.进入rootCoord的CreateDatabase接口。

代码路径:internal\rootcoord\root_coord.go

继续将请求封装为rootcoord里的createDatabaseTask

func (c *Core) CreateDatabase(ctx context.Context, in *milvuspb.CreateDatabaseRequest) (*commonpb.Status, error) {
   
   
    ......
    // 封装为createDatabaseTask
    t := &createDatabaseTask{
   
   
        baseTask: newBaseTask(ctx, c),
        Req:      in,
    }
    // 加入调度
    if err := c.scheduler.AddTask(t); err != nil {
   
   
        ......
    }
    // 等待task完成
    if err := t.WaitToFinish(); err != nil {
   
   
        ......
    }

    ......
}

5.执行createDatabaseTask的Prepare、Execute、NotifyDone方法。

Execute()为核心方法。

代码路径:internal\rootcoord\create_db_task.go

func (t *createDatabaseTask) Execute(ctx context.Context) error {
   
   
    db := model.NewDatabase(t.dbID, t.Req.GetDbName(), etcdpb.DatabaseState_DatabaseCreated)
    return t.core.meta.CreateDatabase(ctx, db, t.GetTs())
}

在这里重点研究t.core.meta.CreateDatabase()这个方法做了什么事情。

调用栈如下:

t.core.meta.CreateDatabase()(internal\rootcoord\create_db_task.go)
  |--mt.createDatabasePrivate()(internal\rootcoord\meta_table.go)
    |--mt.catalog.CreateDatabase()(同上)
      |--kc.Snapshot.Save()(internal\metastore\kv\rootcoord\kv_catalog.go)
        |--ss.MetaKv.MultiSave()(internal\metastore\kv\rootcoord\suffix_snapshot.go)
          |--kv.executeTxn()(internal\kv\etcd\etcd_kv.go)
            |--txn.Then(ops...).Commit()(使用clientv3操作etcd)

在etcd会产生2个key。

==by-dev/meta/root-coord/database/db-info/445629732221365773==

==by-dev/meta/snapshots/root-coord/database/db-info/445629732221365773_ts445630347829575684==

value的值的结构为etcdpb.DatabaseInfo，然后进行protobuf序列化后存入etcd。

因此etcd存储的是二进制数据。

&pb.DatabaseInfo{
   
   
    TenantId:    db.TenantID,
    Id:          db.ID,
    Name:        db.Name,
    State:       db.State,
    CreatedTime: db.CreatedTime,
}

func (kc *Catalog) CreateDatabase(ctx context.Context, db *model.Database, ts typeutil.Timestamp) error {
   
   
    // key的规则
    key := BuildDatabaseKey(db.ID)
    // value的值
    dbInfo := model.MarshalDatabaseModel(db)
    // 序列化
    v, err := proto.Marshal(dbInfo)
    if err != nil {
   
   
        return err
    }
    return kc.Snapshot.Save(key, string(v), ts)
}

跟踪BuildDatabaseKey()函数，即可以得到key的规则。

这里整理如下:

key规则:

• 前缀/root-coord/database/db-info/{dbID}

• 前缀/snapshots/root-coord/database/db-info/{dbID}_ts{时间戳}

默认数据库名为default，dbID为1。

ID和时间戳都是一串数字，由physicalTime+logicalTime组成。

总结:

• CreateDatabase由proxy传递给协调器rootCoord操作etcd。
• CreateDatabase最终会在etcd上写入2个key。