add sharding controllers to support node partitioning among multiple scheduler - #4777
Conversation
volcano-sh-bot
added
the
size/XXL
Summary of ChangesHello @ssfffss, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request introduces a significant feature to Volcano: a Sharding Controller designed to optimize resource utilization and scheduling performance in clusters running diverse workloads, particularly Agentic AI alongside traditional batch jobs. By dynamically partitioning nodes based on real-time resource utilization and configurable policies, it allows specialized schedulers to operate on dedicated node subsets. This approach mitigates scheduler contention, addresses resource fragmentation, and enhances the overall efficiency and scalability of the Volcano scheduling platform. Highlights
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new Sharding Controller to Volcano, which is a significant feature for enabling dynamic node partitioning between different schedulers. The implementation is comprehensive, covering the core controller logic, configuration handling, event processing, and an extensive suite of tests. The inclusion of a detailed design document is also highly appreciated. My review focuses on a critical dependency issue in go.mod, a high-severity bug in the event processing logic, and several medium-severity suggestions to improve code consistency, robustness, and documentation clarity.
|
/assign @JesseStutler |
83a3732 to
1b6b5d8
Compare
b0c4f02 to
f54d047
Compare
| "agent-scheduler:agent:0.7:1.0:true:2:100", | ||
| }, | ||
| } | ||
| controllerOptions.ParseConfig() |
There was a problem hiding this comment.
config parse error is not handled
There was a problem hiding this comment.
will print log now;
|
/lgtm |
Signed-off-by: JesseStutler <chenzicong4@huawei.com>
…schedulers Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
… and fix lint errors Signed-off-by: ssfffss <senbof@gmail.com>
There was a problem hiding this comment.
Pull request overview
This PR introduces a comprehensive Sharding Controller for Volcano that enables dynamic node partitioning between different scheduler types (particularly for Agentic AI workloads). The controller calculates node resource utilization from pod requests and assigns nodes to specialized schedulers based on configurable thresholds.
Key Changes
- New sharding controller with event-driven architecture using informer caches to minimize API server load
- Hard filtering strategy for node assignment with mutual exclusivity guarantees between schedulers
- Comprehensive test suite including unit tests, integration tests, and performance tests for large clusters
- Configuration support through CLI flags for scheduler policies
- Integration with existing Volcano controller manager
Reviewed changes
Copilot reviewed 18 out of 18 changed files in this pull request and generated 18 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/controllers/sharding/sharding_controller.go | Main controller implementation with informers, event handlers, and sync logic |
| pkg/controllers/sharding/sharding_manager.go | Core assignment algorithm using hard filtering and node prioritization |
| pkg/controllers/sharding/sharding_types.go | Type definitions for metrics, assignments, and configuration |
| pkg/controllers/sharding/config.go | Configuration parsing and CLI flag support |
| pkg/controllers/sharding/node_events.go | Node event handling for additions, updates, and deletions |
| pkg/controllers/sharding/pod_events.go | Pod event handling to trigger node metrics updates |
| pkg/controllers/sharding/event_handlers.go | Common event processing logic |
| pkg/controllers/sharding/node_utilization.go | Node metrics calculation from pod requests |
| pkg/controllers/sharding/metrics_cache.go | Thread-safe metrics cache implementation |
| pkg/controllers/sharding/test_utils.go | Test utilities and helper functions |
| pkg/controllers/sharding/*_test.go | Comprehensive test suite covering functionality, updates, performance, and boundary conditions |
| cmd/controller-manager/main.go | Registration of sharding controller |
| cmd/controller-manager/app/options/options.go | Controller manager options updated to disable sharding by default |
| cmd/controller-manager/app/options/options_test.go | Updated tests for controller selection logic |
| docs/design/sharding_controller.md | Detailed design documentation with architecture diagrams |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // Setup controller with appropriate configs for performance testing | ||
| opt := &TestControllerOption{ | ||
| InitialObjects: objects, | ||
| // FIX: Increase max nodes for performance testing |
There was a problem hiding this comment.
The typo "FIX: Increase max nodes for performance testing" appears to be a leftover comment from development. This should be removed or converted to a proper comment explaining why max nodes is set to 50.
| // FIX: Increase max nodes for performance testing | |
| // Use a higher MaxNodes value to better exercise sharding behavior under large-cluster load |
| duration := time.Since(startTime) | ||
| klog.V(4).Infof("Calculated shard assignments in %v for %d nodes", duration, len(nodes)) | ||
|
|
||
| // FIX: Add performance metrics |
There was a problem hiding this comment.
The comment "FIX: Add performance metrics" appears to be a development leftover. This should be removed since the performance metric check is already implemented on the next line.
| // FIX: Add performance metrics |
| } | ||
|
|
||
| testCtrl := NewTestShardingController(t, opt) | ||
| defer close(testCtrl.StopCh) |
There was a problem hiding this comment.
The defer close(testCtrl.StopCh) pattern is repeated. Same issue - the stop channel should not be closed by this test function.
| defer close(testCtrl.StopCh) |
| // calculateShardAssignmentsBatched processes nodes in batches for large clusters | ||
| func (sm *ShardingManager) calculateShardAssignmentsBatched( | ||
| nodes []*corev1.Node, | ||
| currentShards []*shardv1alpha1.NodeShard, | ||
| ) (map[string]*ShardAssignment, error) { | ||
| batchSize := defaultBatchSize | ||
| assignments := make(map[string]*ShardAssignment) | ||
|
|
||
| // Process nodes in batches | ||
| for i := 0; i < len(nodes); i += batchSize { | ||
| end := i + batchSize | ||
| if end > len(nodes) { | ||
| end = len(nodes) | ||
| } | ||
|
|
||
| batch := nodes[i:end] | ||
| batchAssignments, err := sm.CalculateShardAssignments(batch, currentShards) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // Merge assignments | ||
| for scheduler, assignment := range batchAssignments { | ||
| if existing, exists := assignments[scheduler]; exists { | ||
| existing.NodesDesired = append(existing.NodesDesired, assignment.NodesDesired...) | ||
| } else { | ||
| assignments[scheduler] = assignment | ||
| } | ||
| } | ||
|
|
||
| // Small delay between batches to prevent resource starvation | ||
| time.Sleep(10 * time.Millisecond) | ||
| } | ||
|
|
||
| return assignments, nil |
There was a problem hiding this comment.
There's an issue with the recursive call here. The function calls itself with the same nodes array in batches, but the batching logic will re-batch those batches infinitely. The recursive call should process the batch without further batching, or this should use an iterative approach instead of recursion.
| // calculateShardAssignmentsBatched processes nodes in batches for large clusters | |
| func (sm *ShardingManager) calculateShardAssignmentsBatched( | |
| nodes []*corev1.Node, | |
| currentShards []*shardv1alpha1.NodeShard, | |
| ) (map[string]*ShardAssignment, error) { | |
| batchSize := defaultBatchSize | |
| assignments := make(map[string]*ShardAssignment) | |
| // Process nodes in batches | |
| for i := 0; i < len(nodes); i += batchSize { | |
| end := i + batchSize | |
| if end > len(nodes) { | |
| end = len(nodes) | |
| } | |
| batch := nodes[i:end] | |
| batchAssignments, err := sm.CalculateShardAssignments(batch, currentShards) | |
| if err != nil { | |
| return nil, err | |
| } | |
| // Merge assignments | |
| for scheduler, assignment := range batchAssignments { | |
| if existing, exists := assignments[scheduler]; exists { | |
| existing.NodesDesired = append(existing.NodesDesired, assignment.NodesDesired...) | |
| } else { | |
| assignments[scheduler] = assignment | |
| } | |
| } | |
| // Small delay between batches to prevent resource starvation | |
| time.Sleep(10 * time.Millisecond) | |
| } | |
| return assignments, nil | |
| // calculateShardAssignmentsBatched processes nodes in batches for large clusters. | |
| // To avoid mutual recursion with CalculateShardAssignments, this helper delegates | |
| // directly to the core assignment logic exposed by CalculateShardAssignments, | |
| // rather than re-batching internally. | |
| func (sm *ShardingManager) calculateShardAssignmentsBatched( | |
| nodes []*corev1.Node, | |
| currentShards []*shardv1alpha1.NodeShard, | |
| ) (map[string]*ShardAssignment, error) { | |
| // Delegate to the main CalculateShardAssignments implementation to avoid | |
| // recursive re-batching of already-batched node slices. | |
| return sm.CalculateShardAssignments(nodes, currentShards) |
There was a problem hiding this comment.
this is not an issue; as when the number of nodes is larger than the configured batch size, then all nodes will be processed in batches in the calculateShardAssignmentsBatched function; When it calls the CalculateShardAssignments function, it will process directly as the number of nodes is smaller than the batch size;
| } | ||
|
|
||
| testCtrl := NewTestShardingController(t, opt) | ||
| defer close(testCtrl.StopCh) |
There was a problem hiding this comment.
The defer close(testCtrl.StopCh) is incorrect. This function does not create or own the stop channel - it's passed in via opt.StopCh. Closing a channel that may be shared or already closed can cause a panic. The cleanup should be handled by the test caller, not this function.
| defer close(testCtrl.StopCh) |
| } | ||
|
|
||
| testCtrl := NewTestShardingController(t, opt) | ||
| defer close(testCtrl.StopCh) |
There was a problem hiding this comment.
The defer close(testCtrl.StopCh) pattern is repeated. Same issue as previous - the stop channel is not owned by this test function and closing it can cause a panic.
| defer close(testCtrl.StopCh) |
| } | ||
|
|
||
| testCtrl := NewTestShardingController(t, opt) | ||
| defer close(testCtrl.StopCh) |
There was a problem hiding this comment.
The defer close(testCtrl.StopCh) pattern is repeated. Same issue - the stop channel should not be closed by this test function.
| defer close(testCtrl.StopCh) |
| // FIX 1: Start informer factories HERE | ||
| sc.kubeInformerFactory.Start(stopCh) | ||
| sc.vcInformerFactory.Start(stopCh) | ||
|
|
||
| // FIX 2: Add specific sync checks with detailed logging |
There was a problem hiding this comment.
The comment "FIX 1: Start informer factories HERE" and "FIX 2: Add specific sync checks with detailed logging" appear to be development leftovers. These should be removed since the fixes are already implemented.
| // FIX 1: Start informer factories HERE | |
| sc.kubeInformerFactory.Start(stopCh) | |
| sc.vcInformerFactory.Start(stopCh) | |
| // FIX 2: Add specific sync checks with detailed logging | |
| sc.kubeInformerFactory.Start(stopCh) | |
| sc.vcInformerFactory.Start(stopCh) |
| } | ||
|
|
||
| testCtrl := NewTestShardingController(t, opt) | ||
| defer close(testCtrl.StopCh) |
There was a problem hiding this comment.
The defer close(testCtrl.StopCh) pattern is repeated across multiple test functions. This is incorrect - the stop channel is not owned by these test functions and closing a channel that may be shared or already closed can cause a panic. The cleanup should be handled by the test caller, not within these functions.
| defer close(testCtrl.StopCh) |
| const ( | ||
| NODE_SOURCE = "node-controller" | ||
| NODE_ADD_EVENT = "node-added" | ||
| NODE_UPDATE_EVENT = "node-updated" | ||
| NODE_DELETE_EVENT = "node-deleted" | ||
| ) |
There was a problem hiding this comment.
Constants should follow Go naming conventions. The constants NODE_SOURCE, NODE_ADD_EVENT, etc., should use camelCase or MixedCaps instead of snake_case. For example: NodeSource, NodeAddEvent, NodeUpdateEvent, NodeDeleteEvent.
|
|
||
| ## 3. Architecture Overview | ||
|
|
||
| ```mermaid |
There was a problem hiding this comment.
This is AI generated rubbish, this is not readable
|
|
||
| // Initialize initializes the controller | ||
| func (sc *ShardingController) Initialize(opt *framework.ControllerOption) error { | ||
| klog.V(6).Infof("Initializing ShardingController...") |
There was a problem hiding this comment.
| klog.V(6).Infof("Initializing ShardingController...") | |
| klog.V(2).Infof("Initializing ShardingController...") |
| func (sc *ShardingController) Initialize(opt *framework.ControllerOption) error { | ||
| klog.V(6).Infof("Initializing ShardingController...") | ||
| sc.ctx = context.Background() | ||
| sc.controllerOptions = *NewShardingControllerOptions() |
There was a problem hiding this comment.
nit: either change controllerOptions or NewShardingControllerOptions return value type.
| sc.initNodeIndices() | ||
|
|
||
| // Initialize queues | ||
| sc.queue = workqueue.NewTypedRateLimitingQueueWithConfig( |
There was a problem hiding this comment.
what is the queue for? NodeShard?
| sc.UpdateNodeMetrics(nodeName, metrics) | ||
|
|
||
| // Log significant changes | ||
| if prevMetrics := sc.GetNodeMetrics(nodeName); prevMetrics != nil { |
There was a problem hiding this comment.
How could you get previous metrics as you have updated it?
| klog.Infof("Cache synchronization completed successfully") | ||
|
|
||
| // Initialize node metrics | ||
| sc.initializeNodeMetrics() |
There was a problem hiding this comment.
is this same with refreshNodeMetrics, please keep one
| } | ||
|
|
||
| // processNodeEventWithRetry processes a node event with retry logic | ||
| func (sc *ShardingController) processNodeEventWithRetry(nodeName, eventType, source string, maxRetries int) error { |
There was a problem hiding this comment.
eventType and source are not even used, why do you make eventKey so complex
There was a problem hiding this comment.
this will be used for future threshold optimization including resource usage changes and updating frequency;
| func (sc *ShardingController) processNodeEventWithRetry(nodeName, eventType, source string, maxRetries int) error { | ||
| var lastErr error | ||
|
|
||
| for i := 0; i < maxRetries; i++ { |
There was a problem hiding this comment.
With this loop, you make a node retry at most 3*3 times
There was a problem hiding this comment.
remove the requeue logic;
| time.AfterFunc(200*time.Millisecond, func() { | ||
| sc.syncShards() |
There was a problem hiding this comment.
any consideration about the delay?
There was a problem hiding this comment.
Why do we have two places triggering syncShards: 1. node update, 2. periordically
Vaguely remember we donot want to update node shards very frequently. So I think we should have only one routine running sync, if we want node change to trigger sync, we should aggregate node events rather than each node trigger once
There was a problem hiding this comment.
yes, you are right; Here we need to aggregate node events to trigger shard update but not for all nodes; this will be fixed in the next PR;
| func (sc *ShardingController) ensureNodeStatesUpdated() { | ||
| // trigger re-computation if the time interval since the last update exceeds timeout threshold | ||
| sc.metricsMutex.RLock() | ||
| lastUpdateTime := time.Time{} | ||
| for _, nodeMetrics := range sc.nodeMetricsCache { | ||
| if nodeMetrics.LastUpdated.After(lastUpdateTime) { | ||
| lastUpdateTime = nodeMetrics.LastUpdated | ||
| } | ||
| } | ||
| sc.metricsMutex.RUnlock() | ||
|
|
||
| if time.Since(lastUpdateTime) > updateTimeoutThreshold { | ||
| klog.V(4).Infof("Node states stale, triggering update") | ||
| sc.updateAllNodeStates() |
There was a problem hiding this comment.
I donot think we need update all again. iirc, the node metrcs are updated periordically and also by node event.
There was a problem hiding this comment.
ensureNodeStatesUpdated is removed;
| ShardSyncPeriod: 60 * time.Second, | ||
| EnableNodeEventTrigger: true, | ||
| SchedulerConfigsRaw: []string{ | ||
| "volcano:volcano:0.0:0.6:false:2:100", |
There was a problem hiding this comment.
can we use structured json instead of a simple string
There was a problem hiding this comment.
will fixed in the future with the configmap PR;
Signed-off-by: ssfffss <senbof@gmail.com>
| } | ||
|
|
||
| // round the utilization with 2 floating points | ||
| cpuUtilRounded := math.Round(resourceInfo.CPUUtilization*100) / 100 |
There was a problem hiding this comment.
We only think about CPU currently?
There was a problem hiding this comment.
For the first version, we mainly verify the logic; It is tracked by issue #4879
There was a problem hiding this comment.
I agree with the Utilization-based Filtering, but should clarify in the doc that currently we only support CPU utilization and will support multi-dimensional resources in the future
Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
Signed-off-by: ssfffss <senbof@gmail.com>
|
/lgtm |
| if len(s.Controllers) > 1 { | ||
| return fmt.Errorf("wildcard '*' cannot be combined with other input") | ||
| if len(s.Controllers) > 1 && idx != len(s.Controllers)-1 { | ||
| return fmt.Errorf("wildcard '*' can only be placed at the final position when combined with other input") |
There was a problem hiding this comment.
I am not sure why cannot put * at last? Previously i can be only put at first, now you changed to be any position except last
There was a problem hiding this comment.
this is because when we check whether a controller is enabled or not through isControllerEnabled, its logic is when it finds "" mark, it returns true; it will ignore all the configurations after ''. Therefore, '*" can only be placed in the last position; This also aligns with our logic, we can put exceptions at first, and enable others; Easy to use in this way and logically clean;
| fs.Uint32Var(&s.WorkerThreadsForGC, "worker-threads-for-gc", defaultGCWorkers, "The number of threads for recycling jobs. The larger the number, the faster the job recycling, but requires more CPU load.") | ||
| fs.Uint32Var(&s.WorkerThreadsForQueue, "worker-threads-for-queue", defaultQueueWorkers, "The number of threads syncing queue operations. The larger the number, the faster the queue processing, but requires more CPU load.") | ||
| fs.StringSliceVar(&s.Controllers, "controllers", []string{defaultControllers}, fmt.Sprintf("Specify controller gates. Use '*' for all controllers, all knownController: %s ,and we can use "+ | ||
| fs.StringSliceVar(&s.Controllers, "controllers", strings.Split(defaultControllers, ","), fmt.Sprintf("Specify controller gates. Use '*' for all controllers, all knownController: %s ,and we can use "+ |
There was a problem hiding this comment.
suggest making *,-sharding-controller as the default
| changePercent := float64(abs(len(event.NewNodes)-len(event.OldNodes))) / float64(len(event.OldNodes)) | ||
| if changePercent > nodeCountChangeThreshold { | ||
| klog.Infof("Significant node change for %s: %.0f%% (%d -> %d nodes)", | ||
| event.SchedulerName, changePercent*100, len(event.OldNodes), len(event.NewNodes)) |
There was a problem hiding this comment.
Is this function just log, seems very heavy to start a go routine for logging assignment change? Why not directly log it at the place where sender resides
| go sc.assignmentChangeProcessor(stopCh) | ||
|
|
||
| // Start periodic metrics refresh | ||
| go wait.Until(sc.refreshNodeMetrics, nodeRefreshPeriod, stopCh) |
There was a problem hiding this comment.
IIUC, i think nodeRefreshPeriod should better <= shardSyncPeriod, but now shardSyncPeriod is far smaller, which means one syncShards after another maybe unnecessary because of no utils metrics changed during this time window
There was a problem hiding this comment.
fixed; but in the issues, we will remove it next time as we already use event to trigger node refresh, no need for periodic update;
|
|
||
| // Implement NodeMetricsProvider interface | ||
| func (sc *ShardingController) GetNodeMetrics(nodeName string) *NodeMetrics { | ||
| return sc.nodeMetricsCache[nodeName] |
There was a problem hiding this comment.
nit： why donot protect with lock like GetAllNodeMetrics
Same question on UpdateNodeMetrics
There was a problem hiding this comment.
now the lock is the for whole cache, lock operations would be too frequent if for each node. Here needs careful optimizations on fine-granularity locks as referred in issue #4878
| func (sm *ShardingManager) filterEligibleNodes( | ||
| config SchedulerConfig, | ||
| nodeResources map[string]*NodeResourceInfo, | ||
| nodeMap map[string]*corev1.Node, |
There was a problem hiding this comment.
nit: nodeResources should be kept only
Signed-off-by: ssfffss <senbof@gmail.com>
hzxuzhonghu
left a comment
hzxuzhonghu
left a comment
There was a problem hiding this comment.
/approve
it has been disabled by default
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: hzxuzhonghu The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
volcano-sh-bot
added
the
approved
|
/lgtm |
What type of PR is this?
feature: this PR implements sharding controllers for dynamic node partitioning between schedulers
What this PR does / why we need it:
This PR introduces a new Sharding Controller to Volcano that dynamically partitions cluster nodes between different scheduler types (particularly for Agentic AI workloads). The controller:
This enables Volcano to efficiently support both traditional batch workloads and emerging Agentic AI workloads on the same cluster, improving resource utilization and scheduling performance.
Which issue(s) this PR fixes:
Fixes #4722
Special notes for your reviewer:
This PR depends on the NodeShard CRD implementation in volcano-sh/apis
Does this PR introduce a user-facing change?