diff --git a/internal/proxy/impl.go b/internal/proxy/impl.go index 53836f2d8a..9d2f2e9d26 100644 --- a/internal/proxy/impl.go +++ b/internal/proxy/impl.go @@ -875,6 +875,11 @@ func (node *Proxy) DropPartition(ctx context.Context, request *milvuspb.DropPart if !node.checkHealthy() { return unhealthyStatus(), nil } + + sp, ctx := trace.StartSpanFromContextWithOperationName(ctx, "Proxy-Insert") + defer sp.Finish() + traceID, _, _ := trace.InfoFromSpan(sp) + dpt := &dropPartitionTask{ ctx: ctx, Condition: NewTaskCondition(ctx), @@ -883,44 +888,73 @@ func (node *Proxy) DropPartition(ctx context.Context, request *milvuspb.DropPart result: nil, } - log.Debug("DropPartition 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(dpt) - if err != nil { - return &commonpb.Status{ - ErrorCode: commonpb.ErrorCode_UnexpectedError, - Reason: err.Error(), - }, nil - } + method := "DropPartition" - log.Debug("DropPartition", + 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("DropPartition Done", + + if err := node.sched.ddQueue.Enqueue(dpt); 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 = dpt.WaitToFinish() - if err != nil { return &commonpb.Status{ ErrorCode: commonpb.ErrorCode_UnexpectedError, Reason: err.Error(), }, nil } + + log.Debug( + rpcEnqueued(method), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", dpt.ID()), + zap.Uint64("BeginTS", dpt.BeginTs()), + zap.Uint64("EndTS", dpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + + if err := dpt.WaitToFinish(); err != nil { + log.Warn( + rpcFailedToWaitToFinish(method), + zap.Error(err), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", dpt.ID()), + zap.Uint64("BeginTS", dpt.BeginTs()), + zap.Uint64("EndTS", dpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + + return &commonpb.Status{ + ErrorCode: commonpb.ErrorCode_UnexpectedError, + Reason: err.Error(), + }, nil + } + + log.Debug( + rpcDone(method), + zap.String("traceID", traceID), + zap.String("role", Params.RoleName), + zap.Int64("MsgID", dpt.ID()), + zap.Uint64("BeginTS", dpt.BeginTs()), + zap.Uint64("EndTS", dpt.EndTs()), + zap.String("db", request.DbName), + zap.String("collection", request.CollectionName), + zap.String("partition", request.PartitionName)) + return dpt.result, nil }