diff --git a/internal/proxy/impl.go b/internal/proxy/impl.go index 9d2f2e9d26..c61d6771db 100644 --- a/internal/proxy/impl.go +++ b/internal/proxy/impl.go @@ -965,6 +965,11 @@ func (node *Proxy) HasPartition(ctx context.Context, request *milvuspb.HasPartit Status: unhealthyStatus(), }, nil } + + sp, ctx := trace.StartSpanFromContextWithOperationName(ctx, "Proxy-Insert") + defer sp.Finish() + traceID, _, _ := trace.InfoFromSpan(sp) + hpt := &hasPartitionTask{ ctx: ctx, Condition: NewTaskCondition(ctx), @@ -973,42 +978,26 @@ func (node *Proxy) HasPartition(ctx context.Context, request *milvuspb.HasPartit result: nil, } - log.Debug("HasPartition enqueue", - zap.String("role", Params.RoleName), - zap.String("db", request.DbName), - zap.String("collection", request.CollectionName), - zap.String("partition", request.PartitionName)) - err := node.sched.ddQueue.Enqueue(hpt) - if err != nil { - return &milvuspb.BoolResponse{ - Status: &commonpb.Status{ - ErrorCode: commonpb.ErrorCode_UnexpectedError, - Reason: err.Error(), - }, - Value: false, - }, nil - } + method := "HasPartition" - log.Debug("HasPartition", + log.Debug( + rpcReceived(method), + zap.String("traceID", traceID), zap.String("role", Params.RoleName), - zap.Int64("msgID", request.Base.MsgID), - zap.Uint64("timestamp", request.Base.Timestamp), zap.String("db", request.DbName), zap.String("collection", request.CollectionName), zap.String("partition", request.PartitionName)) - defer func() { - log.Debug("HasPartition Done", + + if err := node.sched.ddQueue.Enqueue(hpt); err != nil { + log.Warn( + rpcFailedToEnqueue(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.String("db", request.DbName), zap.String("collection", request.CollectionName), zap.String("partition", request.PartitionName)) - }() - err = hpt.WaitToFinish() - if err != nil { return &milvuspb.BoolResponse{ Status: &commonpb.Status{ ErrorCode: commonpb.ErrorCode_UnexpectedError, @@ -1017,6 +1006,51 @@ func (node *Proxy) HasPartition(ctx context.Context, request *milvuspb.HasPartit Value: false, }, nil } + + log.Debug( + rpcEnqueued(method), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", hpt.ID()), + zap.Uint64("BeginTS", hpt.BeginTs()), + zap.Uint64("EndTS", hpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + + if err := hpt.WaitToFinish(); err != nil { + log.Warn( + rpcFailedToWaitToFinish(method), + zap.Error(err), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", hpt.ID()), + zap.Uint64("BeginTS", hpt.BeginTs()), + zap.Uint64("EndTS", hpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + + return &milvuspb.BoolResponse{ + Status: &commonpb.Status{ + ErrorCode: commonpb.ErrorCode_UnexpectedError, + Reason: err.Error(), + }, + Value: false, + }, nil + } + + log.Debug( + rpcDone(method), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", hpt.ID()), + zap.Uint64("BeginTS", hpt.BeginTs()), + zap.Uint64("EndTS", hpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + return hpt.result, nil }