admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-08 01:58:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

enhance: Rename Compaction to CompactionV2 (#33858)

Browse Source 
Due to the removal of injection and syncSegments from the compaction, we
need to ensure that no compaction is successfully executed during the
rolling upgrade. This PR renames Compaction to CompactionV2, with the
following effects:
- New datacoord + old datanode: Utilizes the CompactionV2 interface,
resulting in the datanode error "CompactionV2 not implemented," causing
compaction to fail;
- Old datacoord + new datanode: Utilizes the CompactionV1 interface,
resulting in the datanode error "CompactionV1 not implemented," causing
compaction to fail.

issue: https://github.com/milvus-io/milvus/issues/32809

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
This commit is contained in:
yihao.dai 2024-06-16 22:07:57 +08:00 committed by GitHub
parent d43ec4db0b
commit 8537f3daeb
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 142 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/datacoord/mock_test.go
View File

@ -265,7 +265,7 @@ func (c *mockDataNodeClient) GetMetrics(ctx context.Context, req *milvuspb.GetMe
}, nil }, nil
} }
func (c *mockDataNodeClient) Compaction(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) { func (c *mockDataNodeClient) CompactionV2(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) {
if c.ch != nil { if c.ch != nil {
c.ch <- struct{}{} c.ch <- struct{}{}
if c.compactionResp != nil { if c.compactionResp != nil {

2
internal/datacoord/session_manager.go
View File

@ -213,7 +213,7 @@ func (c *SessionManagerImpl) Compaction(ctx context.Context, nodeID int64, plan
return err return err
} }
resp, err := cli.Compaction(ctx, plan) resp, err := cli.CompactionV2(ctx, plan)
if err := VerifyResponse(resp, err); err != nil { if err := VerifyResponse(resp, err); err != nil {
log.Warn("failed to execute compaction", zap.Int64("node", nodeID), zap.Error(err), zap.Int64("planID", plan.GetPlanID())) log.Warn("failed to execute compaction", zap.Int64("node", nodeID), zap.Error(err), zap.Int64("planID", plan.GetPlanID()))
return err return err

4
internal/datanode/services.go
View File

@ -200,9 +200,9 @@ func (node *DataNode) GetMetrics(ctx context.Context, req *milvuspb.GetMetricsRe
}, nil }, nil
} }
// Compaction handles compaction request from DataCoord // CompactionV2 handles compaction request from DataCoord
// returns status as long as compaction task enqueued or invalid // returns status as long as compaction task enqueued or invalid
func (node *DataNode) Compaction(ctx context.Context, req *datapb.CompactionPlan) (*commonpb.Status, error) { func (node *DataNode) CompactionV2(ctx context.Context, req *datapb.CompactionPlan) (*commonpb.Status, error) {
log := log.Ctx(ctx).With(zap.Int64("planID", req.GetPlanID())) log := log.Ctx(ctx).With(zap.Int64("planID", req.GetPlanID()))
if err := merr.CheckHealthy(node.GetStateCode()); err != nil { if err := merr.CheckHealthy(node.GetStateCode()); err != nil {
log.Warn("DataNode.Compaction failed", zap.Int64("nodeId", node.GetNodeID()), zap.Error(err)) log.Warn("DataNode.Compaction failed", zap.Int64("nodeId", node.GetNodeID()), zap.Error(err))

6
internal/datanode/services_test.go
View File

@ -226,7 +226,7 @@ func (s *DataNodeServicesSuite) TestCompaction() {
Channel: dmChannelName, Channel: dmChannelName,
} }
resp, err := node.Compaction(ctx, req) resp, err := node.CompactionV2(ctx, req)
s.NoError(err) s.NoError(err)
s.False(merr.Ok(resp)) s.False(merr.Ok(resp))
}) })
@ -245,7 +245,7 @@ func (s *DataNodeServicesSuite) TestCompaction() {
}, },
} }
resp, err := node.Compaction(ctx, req) resp, err := node.CompactionV2(ctx, req)
s.NoError(err) s.NoError(err)
s.False(merr.Ok(resp)) s.False(merr.Ok(resp))
}) })
@ -265,7 +265,7 @@ func (s *DataNodeServicesSuite) TestCompaction() {
Type: datapb.CompactionType_ClusteringCompaction, Type: datapb.CompactionType_ClusteringCompaction,
} }
_, err := node.Compaction(ctx, req) _, err := node.CompactionV2(ctx, req)
s.NoError(err) s.NoError(err)
}) })
} }

