From 5d6189c028fc33d01ac078a9f5ede7047e031809 Mon Sep 17 00:00:00 2001 From: chyezh Date: Thu, 11 Sep 2025 17:20:46 +0800 Subject: [PATCH] enhance: use local method for WAL().Balancer() interface Signed-off-by: chyezh --- .../snmanager/streaming_node_manager.go | 5 + internal/distributed/streaming/balancer.go | 138 +++++++++++++++--- .../distributed/streaming/balancer_test.go | 101 ++++++++++--- internal/streamingcoord/client/client.go | 1 + pkg/proto/streaming.proto | 4 +- pkg/proto/streamingpb/streaming.pb.go | 106 +++++++------- pkg/proto/streamingpb/streaming_grpc.pb.go | 3 + 7 files changed, 263 insertions(+), 95 deletions(-) diff --git a/internal/coordinator/snmanager/streaming_node_manager.go b/internal/coordinator/snmanager/streaming_node_manager.go index f9a10f4ac84..60c3dad50cd 100644 --- a/internal/coordinator/snmanager/streaming_node_manager.go +++ b/internal/coordinator/snmanager/streaming_node_manager.go @@ -71,6 +71,11 @@ type StreamingNodeManager struct { nodeChangedNotifier *syncutil.VersionedNotifier // used to notify that node in streaming node manager has been changed. } +// GetBalancer returns the balancer of the streaming node manager. +func (s *StreamingNodeManager) GetBalancer() balancer.Balancer { + return s.balancer.Get() +} + // GetLatestWALLocated returns the server id of the node that the wal of the vChannel is located. // Return -1 and error if the vchannel is not found or context is canceled. func (s *StreamingNodeManager) GetLatestWALLocated(ctx context.Context, vchannel string) (int64, error) { diff --git a/internal/distributed/streaming/balancer.go b/internal/distributed/streaming/balancer.go index 2d3b49e91c0..596115ba803 100644 --- a/internal/distributed/streaming/balancer.go +++ b/internal/distributed/streaming/balancer.go @@ -3,11 +3,16 @@ package streaming import ( "context" + "github.com/cockroachdb/errors" "google.golang.org/protobuf/types/known/fieldmaskpb" + "github.com/milvus-io/milvus/internal/coordinator/snmanager" + "github.com/milvus-io/milvus/internal/streamingcoord/server/balancer" "github.com/milvus-io/milvus/pkg/v2/proto/streamingpb" "github.com/milvus-io/milvus/pkg/v2/streaming/util/types" "github.com/milvus-io/milvus/pkg/v2/util/merr" + "github.com/milvus-io/milvus/pkg/v2/util/paramtable" + "github.com/milvus-io/milvus/pkg/v2/util/typeutil" ) type balancerImpl struct { @@ -16,49 +21,94 @@ type balancerImpl struct { // GetWALDistribution returns the wal distribution of the streaming node. func (b balancerImpl) ListStreamingNode(ctx context.Context) ([]types.StreamingNodeInfo, error) { - assignments, err := b.streamingCoordClient.Assignment().GetLatestAssignments(ctx) + ready, err := b.checkIfStreamingServiceReady(ctx) if err != nil { return nil, err } + if !ready { + return nil, nil + } - nodes := make([]types.StreamingNodeInfo, 0, len(assignments.Assignments)) - for _, assignment := range assignments.Assignments { - nodes = append(nodes, assignment.NodeInfo) + nodes, err := snmanager.StaticStreamingNodeManager.GetBalancer().GetAllStreamingNodes(ctx) + if err != nil { + return nil, err } - return nodes, nil + nodeInfos := make([]types.StreamingNodeInfo, 0, len(nodes)) + for _, node := range nodes { + nodeInfos = append(nodeInfos, *node) + } + return nodeInfos, nil } // GetWALDistribution returns the wal distribution of the streaming node. func (b balancerImpl) GetWALDistribution(ctx context.Context, nodeID int64) (*types.StreamingNodeAssignment, error) { - assignments, err := b.streamingCoordClient.Assignment().GetLatestAssignments(ctx) + ready, err := b.checkIfStreamingServiceReady(ctx) if err != nil { return nil, err } - for _, assignment := range assignments.Assignments { - if assignment.NodeInfo.ServerID == nodeID { - return &assignment, nil + if !ready { + return nil, nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + var result *types.StreamingNodeAssignment + stopErr := errors.New("stop watching") + err = sbalancer.WatchChannelAssignments(ctx, func(param balancer.WatchChannelAssignmentsCallbackParam) error { + for _, assignment := range param.Relations { + if assignment.Node.ServerID == nodeID { + if result == nil { + result = &types.StreamingNodeAssignment{ + NodeInfo: assignment.Node, + Channels: make(map[string]types.PChannelInfo), + } + } + result.Channels[assignment.Channel.Name] = assignment.Channel + } + } + return errors.New("stop watching") + }) + if errors.Is(err, stopErr) { + if result == nil { + return nil, merr.ErrNodeNotFound } + return result, nil } - return nil, merr.WrapErrNodeNotFound(nodeID, "streaming node not found") + return nil, err } // GetFrozenNodeIDs returns the frozen node ids. func (b balancerImpl) GetFrozenNodeIDs(ctx context.Context) ([]int64, error) { - // Update nothing, just fetch the current resp back. - resp, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return nil, err + } + if !ready { + return nil, nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + resp, err := sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ Config: &streamingpb.WALBalancePolicyConfig{}, UpdateMask: &fieldmaskpb.FieldMask{}, }) if err != nil { return nil, err } - return resp.GetFreezeNodeIds(), nil + return resp.FreezeNodeIds, nil } // IsRebalanceSuspended returns whether the rebalance of the wal is suspended. func (b balancerImpl) IsRebalanceSuspended(ctx context.Context) (bool, error) { - // Update nothing, just fetch the current resp back. - resp, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return false, err + } + if !ready { + return false, nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + resp, err := sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ Config: &streamingpb.WALBalancePolicyConfig{}, UpdateMask: &fieldmaskpb.FieldMask{}, }) @@ -69,7 +119,16 @@ func (b balancerImpl) IsRebalanceSuspended(ctx context.Context) (bool, error) { } func (b balancerImpl) SuspendRebalance(ctx context.Context) error { - _, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return err + } + if !ready { + return nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + _, err = sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ Config: &streamingpb.WALBalancePolicyConfig{ AllowRebalance: false, }, @@ -81,7 +140,16 @@ func (b balancerImpl) SuspendRebalance(ctx context.Context) error { } func (b balancerImpl) ResumeRebalance(ctx context.Context) error { - _, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return err + } + if !ready { + return nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + _, err = sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ Config: &streamingpb.WALBalancePolicyConfig{ AllowRebalance: true, }, @@ -93,7 +161,16 @@ func (b balancerImpl) ResumeRebalance(ctx context.Context) error { } func (b balancerImpl) FreezeNodeIDs(ctx context.Context, nodeIDs []int64) error { - _, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return err + } + if !ready { + return nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + _, err = sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{}}, Nodes: &streamingpb.WALBalancePolicyNodes{ FreezeNodeIds: nodeIDs, @@ -103,7 +180,16 @@ func (b balancerImpl) FreezeNodeIDs(ctx context.Context, nodeIDs []int64) error } func (b balancerImpl) DefreezeNodeIDs(ctx context.Context, nodeIDs []int64) error { - _, err := b.streamingCoordClient.Assignment().UpdateWALBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ + ready, err := b.checkIfStreamingServiceReady(ctx) + if err != nil { + return err + } + if !ready { + return nil + } + + sbalancer := snmanager.StaticStreamingNodeManager.GetBalancer() + _, err = sbalancer.UpdateBalancePolicy(ctx, &types.UpdateWALBalancePolicyRequest{ UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{}}, Nodes: &streamingpb.WALBalancePolicyNodes{ DefreezeNodeIds: nodeIDs, @@ -111,3 +197,17 @@ func (b balancerImpl) DefreezeNodeIDs(ctx context.Context, nodeIDs []int64) erro }) return err } + +func (b balancerImpl) checkIfStreamingServiceReady(ctx context.Context) (bool, error) { + if !paramtable.IsLocalComponentEnabled(typeutil.MixCoordRole) { + panic("should be only called at mix coord") + } + if err := snmanager.StaticStreamingNodeManager.CheckIfStreamingServiceReady(ctx); err != nil { + if errors.Is(err, snmanager.ErrStreamingServiceNotReady) { + // for 2.5.x compatibility, return empty result when streaming service is not ready. + return false, nil + } + return false, err + } + return true, nil +} diff --git a/internal/distributed/streaming/balancer_test.go b/internal/distributed/streaming/balancer_test.go index fd1a6f7ac6d..c4566519190 100644 --- a/internal/distributed/streaming/balancer_test.go +++ b/internal/distributed/streaming/balancer_test.go @@ -3,59 +3,87 @@ package streaming import ( "context" "testing" + "time" "github.com/cockroachdb/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" - "github.com/milvus-io/milvus/internal/mocks/streamingcoord/mock_client" + "github.com/milvus-io/milvus/internal/coordinator/snmanager" + "github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_balancer" + "github.com/milvus-io/milvus/internal/streamingcoord/server/balancer" + "github.com/milvus-io/milvus/pkg/v2/proto/streamingpb" "github.com/milvus-io/milvus/pkg/v2/streaming/util/types" "github.com/milvus-io/milvus/pkg/v2/util/merr" + "github.com/milvus-io/milvus/pkg/v2/util/paramtable" + "github.com/milvus-io/milvus/pkg/v2/util/syncutil" + "github.com/milvus-io/milvus/pkg/v2/util/typeutil" ) func TestBalancer(t *testing.T) { - scClient := mock_client.NewMockClient(t) - assignmentService := mock_client.NewMockAssignmentService(t) - scClient.EXPECT().Assignment().Return(assignmentService) - assignmentService.EXPECT().GetLatestAssignments(mock.Anything).Return(&types.VersionedStreamingNodeAssignments{ - Assignments: map[int64]types.StreamingNodeAssignment{ - 1: { - NodeInfo: types.StreamingNodeInfo{ServerID: 1}, - Channels: map[string]types.PChannelInfo{ - "v1": {}, + paramtable.SetLocalComponentEnabled(typeutil.MixCoordRole) + sbalancer := mock_balancer.NewMockBalancer(t) + sbalancer.EXPECT().GetAllStreamingNodes(mock.Anything).Return(map[int64]*types.StreamingNodeInfo{ + 1: {ServerID: 1}, + 2: {ServerID: 2}, + }, nil) + sbalancer.EXPECT().WatchChannelAssignments(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, cb balancer.WatchChannelAssignmentsCallback) error { + if err := cb(balancer.WatchChannelAssignmentsCallbackParam{ + Version: typeutil.VersionInt64Pair{ + Global: 1, + Local: 1, + }, + Relations: []types.PChannelInfoAssigned{ + { + Channel: types.PChannelInfo{Name: "v1"}, + Node: types.StreamingNodeInfo{ServerID: 1}, + }, + { + Channel: types.PChannelInfo{Name: "v2"}, + Node: types.StreamingNodeInfo{ServerID: 1}, }, }, - }, - }, nil) + }); err != nil { + return err + } + time.Sleep(100 * time.Millisecond) + return nil + }) + sbalancer.EXPECT().RegisterStreamingEnabledNotifier(mock.Anything).RunAndReturn(func(notifier *syncutil.AsyncTaskNotifier[struct{}]) { + notifier.Cancel() + }) + + snmanager.ResetStreamingNodeManager() + snmanager.StaticStreamingNodeManager.SetBalancerReady(sbalancer) balancer := balancerImpl{ - walAccesserImpl: &walAccesserImpl{ - streamingCoordClient: scClient, - }, + walAccesserImpl: &walAccesserImpl{}, } nodes, err := balancer.ListStreamingNode(context.Background()) assert.NoError(t, err) - assert.Equal(t, 1, len(nodes)) + assert.Equal(t, 2, len(nodes)) assignment, err := balancer.GetWALDistribution(context.Background(), 1) assert.NoError(t, err) - assert.Equal(t, 1, len(assignment.Channels)) + assert.Equal(t, 2, len(assignment.Channels)) assignment, err = balancer.GetWALDistribution(context.Background(), 2) assert.True(t, errors.Is(err, merr.ErrNodeNotFound)) assert.Nil(t, assignment) - assignmentService.EXPECT().GetLatestAssignments(mock.Anything).Unset() - assignmentService.EXPECT().GetLatestAssignments(mock.Anything).Return(nil, errors.New("test")) + sbalancer.EXPECT().GetAllStreamingNodes(mock.Anything).Unset() + sbalancer.EXPECT().GetAllStreamingNodes(mock.Anything).Return(nil, errors.New("test")) nodes, err = balancer.ListStreamingNode(context.Background()) assert.Error(t, err) assert.Nil(t, nodes) + sbalancer.EXPECT().WatchChannelAssignments(mock.Anything, mock.Anything).Unset() + sbalancer.EXPECT().WatchChannelAssignments(mock.Anything, mock.Anything).Return(errors.New("test")) assignment, err = balancer.GetWALDistribution(context.Background(), 1) assert.Error(t, err) assert.Nil(t, assignment) - assignmentService.EXPECT().UpdateWALBalancePolicy(mock.Anything, mock.Anything).Return(&types.UpdateWALBalancePolicyResponse{}, nil) + sbalancer.EXPECT().UpdateBalancePolicy(mock.Anything, mock.Anything).Return(&streamingpb.UpdateWALBalancePolicyResponse{}, nil) err = balancer.SuspendRebalance(context.Background()) assert.NoError(t, err) err = balancer.ResumeRebalance(context.Background()) @@ -64,9 +92,13 @@ func TestBalancer(t *testing.T) { assert.NoError(t, err) err = balancer.DefreezeNodeIDs(context.Background(), []int64{1}) assert.NoError(t, err) + _, err = balancer.GetFrozenNodeIDs(context.Background()) + assert.NoError(t, err) + _, err = balancer.IsRebalanceSuspended(context.Background()) + assert.NoError(t, err) - assignmentService.EXPECT().UpdateWALBalancePolicy(mock.Anything, mock.Anything).Unset() - assignmentService.EXPECT().UpdateWALBalancePolicy(mock.Anything, mock.Anything).Return(nil, errors.New("test")) + sbalancer.EXPECT().UpdateBalancePolicy(mock.Anything, mock.Anything).Unset() + sbalancer.EXPECT().UpdateBalancePolicy(mock.Anything, mock.Anything).Return(nil, errors.New("test")) err = balancer.SuspendRebalance(context.Background()) assert.Error(t, err) err = balancer.ResumeRebalance(context.Background()) @@ -75,4 +107,29 @@ func TestBalancer(t *testing.T) { assert.Error(t, err) err = balancer.DefreezeNodeIDs(context.Background(), []int64{1}) assert.Error(t, err) + _, err = balancer.GetFrozenNodeIDs(context.Background()) + assert.Error(t, err) + _, err = balancer.IsRebalanceSuspended(context.Background()) + assert.Error(t, err) + + sbalancer.EXPECT().RegisterStreamingEnabledNotifier(mock.Anything).Unset() + sbalancer.EXPECT().RegisterStreamingEnabledNotifier(mock.Anything).RunAndReturn(func(notifier *syncutil.AsyncTaskNotifier[struct{}]) { + }) + + _, err = balancer.ListStreamingNode(context.Background()) + assert.NoError(t, err) + _, err = balancer.GetWALDistribution(context.Background(), 1) + assert.NoError(t, err) + err = balancer.SuspendRebalance(context.Background()) + assert.NoError(t, err) + err = balancer.ResumeRebalance(context.Background()) + assert.NoError(t, err) + err = balancer.FreezeNodeIDs(context.Background(), []int64{1}) + assert.NoError(t, err) + err = balancer.DefreezeNodeIDs(context.Background(), []int64{1}) + assert.NoError(t, err) + _, err = balancer.GetFrozenNodeIDs(context.Background()) + assert.NoError(t, err) + _, err = balancer.IsRebalanceSuspended(context.Background()) + assert.NoError(t, err) } diff --git a/internal/streamingcoord/client/client.go b/internal/streamingcoord/client/client.go index 6c8ee7d5e98..59f7d2b324a 100644 --- a/internal/streamingcoord/client/client.go +++ b/internal/streamingcoord/client/client.go @@ -38,6 +38,7 @@ type AssignmentService interface { // UpdateWALBalancePolicy is used to update the WAL balance policy. // Return the WAL balance policy after the update. + // Deprecated: This function is deprecated and will be removed in the future. UpdateWALBalancePolicy(ctx context.Context, req *types.UpdateWALBalancePolicyRequest) (*types.UpdateWALBalancePolicyResponse, error) } diff --git a/pkg/proto/streaming.proto b/pkg/proto/streaming.proto index 663f12ee598..599015aa854 100644 --- a/pkg/proto/streaming.proto +++ b/pkg/proto/streaming.proto @@ -152,7 +152,9 @@ message BroadcastAckResponse { service StreamingCoordAssignmentService { // UpdateWALBalancePolicy is used to update the WAL balance policy. // The policy is used to control the balance of the WAL. - rpc UpdateWALBalancePolicy(UpdateWALBalancePolicyRequest) returns (UpdateWALBalancePolicyResponse) {}; + rpc UpdateWALBalancePolicy(UpdateWALBalancePolicyRequest) returns (UpdateWALBalancePolicyResponse) { + option deprecated = true; + }; // AssignmentDiscover is used to discover all log nodes managed by the // streamingcoord. Channel assignment information will be pushed to client diff --git a/pkg/proto/streamingpb/streaming.pb.go b/pkg/proto/streamingpb/streaming.pb.go index eaa1f6d9ef0..4b4f31691ee 100644 --- a/pkg/proto/streamingpb/streaming.pb.go +++ b/pkg/proto/streamingpb/streaming.pb.go @@ -5403,9 +5403,9 @@ var file_streaming_proto_rawDesc = []byte{ 0x75, 0x65, 0x73, 0x74, 0x1a, 0x2c, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x42, 0x72, 0x6f, 0x61, 0x64, 0x63, 0x61, 0x73, 0x74, 0x41, 0x63, 0x6b, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x22, 0x00, 0x32, 0xb1, 0x02, 0x0a, 0x1f, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, + 0x73, 0x65, 0x22, 0x00, 0x32, 0xb4, 0x02, 0x0a, 0x1f, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x43, 0x6f, 0x6f, 0x72, 0x64, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, - 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x89, 0x01, 0x0a, 0x16, 0x55, 0x70, 0x64, + 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x8c, 0x01, 0x0a, 0x16, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x57, 0x41, 0x4c, 0x42, 0x61, 0x6c, 0x61, 0x6e, 0x63, 0x65, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x35, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x55, 0x70, 0x64, @@ -5414,61 +5414,61 @@ var file_streaming_proto_rawDesc = []byte{ 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x55, 0x70, 0x64, 0x61, 0x74, 0x65, 0x57, 0x41, 0x4c, 0x42, 0x61, 0x6c, 0x61, 0x6e, 0x63, 0x65, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x22, 0x00, 0x12, 0x81, 0x01, 0x0a, 0x12, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, - 0x65, 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x12, 0x31, 0x2e, 0x6d, 0x69, - 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x44, - 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x32, + 0x73, 0x65, 0x22, 0x03, 0x88, 0x02, 0x01, 0x12, 0x81, 0x01, 0x0a, 0x12, 0x41, 0x73, 0x73, 0x69, + 0x67, 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x12, 0x31, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x6d, 0x65, - 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x32, 0xe1, 0x01, 0x0a, 0x1b, 0x53, 0x74, 0x72, - 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x48, 0x61, 0x6e, 0x64, 0x6c, 0x65, - 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x60, 0x0a, 0x07, 0x50, 0x72, 0x6f, 0x64, - 0x75, 0x63, 0x65, 0x12, 0x26, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x50, 0x72, 0x6f, - 0x64, 0x75, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, 0x6d, 0x69, - 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x50, 0x72, 0x6f, 0x64, 0x75, 0x63, 0x65, 0x52, 0x65, 0x73, 0x70, - 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x60, 0x0a, 0x07, 0x43, 0x6f, - 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x12, 0x26, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, - 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x43, - 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, 0x2e, - 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, - 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x52, 0x65, - 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x32, 0xbe, 0x03, 0x0a, - 0x1b, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, - 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x81, 0x01, 0x0a, - 0x06, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x12, 0x39, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, - 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, - 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, - 0x6e, 0x61, 0x67, 0x65, 0x72, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, - 0x73, 0x74, 0x1a, 0x3a, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, - 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, - 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, - 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, - 0x12, 0x81, 0x01, 0x0a, 0x06, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x12, 0x39, 0x2e, 0x6d, 0x69, - 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, - 0x64, 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x52, - 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x3a, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, - 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, - 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x6e, - 0x61, 0x67, 0x65, 0x72, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x22, 0x00, 0x12, 0x96, 0x01, 0x0a, 0x0d, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, - 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x40, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, + 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x32, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x41, 0x73, 0x73, 0x69, 0x67, + 0x6e, 0x6d, 0x65, 0x6e, 0x74, 0x44, 0x69, 0x73, 0x63, 0x6f, 0x76, 0x65, 0x72, 0x52, 0x65, 0x73, + 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x32, 0xe1, 0x01, 0x0a, 0x1b, + 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x48, 0x61, 0x6e, + 0x64, 0x6c, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x60, 0x0a, 0x07, 0x50, + 0x72, 0x6f, 0x64, 0x75, 0x63, 0x65, 0x12, 0x26, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, - 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x6e, - 0x61, 0x67, 0x65, 0x72, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, - 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x41, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, + 0x50, 0x72, 0x6f, 0x64, 0x75, 0x63, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x27, + 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, + 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x50, 0x72, 0x6f, 0x64, 0x75, 0x63, 0x65, 0x52, + 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x12, 0x60, 0x0a, + 0x07, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x12, 0x26, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, - 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, - 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x53, 0x74, 0x61, - 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x36, 0x5a, - 0x34, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6d, 0x69, 0x6c, 0x76, - 0x75, 0x73, 0x2d, 0x69, 0x6f, 0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2f, 0x70, 0x6b, 0x67, - 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, - 0x69, 0x6e, 0x67, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x67, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, + 0x1a, 0x27, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, + 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, + 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01, 0x32, + 0xbe, 0x03, 0x0a, 0x1b, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, + 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, + 0x81, 0x01, 0x0a, 0x06, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x12, 0x39, 0x2e, 0x6d, 0x69, 0x6c, + 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x69, 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, + 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x3a, 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, + 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x53, + 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x6e, 0x61, + 0x67, 0x65, 0x72, 0x41, 0x73, 0x73, 0x69, 0x67, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, + 0x65, 0x22, 0x00, 0x12, 0x81, 0x01, 0x0a, 0x06, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x12, 0x39, + 0x2e, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, + 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, + 0x67, 0x4e, 0x6f, 0x64, 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x52, 0x65, 0x6d, 0x6f, + 0x76, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x3a, 0x2e, 0x6d, 0x69, 0x6c, 0x76, + 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, + 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, + 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x52, 0x65, 0x6d, 0x6f, 0x76, 0x65, 0x52, 0x65, 0x73, + 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x96, 0x01, 0x0a, 0x0d, 0x43, 0x6f, 0x6c, 0x6c, + 0x65, 0x63, 0x74, 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x12, 0x40, 0x2e, 0x6d, 0x69, 0x6c, 0x76, + 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, + 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, 0x64, 0x65, + 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, 0x53, 0x74, + 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x41, 0x2e, 0x6d, 0x69, + 0x6c, 0x76, 0x75, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x73, 0x74, 0x72, 0x65, 0x61, + 0x6d, 0x69, 0x6e, 0x67, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x4e, 0x6f, + 0x64, 0x65, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x43, 0x6f, 0x6c, 0x6c, 0x65, 0x63, 0x74, + 0x53, 0x74, 0x61, 0x74, 0x75, 0x73, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, + 0x42, 0x36, 0x5a, 0x34, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x6d, + 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2d, 0x69, 0x6f, 0x2f, 0x6d, 0x69, 0x6c, 0x76, 0x75, 0x73, 0x2f, + 0x70, 0x6b, 0x67, 0x2f, 0x76, 0x32, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x73, 0x74, 0x72, + 0x65, 0x61, 0x6d, 0x69, 0x6e, 0x67, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( diff --git a/pkg/proto/streamingpb/streaming_grpc.pb.go b/pkg/proto/streamingpb/streaming_grpc.pb.go index 8b2a8cf6f75..7bd3a172405 100644 --- a/pkg/proto/streamingpb/streaming_grpc.pb.go +++ b/pkg/proto/streamingpb/streaming_grpc.pb.go @@ -247,6 +247,7 @@ const ( // // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type StreamingCoordAssignmentServiceClient interface { + // Deprecated: Do not use. // UpdateWALBalancePolicy is used to update the WAL balance policy. // The policy is used to control the balance of the WAL. UpdateWALBalancePolicy(ctx context.Context, in *UpdateWALBalancePolicyRequest, opts ...grpc.CallOption) (*UpdateWALBalancePolicyResponse, error) @@ -264,6 +265,7 @@ func NewStreamingCoordAssignmentServiceClient(cc grpc.ClientConnInterface) Strea return &streamingCoordAssignmentServiceClient{cc} } +// Deprecated: Do not use. func (c *streamingCoordAssignmentServiceClient) UpdateWALBalancePolicy(ctx context.Context, in *UpdateWALBalancePolicyRequest, opts ...grpc.CallOption) (*UpdateWALBalancePolicyResponse, error) { out := new(UpdateWALBalancePolicyResponse) err := c.cc.Invoke(ctx, StreamingCoordAssignmentService_UpdateWALBalancePolicy_FullMethodName, in, out, opts...) @@ -308,6 +310,7 @@ func (x *streamingCoordAssignmentServiceAssignmentDiscoverClient) Recv() (*Assig // All implementations should embed UnimplementedStreamingCoordAssignmentServiceServer // for forward compatibility type StreamingCoordAssignmentServiceServer interface { + // Deprecated: Do not use. // UpdateWALBalancePolicy is used to update the WAL balance policy. // The policy is used to control the balance of the WAL. UpdateWALBalancePolicy(context.Context, *UpdateWALBalancePolicyRequest) (*UpdateWALBalancePolicyResponse, error)