[WIP] Add workflow concurrency controller #6309
Conversation
|
Thank you for opening this pull request! 🙌 These tips will help get your PR across the finish line:
|
thomasjhuang
changed the title
Code Review Agent Run #5d27f8Actionable Suggestions - 7
Filtered by Review RulesBito filtered these suggestions based on rules created automatically for your feedback. Manage rules.
Review Details
AI Code Review powered by |
Changelist by BitoThis pull request implements the following key changes.
|
| if x != nil { | ||
| return x.Max | ||
| } | ||
| return 0 |
There was a problem hiding this comment.
The SchedulerPolicy struct is being added with a Max field for concurrency control, but there's no validation to ensure the Max value is greater than zero. Consider adding validation to prevent setting Max to zero, which would effectively block all executions.
Code suggestion
Check the AI-generated fix before applying
| return 0 | |
| return 1 // Default to 1 to allow at least one execution to run |
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| /** ExecutionStateChangeDetails description. */ | ||
| public description: string; |
There was a problem hiding this comment.
The added description property is defined as a non-optional string, but in the interface definition at line 14884, it's defined as an optional property (description?: (string|null)). Consider making the implementation consistent with the interface by changing public description: string; to public description?: string; or public description: string | null; to match the interface definition.
Code suggestion
Check the AI-generated fix before applying
| /** ExecutionStateChangeDetails description. */ | |
| public description: string; | |
| /** ExecutionStateChangeDetails description. */ | |
| public description: string | null; |
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| /** ExecutionStateChangeDetails description. */ | ||
| public description: string; |
There was a problem hiding this comment.
The PR adds a new description field to the ExecutionStateChangeDetails interface. This looks good, but I noticed that the implementation in the class (line 14905-14906) declares it as a required field (public description: string), while the interface defines it as optional (description?: (string|null)). Consider making these consistent to avoid potential type errors.
Code suggestion
Check the AI-generated fix before applying
| /** ExecutionStateChangeDetails description. */ | |
| public description: string; | |
| /** ExecutionStateChangeDetails description. */ | |
| public description?: string; |
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| lpKey := getLaunchPlanKey(lp.Project, lp.Domain, lp.Name) | ||
| executionMap := make(map[string]bool) | ||
|
|
||
| for _, exec := range executions { | ||
| executionKey := getExecutionKey(exec.Project, exec.Domain, exec.Name) |
There was a problem hiding this comment.
The code calls getLaunchPlanKey and getExecutionKey functions, and splitExecutionKey on line 267, but these functions aren't defined in the provided code. We need to ensure these utility functions exist and work correctly.
Code suggestion
Check the AI-generated fix before applying
@@ -369,0 +370,25 @@
+
+// getLaunchPlanKey returns a string key for a launch plan
+func getLaunchPlanKey(project, domain, name string) string {
+ return fmt.Sprintf("%s/%s/%s", project, domain, name)
+}
+
+// getExecutionKey returns a string key for an execution
+func getExecutionKey(project, domain, name string) string {
+ return fmt.Sprintf("%s/%s/%s", project, domain, name)
+}
+
+// splitExecutionKey splits an execution key into project, domain, and name
+func splitExecutionKey(key string) []string {
+ return strings.Split(key, "/")
+}
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| // Get current running executions count | ||
| runningCount, err := c.GetExecutionCounts(ctx, launchPlanID) |
There was a problem hiding this comment.
The code calls c.GetExecutionCounts(ctx, launchPlanID) to get the current running executions count, but this function isn't defined in the provided code. We need to ensure this function exists and returns the correct count of running executions.
Code suggestion
Check the AI-generated fix before applying
@@ -369,0 +370,19 @@
+
+// GetExecutionCounts returns the count of running executions for a launch plan
+func (c *ConcurrencyController) GetExecutionCounts(ctx context.Context, launchPlanID idlCore.Identifier) (int, error) {
+ lpKey := getLaunchPlanKey(launchPlanID.Project, launchPlanID.Domain, launchPlanID.Name)
+
+ c.runningExecutionsMutex.RLock()
+ defer c.runningExecutionsMutex.RUnlock()
+
+ executions, ok := c.runningExecutions[lpKey]
+ if !ok {
+ return 0, nil
+ }
+
+ return len(executions), nil
+}
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| c.metrics.executionsAbortedCount.Inc() | ||
|
|
||
| // Untrack the aborted execution | ||
| c.UntrackExecution(ctx, oldestExecID) |
There was a problem hiding this comment.
The code calls c.UntrackExecution(ctx, oldestExecID) to untrack an aborted execution, but this function isn't defined in the provided code. We need to ensure this function exists and correctly removes the execution from the tracking map.
Code suggestion
Check the AI-generated fix before applying
@@ -369,0 +370,19 @@
+
+// UntrackExecution removes an execution from the tracking map
+func (c *ConcurrencyController) UntrackExecution(ctx context.Context, executionID idlCore.WorkflowExecutionIdentifier) {
+ executionKey := getExecutionKey(executionID.Project, executionID.Domain, executionID.Name)
+
+ c.runningExecutionsMutex.Lock()
+ defer c.runningExecutionsMutex.Unlock()
+
+ // Find the launch plan for this execution
+ for lpKey, executions := range c.runningExecutions {
+ if _, ok := executions[executionKey]; ok {
+ // Remove the execution from the map
+ delete(c.runningExecutions[lpKey], executionKey)
+
+ // If no more executions for this launch plan, remove the launch plan entry
+ if len(c.runningExecutions[lpKey]) == 0 {
+ delete(c.runningExecutions, lpKey)
+ }
+ break
+ }
+ }
+}
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
| func (c *ConcurrencyController) RecordLatency(operation string, startTime time.Time) { | ||
| switch operation { | ||
| case "processing": | ||
| c.metrics.processingLatency.Observe(startTime) |
There was a problem hiding this comment.
The RecordLatency method is using startTime directly with Observe(), but typically latency metrics require calculating the duration between the start time and now. Consider checking if the Observe method expects a duration rather than a timestamp.
Code suggestion
Check the AI-generated fix before applying
| c.metrics.processingLatency.Observe(startTime) | |
| c.metrics.processingLatency.Observe(time.Since(startTime).Seconds()) |
Code Review Run #5d27f8
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
DRAFT
This work is currently in progress, some portion of this work is incomplete and still being tested.
Tracking issue
Flyte backend implementation of #5659.
Why are the changes needed?
This implements a flyte concurrency manager, which allows for granular control of concurrent workflows at a LaunchPlan version level.
What changes were proposed in this pull request?
How was this patch tested?
Labels
Please add one or more of the following labels to categorize your PR:
This is important to improve the readability of release notes.
Setup process
Screenshots
Check all the applicable boxes
Related PRs
Docs link