Fix: Don't fail suspended Kubeflow jobs #6295
Open
Conversation
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
Code Review Agent Run #ae6c12Actionable Suggestions - 0Review Details
AI Code Review powered by |
Changelist by BitoThis pull request implements the following key changes.
|
Codecov ReportAll modified and coverable lines are covered by tests ✅
Additional details and impacted files@@ Coverage Diff @@
## fg91/dep/upgrade-kubeflow-training-operator #6295 +/- ##
===============================================================================
+ Coverage 58.48% 58.50% +0.02%
===============================================================================
Files 937 937
Lines 71088 71091 +3
===============================================================================
+ Hits 41577 41594 +17
+ Misses 26359 26348 -11
+ Partials 3152 3149 -3
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
fg91
force-pushed
the
fg91/dep/upgrade-kubeflow-training-operator
branch
from
72d0107 to
5effd97
Compare
3 tasks
Signed-off-by: Fabio Graetz <[email protected]>
fg91
force-pushed
the
fg91/fix/dont-fail-suspended-kubeflow-jobs
branch
from
9559d64 to
ce5a5d8
Compare
Code Review Agent Run #58db2dActionable Suggestions - 1
Review Details
AI Code Review powered by |
flyte-bot
reviewed
Mar 7, 2025
Comment on lines
+230
to
+231
| isSuspended := app.Spec.RunPolicy.Suspend != nil && *app.Spec.RunPolicy.Suspend | ||
| if !isSuspended && app.Status.StartTime == nil && app.CreationTimestamp.Add(common.GetConfig().Timeout.Duration).Before(time.Now()) { |
There was a problem hiding this comment.
Potential nil pointer dereference check
Consider checking if app.Spec.RunPolicy is nil before accessing app.Spec.RunPolicy.Suspend. This would prevent a potential nil pointer dereference if RunPolicy is not initialized.
Code suggestion
Check the AI-generated fix before applying
Suggested change
| isSuspended := app.Spec.RunPolicy.Suspend != nil && *app.Spec.RunPolicy.Suspend | |
| if !isSuspended && app.Status.StartTime == nil && app.CreationTimestamp.Add(common.GetConfig().Timeout.Duration).Before(time.Now()) { | |
| isSuspended := false | |
| if app.Spec.RunPolicy != nil && app.Spec.RunPolicy.Suspend != nil && *app.Spec.RunPolicy.Suspend { | |
| isSuspended = true | |
| } | |
| if !isSuspended && app.Status.StartTime == nil && app.CreationTimestamp.Add(common.GetConfig().Timeout.Duration).Before(time.Now()) { |
Code Review Run #58db2d
Should Bito avoid suggestions like this for future reviews? (Manage Rules)
- Yes, avoid them
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Why are the changes needed?
The flyteplugins kubeflow plugins currently fail tasks if the underlying CRD object (PyTorchJob, TfJob, MpiJob) haven't been updated by the training operator for a certain timeout period (see here):
However, the jobs provided by the kubeflow training operator can be in a so-called "suspended" state (e.g. when using them with an external queueing system like Kueue). Jobs in a suspended state are expected to not be updated by the training operator.
What changes were proposed in this pull request?
I propose to add a check for suspension to the error condition in the code snippet above so that the flyteplugin ignores unmodified kubeflow jobs if they are in a suspended state.
Please note that other flyteplugins like spark or ray don't have this "CRD has been updated" check at all so this change does not introduce inconsistency.
How was this patch tested?
Added unit tests. Ran flytepropeller image with this change in cluster.
Check all the applicable boxes
Related PRs
Summary by Bito
This PR updates Kubeflow plugins (MPI, PyTorch, TensorFlow) to properly handle suspended jobs by adding suspension checks to timeout failure conditions. It prevents premature task failures by verifying job suspension status before raising errors, ensuring suspended jobs are treated appropriately without affecting other plugins.Unit tests added: False
Estimated effort to review (1-5, lower is better): 1