6
internal/distributed/datanode/client/client.go
View File

@ -173,10 +173,10 @@ func (c *Client) GetMetrics(ctx context.Context, req *milvuspb.GetMetricsRequest
}) })
} }
// Compaction return compaction by given plan // CompactionV2 return compaction by given plan
func (c *Client) Compaction(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) { func (c *Client) CompactionV2(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) {
return wrapGrpcCall(ctx, c, func(client datapb.DataNodeClient) (*commonpb.Status, error) { return wrapGrpcCall(ctx, c, func(client datapb.DataNodeClient) (*commonpb.Status, error) {
return client.Compaction(ctx, req) return client.CompactionV2(ctx, req)
}) })
} }

2
internal/distributed/datanode/client/client_test.go
View File

@ -66,7 +66,7 @@ func Test_NewClient(t *testing.T) {
r5, err := client.GetMetrics(ctx, nil) r5, err := client.GetMetrics(ctx, nil)
retCheck(retNotNil, r5, err) retCheck(retNotNil, r5, err)
r6, err := client.Compaction(ctx, nil) r6, err := client.CompactionV2(ctx, nil)
retCheck(retNotNil, r6, err) retCheck(retNotNil, r6, err)
r8, err := client.ResendSegmentStats(ctx, nil) r8, err := client.ResendSegmentStats(ctx, nil)

4
internal/distributed/datanode/service.go
View File

@ -354,8 +354,8 @@ func (s *Server) GetMetrics(ctx context.Context, request *milvuspb.GetMetricsReq
return s.datanode.GetMetrics(ctx, request) return s.datanode.GetMetrics(ctx, request)
} }
func (s *Server) Compaction(ctx context.Context, request *datapb.CompactionPlan) (*commonpb.Status, error) { func (s *Server) CompactionV2(ctx context.Context, request *datapb.CompactionPlan) (*commonpb.Status, error) {
return s.datanode.Compaction(ctx, request) return s.datanode.CompactionV2(ctx, request)
} }
// GetCompactionState gets the Compaction tasks state of DataNode // GetCompactionState gets the Compaction tasks state of DataNode

4
internal/distributed/datanode/service_test.go
View File

@ -126,7 +126,7 @@ func (m *MockDataNode) GetMetrics(ctx context.Context, request *milvuspb.GetMetr
return m.metricResp, m.err return m.metricResp, m.err
} }
func (m *MockDataNode) Compaction(ctx context.Context, req *datapb.CompactionPlan) (*commonpb.Status, error) { func (m *MockDataNode) CompactionV2(ctx context.Context, req *datapb.CompactionPlan) (*commonpb.Status, error) {
return m.status, m.err return m.status, m.err
} }
@ -289,7 +289,7 @@ func Test_NewServer(t *testing.T) {
server.datanode = &MockDataNode{ server.datanode = &MockDataNode{
status: &commonpb.Status{}, status: &commonpb.Status{},
} }
resp, err := server.Compaction(ctx, nil) resp, err := server.CompactionV2(ctx, nil)
assert.NoError(t, err) assert.NoError(t, err)
assert.NotNil(t, resp) assert.NotNil(t, resp)
}) })

20
internal/mocks/mock_datanode.go
View File

