milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refine log of Flush (#13165) 

Signed-off-by: dragondriver <jiquan.long@zilliz.com>
pull/13190/head
dragondriver 2021-12-10 17:01:42 +08:00 committed by GitHub
parent f107f9d03e
commit 89d712ebc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 44 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
internal/proxy/impl.go
View File

@ -2214,6 +2214,11 @@ func (node *Proxy) Flush(ctx context.Context, request *milvuspb.FlushRequest) (*
resp.Status.Reason = "proxy is not healthy"
return resp, nil
}
sp, ctx := trace.StartSpanFromContextWithOperationName(ctx, "Proxy-Flush")
defer sp.Finish()
traceID, _, _ := trace.InfoFromSpan(sp)
ft := &flushTask{
ctx: ctx,
Condition: NewTaskCondition(ctx),
@ -2221,39 +2226,65 @@ func (node *Proxy) Flush(ctx context.Context, request *milvuspb.FlushRequest) (*
dataCoord: node.dataCoord,
}
log.Debug("Flush enqueue",
method := "Flush"
log.Debug(
rpcReceived(method),
zap.String("traceID", traceID),
zap.String("role", Params.RoleName),
zap.String("db", request.DbName),
zap.Any("collections", request.CollectionNames))
err := node.sched.ddQueue.Enqueue(ft)
if err != nil {
if err := node.sched.ddQueue.Enqueue(ft); err != nil {
log.Warn(
rpcFailedToEnqueue(method),
zap.Error(err),
zap.String("traceID", traceID),
zap.String("role", Params.RoleName),
zap.String("db", request.DbName),
zap.Any("collections", request.CollectionNames))
resp.Status.Reason = err.Error()
return resp, nil
}
log.Debug("Flush",
log.Debug(
rpcEnqueued(method),
zap.String("traceID", traceID),
zap.String("role", Params.RoleName),
zap.Int64("msgID", request.Base.MsgID),
zap.Uint64("timestamp", request.Base.Timestamp),
zap.Int64("MsgID", ft.ID()),
zap.Uint64("BeginTs", ft.BeginTs()),
zap.Uint64("EndTs", ft.EndTs()),
zap.String("db", request.DbName),
zap.Any("collections", request.CollectionNames))
defer func() {
log.Debug("Flush Done",
if err := ft.WaitToFinish(); err != nil {
log.Warn(
rpcFailedToWaitToFinish(method),
zap.Error(err),
zap.String("traceID", traceID),
zap.String("role", Params.RoleName),
zap.Int64("msgID", request.Base.MsgID),
zap.Uint64("timestamp", request.Base.Timestamp),
zap.Int64("MsgID", ft.ID()),
zap.Uint64("BeginTs", ft.BeginTs()),
zap.Uint64("EndTs", ft.EndTs()),
zap.String("db", request.DbName),
zap.Any("collections", request.CollectionNames))
}()
err = ft.WaitToFinish()
if err != nil {
resp.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
resp.Status.Reason = err.Error()
return resp, nil
}
log.Debug(
rpcDone(method),
zap.String("traceID", traceID),
zap.String("role", Params.RoleName),
zap.Int64("MsgID", ft.ID()),
zap.Uint64("BeginTs", ft.BeginTs()),
zap.Uint64("EndTs", ft.EndTs()),
zap.String("db", request.DbName),
zap.Any("collections", request.CollectionNames))
return ft.result, nil
}