-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
enhance: add rw/ro streaming query node replica management (#38677)
issue: #38399 - Embed the query node into streaming node to make delegator available at streaming node. - The embedded query node has a special server label `QUERYNODE_STREAMING-EMBEDDED`. - Change the balance strategy to make the channel assigned to streaming node as much as possible. Signed-off-by: chyezh <[email protected]>
- Loading branch information
Showing
32 changed files
with
1,145 additions
and
123 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
106 changes: 106 additions & 0 deletions
106
internal/coordinator/snmanager/streaming_node_manager.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| package snmanager | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| "github.com/cockroachdb/errors" | ||
| "go.uber.org/zap" | ||
|
|
||
| "github.com/milvus-io/milvus/internal/streamingcoord/server/balancer" | ||
| "github.com/milvus-io/milvus/pkg/log" | ||
| "github.com/milvus-io/milvus/pkg/streaming/util/types" | ||
| "github.com/milvus-io/milvus/pkg/util/funcutil" | ||
| "github.com/milvus-io/milvus/pkg/util/syncutil" | ||
| "github.com/milvus-io/milvus/pkg/util/typeutil" | ||
| ) | ||
|
|
||
| var StaticStreamingNodeManager = newStreamingNodeManager() | ||
|
|
||
| func newStreamingNodeManager() *StreamingNodeManager { | ||
| snm := &StreamingNodeManager{ | ||
| notifier: syncutil.NewAsyncTaskNotifier[struct{}](), | ||
| balancer: syncutil.NewFuture[balancer.Balancer](), | ||
| cond: syncutil.NewContextCond(&sync.Mutex{}), | ||
| latestAssignments: make(map[string]types.PChannelInfoAssigned), | ||
| streamingNodes: typeutil.NewUniqueSet(), | ||
| nodeChangedNotifier: syncutil.NewVersionedNotifier(), | ||
| } | ||
| go snm.execute() | ||
| return snm | ||
| } | ||
|
|
||
| // StreamingNodeManager is a manager for manage the querynode that embedded into streaming node. | ||
| // StreamingNodeManager is exclusive with ResourceManager. | ||
| type StreamingNodeManager struct { | ||
| notifier *syncutil.AsyncTaskNotifier[struct{}] | ||
| balancer *syncutil.Future[balancer.Balancer] | ||
| // The coord is merged after 2.6, so we don't need to make distribution safe. | ||
| cond *syncutil.ContextCond | ||
| latestAssignments map[string]types.PChannelInfoAssigned // The latest assignments info got from streaming coord balance module. | ||
| streamingNodes typeutil.UniqueSet | ||
| nodeChangedNotifier *syncutil.VersionedNotifier // used to notify that node in streaming node manager has been changed. | ||
| } | ||
|
|
||
| // GetWALLocated returns the server id of the node that the wal of the vChannel is located. | ||
| func (s *StreamingNodeManager) GetWALLocated(vChannel string) int64 { | ||
| pchannel := funcutil.ToPhysicalChannel(vChannel) | ||
| var targetServerID int64 | ||
|
|
||
| s.cond.L.Lock() | ||
| for { | ||
| if assignment, ok := s.latestAssignments[pchannel]; ok { | ||
| targetServerID = assignment.Node.ServerID | ||
| break | ||
| } | ||
| s.cond.Wait(context.Background()) | ||
| } | ||
| s.cond.L.Unlock() | ||
| return targetServerID | ||
| } | ||
|
|
||
| // GetStreamingQueryNodeIDs returns the server ids of the streaming query nodes. | ||
| func (s *StreamingNodeManager) GetStreamingQueryNodeIDs() typeutil.UniqueSet { | ||
| s.cond.L.Lock() | ||
| defer s.cond.L.Unlock() | ||
| return s.streamingNodes.Clone() | ||
| } | ||
|
|
||
| // ListenNodeChanged returns a listener for node changed event. | ||
| func (s *StreamingNodeManager) ListenNodeChanged() *syncutil.VersionedListener { | ||
| return s.nodeChangedNotifier.Listen(syncutil.VersionedListenAtEarliest) | ||
| } | ||
|
|
||
| // SetBalancerReady set the balancer ready for the streaming node manager from streamingcoord initialization. | ||
| func (s *StreamingNodeManager) SetBalancerReady(b balancer.Balancer) { | ||
| s.balancer.Set(b) | ||
| } | ||
|
|
||
| func (s *StreamingNodeManager) execute() (err error) { | ||
| defer s.notifier.Finish(struct{}{}) | ||
|
|
||
| balancer, err := s.balancer.GetWithContext(s.notifier.Context()) | ||
| if err != nil { | ||
| return errors.Wrap(err, "failed to wait balancer ready") | ||
| } | ||
| for { | ||
| if err := balancer.WatchChannelAssignments(s.notifier.Context(), func( | ||
| version typeutil.VersionInt64Pair, | ||
| relations []types.PChannelInfoAssigned, | ||
| ) error { | ||
| s.cond.LockAndBroadcast() | ||
| s.latestAssignments = make(map[string]types.PChannelInfoAssigned) | ||
| s.streamingNodes = typeutil.NewUniqueSet() | ||
| for _, relation := range relations { | ||
| s.latestAssignments[relation.Channel.Name] = relation | ||
| s.streamingNodes.Insert(relation.Node.ServerID) | ||
| } | ||
| s.nodeChangedNotifier.NotifyAll() | ||
| log.Info("streaming node manager updated", zap.Any("assignments", s.latestAssignments), zap.Any("streamingNodes", s.streamingNodes)) | ||
| s.cond.L.Unlock() | ||
| return nil | ||
| }); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } |
62 changes: 62 additions & 0 deletions
62
internal/coordinator/snmanager/streaming_node_manager_test.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package snmanager | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "github.com/stretchr/testify/mock" | ||
|
|
||
| "github.com/milvus-io/milvus/internal/mocks/streamingcoord/server/mock_balancer" | ||
| "github.com/milvus-io/milvus/pkg/streaming/util/types" | ||
| "github.com/milvus-io/milvus/pkg/util/typeutil" | ||
| ) | ||
|
|
||
| type pChannelInfoAssigned struct { | ||
| version typeutil.VersionInt64Pair | ||
| pchannels []types.PChannelInfoAssigned | ||
| } | ||
|
|
||
| func TestStreamingNodeManager(t *testing.T) { | ||
| m := newStreamingNodeManager() | ||
| b := mock_balancer.NewMockBalancer(t) | ||
|
|
||
| ch := make(chan pChannelInfoAssigned, 1) | ||
| b.EXPECT().WatchChannelAssignments(mock.Anything, mock.Anything).Run( | ||
| func(ctx context.Context, cb func(typeutil.VersionInt64Pair, []types.PChannelInfoAssigned) error) { | ||
| for { | ||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case p := <-ch: | ||
| cb(p.version, p.pchannels) | ||
| } | ||
| } | ||
| }) | ||
| m.SetBalancerReady(b) | ||
|
|
||
| streamingNodes := m.GetStreamingQueryNodeIDs() | ||
| assert.Empty(t, streamingNodes) | ||
|
|
||
| ch <- pChannelInfoAssigned{ | ||
| version: typeutil.VersionInt64Pair{ | ||
| Global: 1, | ||
| Local: 1, | ||
| }, | ||
| pchannels: []types.PChannelInfoAssigned{ | ||
| { | ||
| Channel: types.PChannelInfo{Name: "a_test", Term: 1}, | ||
| Node: types.StreamingNodeInfo{ServerID: 1, Address: "localhost:1"}, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| listener := m.ListenNodeChanged() | ||
| err := listener.Wait(context.Background()) | ||
| assert.NoError(t, err) | ||
|
|
||
| node := m.GetWALLocated("a_test") | ||
| assert.Equal(t, node, int64(1)) | ||
| streamingNodes = m.GetStreamingQueryNodeIDs() | ||
| assert.Equal(t, len(streamingNodes), 1) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.