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。