@ -87,8 +87,8 @@ func (_c *MockDataNode_CheckChannelOperationProgress_Call) RunAndReturn(run func
return _c return _c
} }
// Compaction provides a mock function with given fields: _a0, _a1 // CompactionV2 provides a mock function with given fields: _a0, _a1
func (_m *MockDataNode) Compaction(_a0 context.Context, _a1 *datapb.CompactionPlan) (*commonpb.Status, error) { func (_m *MockDataNode) CompactionV2(_a0 context.Context, _a1 *datapb.CompactionPlan) (*commonpb.Status, error) {
ret := _m.Called(_a0, _a1) ret := _m.Called(_a0, _a1)
var r0 *commonpb.Status var r0 *commonpb.Status
@ -113,31 +113,31 @@ func (_m *MockDataNode) Compaction(_a0 context.Context, _a1 *datapb.CompactionPl
return r0, r1 return r0, r1
} }
// MockDataNode_Compaction_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Compaction' // MockDataNode_CompactionV2_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CompactionV2'
type MockDataNode_Compaction_Call struct { type MockDataNode_CompactionV2_Call struct {
*mock.Call *mock.Call
} }
// Compaction is a helper method to define mock.On call // CompactionV2 is a helper method to define mock.On call
// - _a0 context.Context // - _a0 context.Context
// - _a1 *datapb.CompactionPlan // - _a1 *datapb.CompactionPlan
func (_e *MockDataNode_Expecter) Compaction(_a0 interface{}, _a1 interface{}) *MockDataNode_Compaction_Call { func (_e *MockDataNode_Expecter) CompactionV2(_a0 interface{}, _a1 interface{}) *MockDataNode_CompactionV2_Call {
return &MockDataNode_Compaction_Call{Call: _e.mock.On("Compaction", _a0, _a1)} return &MockDataNode_CompactionV2_Call{Call: _e.mock.On("CompactionV2", _a0, _a1)}
} }
func (_c *MockDataNode_Compaction_Call) Run(run func(_a0 context.Context, _a1 *datapb.CompactionPlan)) *MockDataNode_Compaction_Call { func (_c *MockDataNode_CompactionV2_Call) Run(run func(_a0 context.Context, _a1 *datapb.CompactionPlan)) *MockDataNode_CompactionV2_Call {
_c.Call.Run(func(args mock.Arguments) { _c.Call.Run(func(args mock.Arguments) {
run(args[0].(context.Context), args[1].(*datapb.CompactionPlan)) run(args[0].(context.Context), args[1].(*datapb.CompactionPlan))
}) })
return _c return _c
} }
func (_c *MockDataNode_Compaction_Call) Return(_a0 *commonpb.Status, _a1 error) *MockDataNode_Compaction_Call { func (_c *MockDataNode_CompactionV2_Call) Return(_a0 *commonpb.Status, _a1 error) *MockDataNode_CompactionV2_Call {
_c.Call.Return(_a0, _a1) _c.Call.Return(_a0, _a1)
return _c return _c
} }
func (_c *MockDataNode_Compaction_Call) RunAndReturn(run func(context.Context, *datapb.CompactionPlan) (*commonpb.Status, error)) *MockDataNode_Compaction_Call { func (_c *MockDataNode_CompactionV2_Call) RunAndReturn(run func(context.Context, *datapb.CompactionPlan) (*commonpb.Status, error)) *MockDataNode_CompactionV2_Call {
_c.Call.Return(run) _c.Call.Return(run)
return _c return _c
} }

20
internal/mocks/mock_datanode_client.go
View File

@ -142,8 +142,8 @@ func (_c *MockDataNodeClient_Close_Call) RunAndReturn(run func() error) *MockDat
return _c return _c
} }
// Compaction provides a mock function with given fields: ctx, in, opts // CompactionV2 provides a mock function with given fields: ctx, in, opts
func (_m *MockDataNodeClient) Compaction(ctx context.Context, in *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) { func (_m *MockDataNodeClient) CompactionV2(ctx context.Context, in *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) {
_va := make([]interface{}, len(opts)) _va := make([]interface{}, len(opts))
for _i := range opts { for _i := range opts {
_va[_i] = opts[_i] _va[_i] = opts[_i]
@ -175,21 +175,21 @@ func (_m *MockDataNodeClient) Compaction(ctx context.Context, in *datapb.Compact
return r0, r1 return r0, r1
} }
// MockDataNodeClient_Compaction_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Compaction' // MockDataNodeClient_CompactionV2_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'CompactionV2'
type MockDataNodeClient_Compaction_Call struct { type MockDataNodeClient_CompactionV2_Call struct {
*mock.Call *mock.Call
} }
// Compaction is a helper method to define mock.On call // CompactionV2 is a helper method to define mock.On call
// - ctx context.Context // - ctx context.Context
// - in *datapb.CompactionPlan // - in *datapb.CompactionPlan
// - opts ...grpc.CallOption // - opts ...grpc.CallOption
func (_e *MockDataNodeClient_Expecter) Compaction(ctx interface{}, in interface{}, opts ...interface{}) *MockDataNodeClient_Compaction_Call { func (_e *MockDataNodeClient_Expecter) CompactionV2(ctx interface{}, in interface{}, opts ...interface{}) *MockDataNodeClient_CompactionV2_Call {
return &MockDataNodeClient_Compaction_Call{Call: _e.mock.On("Compaction", return &MockDataNodeClient_CompactionV2_Call{Call: _e.mock.On("CompactionV2",
append([]interface{}{ctx, in}, opts...)...)} append([]interface{}{ctx, in}, opts...)...)}
} }
func (_c *MockDataNodeClient_Compaction_Call) Run(run func(ctx context.Context, in *datapb.CompactionPlan, opts ...grpc.CallOption)) *MockDataNodeClient_Compaction_Call { func (_c *MockDataNodeClient_CompactionV2_Call) Run(run func(ctx context.Context, in *datapb.CompactionPlan, opts ...grpc.CallOption)) *MockDataNodeClient_CompactionV2_Call {
_c.Call.Run(func(args mock.Arguments) { _c.Call.Run(func(args mock.Arguments) {
variadicArgs := make([]grpc.CallOption, len(args)-2) variadicArgs := make([]grpc.CallOption, len(args)-2)
for i, a := range args[2:] { for i, a := range args[2:] {
@ -202,12 +202,12 @@ func (_c *MockDataNodeClient_Compaction_Call) Run(run func(ctx context.Context,
return _c return _c
} }
func (_c *MockDataNodeClient_Compaction_Call) Return(_a0 *commonpb.Status, _a1 error) *MockDataNodeClient_Compaction_Call { func (_c *MockDataNodeClient_CompactionV2_Call) Return(_a0 *commonpb.Status, _a1 error) *MockDataNodeClient_CompactionV2_Call {
_c.Call.Return(_a0, _a1) _c.Call.Return(_a0, _a1)
return _c return _c
} }
func (_c *MockDataNodeClient_Compaction_Call) RunAndReturn(run func(context.Context, *datapb.CompactionPlan, ...grpc.CallOption) (*commonpb.Status, error)) *MockDataNodeClient_Compaction_Call { func (_c *MockDataNodeClient_CompactionV2_Call) RunAndReturn(run func(context.Context, *datapb.CompactionPlan, ...grpc.CallOption) (*commonpb.Status, error)) *MockDataNodeClient_CompactionV2_Call {
_c.Call.Return(run) _c.Call.Return(run)
return _c return _c
} }

2
internal/proto/data_coord.proto
View File

@ -110,7 +110,7 @@ service DataNode {
// https://wiki.lfaidata.foundation/display/MIL/MEP+8+--+Add+metrics+for+proxy // https://wiki.lfaidata.foundation/display/MIL/MEP+8+--+Add+metrics+for+proxy
rpc GetMetrics(milvus.GetMetricsRequest) returns (milvus.GetMetricsResponse) {} rpc GetMetrics(milvus.GetMetricsRequest) returns (milvus.GetMetricsResponse) {}
rpc Compaction(CompactionPlan) returns (common.Status) {} rpc CompactionV2(CompactionPlan) returns (common.Status) {}
rpc GetCompactionState(CompactionStateRequest) returns (CompactionStateResponse) {} rpc GetCompactionState(CompactionStateRequest) returns (CompactionStateResponse) {}
rpc SyncSegments(SyncSegmentsRequest) returns (common.Status) {} rpc SyncSegments(SyncSegmentsRequest) returns (common.Status) {}

2
internal/util/mock/grpc_datanode_client.go
View File

@ -57,7 +57,7 @@ func (m *GrpcDataNodeClient) GetMetrics(ctx context.Context, in *milvuspb.GetMet
return &milvuspb.GetMetricsResponse{}, m.Err return &milvuspb.GetMetricsResponse{}, m.Err
} }
func (m *GrpcDataNodeClient) Compaction(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) { func (m *GrpcDataNodeClient) CompactionV2(ctx context.Context, req *datapb.CompactionPlan, opts ...grpc.CallOption) (*commonpb.Status, error) {
return &commonpb.Status{}, m.Err return &commonpb.Status{}, m.Err
} }