admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-04 11:18:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add log of insert to record time spent in mq (#13527)

Browse Source 
Signed-off-by: dragondriver <jiquan.long@zilliz.com>
This commit is contained in:
Jiquan Long 2021-12-16 22:27:42 +08:00 committed by GitHub
parent 256644002f
commit 00c9797d91
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 8 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
internal/proxy/task.go
View File

@ -985,6 +985,10 @@ func (it *insertTask) _assignSegmentID(stream msgstream.MsgStream, pack *msgstre
func (it *insertTask) Execute(ctx context.Context) error {
sp, ctx := trace.StartSpanFromContextWithOperationName(it.ctx, "Proxy-Insert-Execute")
defer sp.Finish()
tr := timerecord.NewTimeRecorder(fmt.Sprintf("proxy execute insert %d", it.ID()))
defer tr.Elapse("done")
collectionName := it.BaseInsertTask.CollectionName
collID, err := globalMetaCache.GetCollectionID(ctx, collectionName)
if err != nil {
@ -1004,6 +1008,7 @@ func (it *insertTask) Execute(ctx context.Context) error {
}
}
it.PartitionID = partitionID
tr.Record("get collection id & partition id from cache")
var tsMsg msgstream.TsMsg = &it.BaseInsertTask
it.BaseMsg.Ctx = ctx
@ -1030,6 +1035,7 @@ func (it *insertTask) Execute(ctx context.Context) error {
return err
}
}
tr.Record("get used message stream")
// Assign SegmentID
var pack *msgstream.MsgPack
@ -1037,6 +1043,7 @@ func (it *insertTask) Execute(ctx context.Context) error {
if err != nil {
return err
}
tr.Record("assign segment id")
err = stream.Produce(pack)
if err != nil {
@ -1044,6 +1051,7 @@ func (it *insertTask) Execute(ctx context.Context) error {
it.result.Status.Reason = err.Error()
return err
}
tr.Record("send insert request to message stream")
return nil
}