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

Add test case for delete task (#7939) 

Signed-off-by: dragondriver <jiquan.long@zilliz.com>
pull/7953/head
dragondriver 2021-09-15 14:04:54 +08:00 committed by GitHub
parent a0fd2707cb
commit 5f3b961502
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 122 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
internal/proxy/impl.go
View File

@ -1355,7 +1355,7 @@ func (node *Proxy) Delete(ctx context.Context, request *milvuspb.DeleteRequest)
}, nil
}
dt := &DeleteTask{
dt := &deleteTask{
ctx: ctx,
Condition: NewTaskCondition(ctx),
DeleteRequest: request,
@ -2210,7 +2210,7 @@ func (node *Proxy) GetMetrics(ctx context.Context, req *milvuspb.GetMetricsReque
log.Debug("failed to get system info metrics from cache, recompute instead",
zap.Error(err))
metrics, err := getSystemInfoMetrics(ctx, req, node, node.session.Address)
metrics, err := getSystemInfoMetrics(ctx, req, node)
log.Debug("Proxy.GetMetrics",
zap.Int64("node_id", Params.ProxyID),

3
internal/proxy/metrics_info.go
View File

@ -32,7 +32,6 @@ func getSystemInfoMetrics(
ctx context.Context,
request *milvuspb.GetMetricsRequest,
node *Proxy,
ip string,
) (*milvuspb.GetMetricsResponse, error) {
var err error
@ -54,7 +53,7 @@ func getSystemInfoMetrics(
ErrorReason: "",
Name: proxyRoleName,
HardwareInfos: metricsinfo.HardwareMetrics{
IP: ip,
IP: node.session.Address,
CPUCoreCount: metricsinfo.GetCPUCoreCount(false),
CPUCoreUsage: metricsinfo.GetCPUUsage(),
Memory: metricsinfo.GetMemoryCount(),

9
internal/proxy/metrics_info_test.go
View File

@ -15,6 +15,9 @@ import (
"context"
"testing"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/internal/util/uniquegenerator"
"github.com/milvus-io/milvus/internal/proto/commonpb"
@ -24,8 +27,6 @@ import (
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/funcutil"
)
func TestProxy_metrics(t *testing.T) {
@ -54,6 +55,7 @@ func TestProxy_metrics(t *testing.T) {
queryCoord: qc,
dataCoord: dc,
indexCoord: ic,
session: &sessionutil.Session{Address: funcutil.GenRandomStr()},
}
rc.getMetricsFunc = func(ctx context.Context, request *milvuspb.GetMetricsRequest) (*milvuspb.GetMetricsResponse, error) {
@ -267,9 +269,8 @@ func TestProxy_metrics(t *testing.T) {
}
ip := funcutil.GenRandomStr()
req, _ := metricsinfo.ConstructRequestByMetricType(metricsinfo.SystemInfoMetrics)
resp, err := getSystemInfoMetrics(ctx, req, proxy, ip)
resp, err := getSystemInfoMetrics(ctx, req, proxy)
assert.NoError(t, err)
assert.NotNil(t, resp)

31
internal/proxy/task.go
View File

@ -79,6 +79,7 @@ const (
ReleaseCollectionTaskName = "ReleaseCollectionTask"
LoadPartitionTaskName = "LoadPartitionsTask"
ReleasePartitionTaskName = "ReleasePartitionsTask"
deleteTaskName = "DeleteTask"
)
type task interface {
@ -4620,52 +4621,52 @@ func (rpt *releasePartitionsTask) PostExecute(ctx context.Context) error {
return nil
}
type DeleteTask struct {
type deleteTask struct {
Condition
*milvuspb.DeleteRequest
ctx context.Context
result *milvuspb.MutationResult
}
func (dt *DeleteTask) TraceCtx() context.Context {
func (dt *deleteTask) TraceCtx() context.Context {
return dt.ctx
}
func (dt *DeleteTask) ID() UniqueID {
func (dt *deleteTask) ID() UniqueID {
return dt.Base.MsgID
}
func (dt *DeleteTask) SetID(uid UniqueID) {
func (dt *deleteTask) SetID(uid UniqueID) {
dt.Base.MsgID = uid
}
func (dt *DeleteTask) Type() commonpb.MsgType {
func (dt *deleteTask) Type() commonpb.MsgType {
return dt.Base.MsgType
}
func (dt *DeleteTask) Name() string {
return ReleasePartitionTaskName
func (dt *deleteTask) Name() string {
return deleteTaskName
}
func (dt *DeleteTask) BeginTs() Timestamp {
func (dt *deleteTask) BeginTs() Timestamp {
return dt.Base.Timestamp
}
func (dt *DeleteTask) EndTs() Timestamp {
func (dt *deleteTask) EndTs() Timestamp {
return dt.Base.Timestamp
}
func (dt *DeleteTask) SetTs(ts Timestamp) {
func (dt *deleteTask) SetTs(ts Timestamp) {
dt.Base.Timestamp = ts
}
func (dt *DeleteTask) OnEnqueue() error {
func (dt *deleteTask) OnEnqueue() error {
dt.Base = &commonpb.MsgBase{}
return nil
}
func (dt *DeleteTask) PreExecute(ctx context.Context) error {
dt.Base.MsgType = commonpb.MsgType_ReleasePartitions
func (dt *deleteTask) PreExecute(ctx context.Context) error {
dt.Base.MsgType = commonpb.MsgType_Delete
dt.Base.SourceID = Params.ProxyID
collName := dt.CollectionName
@ -4681,10 +4682,10 @@ func (dt *DeleteTask) PreExecute(ctx context.Context) error {
return nil
}
func (dt *DeleteTask) Execute(ctx context.Context) (err error) {
func (dt *deleteTask) Execute(ctx context.Context) (err error) {
return nil
}
func (dt *DeleteTask) PostExecute(ctx context.Context) error {
func (dt *deleteTask) PostExecute(ctx context.Context) error {
return nil
}

98
internal/proxy/task_test.go
View File

@ -2979,3 +2979,101 @@ func TestInsertTask_all(t *testing.T) {
assert.NoError(t, task.Execute(ctx))
assert.NoError(t, task.PostExecute(ctx))
}
func TestDeleteTask_all(t *testing.T) {
Params.Init()
ctx := context.Background()
prefix := "TestDeleteTask_all"
dbName := ""
collectionName := prefix + funcutil.GenRandomStr()
partitionName := prefix + funcutil.GenRandomStr()
task := &deleteTask{
Condition: NewTaskCondition(ctx),
DeleteRequest: &milvuspb.DeleteRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_Delete,
MsgID: 0,
Timestamp: 0,
SourceID: 0,
},
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
Expr: "",
},
ctx: ctx,
result: &milvuspb.MutationResult{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
},
IDs: nil,
SuccIndex: nil,
ErrIndex: nil,
Acknowledged: false,
InsertCnt: 0,
DeleteCnt: 0,
UpsertCnt: 0,
Timestamp: 0,
},
}
assert.NoError(t, task.OnEnqueue())
assert.NotNil(t, task.TraceCtx())
id := UniqueID(uniquegenerator.GetUniqueIntGeneratorIns().GetInt())
task.SetID(id)
assert.Equal(t, id, task.ID())
task.Base.MsgType = commonpb.MsgType_Delete
assert.Equal(t, commonpb.MsgType_Delete, task.Type())
ts := Timestamp(time.Now().UnixNano())
task.SetTs(ts)
assert.Equal(t, ts, task.BeginTs())
assert.Equal(t, ts, task.EndTs())
assert.NoError(t, task.PreExecute(ctx))
assert.NoError(t, task.Execute(ctx))
assert.NoError(t, task.PostExecute(ctx))
}
func TestDeleteTask_PreExecute(t *testing.T) {
Params.Init()
ctx := context.Background()
prefix := "TestDeleteTask_all"
dbName := ""
collectionName := prefix + funcutil.GenRandomStr()
partitionName := prefix + funcutil.GenRandomStr()
task := &deleteTask{
DeleteRequest: &milvuspb.DeleteRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_Delete,
MsgID: 0,
Timestamp: 0,
SourceID: 0,
},
DbName: dbName,
CollectionName: collectionName,
PartitionName: partitionName,
Expr: "",
},
}
assert.NoError(t, task.PreExecute(ctx))
task.DeleteRequest.CollectionName = "" // empty
assert.Error(t, task.PreExecute(ctx))
task.DeleteRequest.CollectionName = collectionName
task.DeleteRequest.PartitionName = "" // empty
assert.Error(t, task.PreExecute(ctx))
task.DeleteRequest.PartitionName = partitionName
}