feat: inject NodeAffinity into DataBackup status after successful backup#6110
feat: inject NodeAffinity into DataBackup status after successful backup#6110adityaupasani2 wants to merge 1 commit into
Conversation
…ckup DataLoad, DataMigrate, and DataProcess all populate status.NodeAffinity after a successful operation so that downstream runAfter operations are scheduled on nodes where data is already cached. DataBackup was missing this because it uses a Pod (not a Job), and the existing dataflow.GenerateNodeAffinity() only accepts *batchv1.Job. Add dataflow.GenerateNodeAffinityFromPod(*corev1.Pod) in pkg/dataflow/helper.go using the same annotation-based logic as GenerateNodeAffinity, and call it in DataBackup OnceHandler after the backup pod succeeds. Errors (e.g. no affinity annotations) are logged at V(1) and suppressed so the operation still completes successfully. Also adds: - TestGenerateNodeAffinityFromPod in pkg/dataflow/helper_test.go covering nil pod, missing inject annotation, inject with no labels, and inject with labels - Two new Ginkgo cases in status_handler_test.go asserting NodeAffinity is injected (with annotations) and not injected (without annotations) Fixes fluid-cloudnative#6107 Signed-off-by: Aditya Upasani <adityaupasani29@gmail.com>
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Hi @adityaupasani2. Thanks for your PR. I'm waiting for a fluid-cloudnative member to verify that this patch is reasonable to test. If it is, they should reply with Once the patch is verified, the new status will be reflected by the I understand the commands that are listed here. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository. |
|
![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 the ability to inject NodeAffinity into DataBackup status from a successful backup pod's annotations, implementing GenerateNodeAffinityFromPod along with corresponding unit tests. The review feedback suggests avoiding the reuse of the named return parameter err for transient errors to prevent side effects, and extracting the shared annotation-parsing logic into a helper function to eliminate code duplication between pod and job affinity generation.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| if kubeclient.IsSucceededPod(backupPod) && result.NodeAffinity == nil { | ||
| result.NodeAffinity, err = dataflow.GenerateNodeAffinityFromPod(backupPod) | ||
| if err != nil { | ||
| ctx.Log.V(1).Info("NodeAffinity not injected", "reason", err.Error()) | ||
| err = nil | ||
| } | ||
| } |
There was a problem hiding this comment.
Reusing the named return parameter err for a transient error inside a nested block can be error-prone and may lead to accidental side effects if the function is refactored in the future. It is safer and more idiomatic in Go to use a local variable (e.g., affinityErr) to handle this transient error.
| if kubeclient.IsSucceededPod(backupPod) && result.NodeAffinity == nil { | |
| result.NodeAffinity, err = dataflow.GenerateNodeAffinityFromPod(backupPod) | |
| if err != nil { | |
| ctx.Log.V(1).Info("NodeAffinity not injected", "reason", err.Error()) | |
| err = nil | |
| } | |
| } | |
| if kubeclient.IsSucceededPod(backupPod) && result.NodeAffinity == nil { | |
| nodeAffinity, affinityErr := dataflow.GenerateNodeAffinityFromPod(backupPod) | |
| if affinityErr != nil { | |
| ctx.Log.V(1).Info("NodeAffinity not injected", "reason", affinityErr.Error()) | |
| } else { | |
| result.NodeAffinity = nodeAffinity | |
| } | |
| } |
| // GenerateNodeAffinityFromPod generates a NodeAffinity from a Pod's annotations, | ||
| // using the same annotation-based logic as GenerateNodeAffinity for Jobs. | ||
| // This is used by DataBackup which runs as a Pod rather than a Job. | ||
| func GenerateNodeAffinityFromPod(pod *corev1.Pod) (*corev1.NodeAffinity, error) { | ||
| if pod == nil { | ||
| return nil, nil | ||
| } | ||
| // not inject, i.e. feature gate not enabled | ||
| if v := pod.Annotations[common.AnnotationDataFlowAffinityInject]; v != "true" { | ||
| return nil, nil | ||
| } | ||
|
|
||
| annotations := pod.Annotations | ||
|
|
||
| nodeAffinity := &corev1.NodeAffinity{ | ||
| RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{ | ||
| NodeSelectorTerms: []corev1.NodeSelectorTerm{ | ||
| { | ||
| MatchExpressions: nil, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| hasInjectedLabels := false | ||
| for key, value := range annotations { | ||
| if strings.HasPrefix(key, common.AnnotationDataFlowCustomizedAffinityPrefix) { | ||
| nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions = | ||
| append(nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions, | ||
| corev1.NodeSelectorRequirement{ | ||
| Key: strings.TrimPrefix(key, common.AnnotationDataFlowCustomizedAffinityPrefix), | ||
| Operator: corev1.NodeSelectorOpIn, | ||
| Values: []string{value}, | ||
| }) | ||
| hasInjectedLabels = true | ||
| } | ||
| } | ||
| if !hasInjectedLabels { | ||
| return nil, errors.New("the affinity label is not set, wait for next reconcile") | ||
| } | ||
|
|
||
| return nodeAffinity, nil | ||
| } |
There was a problem hiding this comment.
The logic in GenerateNodeAffinityFromPod is almost identical to GenerateNodeAffinity for Jobs, resulting in significant code duplication. We can extract the common annotation-parsing logic into a helper function GenerateNodeAffinityFromAnnotations(annotations map[string]string) and reuse it in both functions. This improves maintainability and readability.
// GenerateNodeAffinityFromAnnotations generates a NodeAffinity from annotations.
func GenerateNodeAffinityFromAnnotations(annotations map[string]string) (*corev1.NodeAffinity, error) {
if v := annotations[common.AnnotationDataFlowAffinityInject]; v != "true" {
return nil, nil
}
nodeAffinity := &corev1.NodeAffinity{
RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
NodeSelectorTerms: []corev1.NodeSelectorTerm{
{
MatchExpressions: nil,
},
},
},
}
hasInjectedLabels := false
for key, value := range annotations {
if strings.HasPrefix(key, common.AnnotationDataFlowCustomizedAffinityPrefix) {
nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions =
append(nodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms[0].MatchExpressions,
corev1.NodeSelectorRequirement{
Key: strings.TrimPrefix(key, common.AnnotationDataFlowCustomizedAffinityPrefix),
Operator: corev1.NodeSelectorOpIn,
Values: []string{value},
})
hasInjectedLabels = true
}
}
if !hasInjectedLabels {
return nil, errors.New("the affinity label is not set, wait for next reconcile")
}
return nodeAffinity, nil
}
// GenerateNodeAffinityFromPod generates a NodeAffinity from a Pod's annotations,
// using the same annotation-based logic as GenerateNodeAffinity for Jobs.
// This is used by DataBackup which runs as a Pod rather than a Job.
func GenerateNodeAffinityFromPod(pod *corev1.Pod) (*corev1.NodeAffinity, error) {
if pod == nil {
return nil, nil
}
return GenerateNodeAffinityFromAnnotations(pod.Annotations)
}
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #6110 +/- ##
==========================================
+ Coverage 64.76% 64.80% +0.03%
==========================================
Files 485 485
Lines 33895 33930 +35
==========================================
+ Hits 21952 21987 +35
Misses 10220 10220
Partials 1723 1723 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
/copilot review |
|
Thanks for picking up the old TODO here. The mechanics of building the NodeAffinity look right and mirror the DataLoad/DataMigrate handlers, but I have one structural concern before this can land: I can't find anything that puts the |
|
|
||
| // TODO: inject nodeaffinity like other data operations when using job instead of pod | ||
| if kubeclient.IsSucceededPod(backupPod) && result.NodeAffinity == nil { | ||
| result.NodeAffinity, err = dataflow.GenerateNodeAffinityFromPod(backupPod) |
There was a problem hiding this comment.
I traced where the annotations this reads actually come from, and I don't think this branch ever fires in a real cluster. GenerateNodeAffinityFromPod returns early unless the pod carries dataflow-affinity.fluid.io.inject=true plus the affinity.dataflow.fluid.io.* node-label annotations. Those annotations are only ever written by DataOpJobReconciler in pkg/controllers/v1alpha1/fluidapp/dataflowaffinity/, whose ManagedResource() returns &batchv1.Job{} and which writes the labels onto the Job (injectPodNodeLabelsToJob). DataBackup runs as a Pod, not a Job — the pod template in charts/fluid-databackup/*/templates/databackup.yaml sets no such annotations, and there's no pod-level reconciler that backfills them. So for a real DataBackup, pod.Annotations[AnnotationDataFlowAffinityInject] != "true" and this returns (nil, nil); result.NodeAffinity stays nil. The new unit tests only pass because they hand-set those annotations on a fake pod. Where is the backup pod supposed to get the inject/affinity annotations? If there isn't a source yet, this needs the pod-side injection (equivalent of the Job reconciler) to actually deliver the feature.
| } | ||
| } | ||
|
|
||
| func TestGenerateNodeAffinityFromPod(t *testing.T) { |
There was a problem hiding this comment.
This is a feat PR that adds a new user-visible behavior (populating DataBackup.Status.NodeAffinity so dataflow runAfter chains can schedule against it), but it only ships unit tests. The repo's e2e suite lives under test/gha-e2e/ (driven by .github/workflows/kind-e2e.yml), and there's no databackup/affinity scenario there. Please add an e2e case that runs a DataBackup and a dependent dataflow operation and asserts the affinity actually propagates end to end — that would also directly confirm or refute the annotation-source concern above, since a real run is exactly where the no-op would show up. If you believe an e2e case isn't feasible for this change, let's get a maintainer's call on that rather than skipping it.
| result.NodeAffinity, err = dataflow.GenerateNodeAffinityFromPod(backupPod) | ||
| if err != nil { | ||
| ctx.Log.V(1).Info("NodeAffinity not injected", "reason", err.Error()) | ||
| err = nil |
There was a problem hiding this comment.
The the affinity label is not set, wait for next reconcile error is logged at V(1) and then cleared (err = nil), after which this same pass goes on to set Phase = Complete. Once the status is terminal there's no follow-up reconcile to backfill the affinity, so the "wait for next reconcile" wording doesn't match what happens for a backup. Separately, reusing the named return err inside this nested block is a bit fragile — a local variable (e.g. affinityErr) makes the intent clearer and avoids accidental leakage if this function grows later.
| // GenerateNodeAffinityFromPod generates a NodeAffinity from a Pod's annotations, | ||
| // using the same annotation-based logic as GenerateNodeAffinity for Jobs. | ||
| // This is used by DataBackup which runs as a Pod rather than a Job. | ||
| func GenerateNodeAffinityFromPod(pod *corev1.Pod) (*corev1.NodeAffinity, error) { |
There was a problem hiding this comment.
GenerateNodeAffinityFromPod is a near-verbatim copy of GenerateNodeAffinity — both only read .Annotations. Consider extracting the shared body into something like generateNodeAffinityFromAnnotations(annotations map[string]string) (*corev1.NodeAffinity, error) and having both the Job and Pod entrypoints delegate to it, so the two paths can't drift apart.
|
@cheyang Thanks for tracing this all the way through you're right, and it's actually a bit worse than annotation-timing: even a pod-side reconciler mirroring DataOpJobReconciler would race against this handler's own Phase = Complete transition, since there's no follow-up reconcile once the backup is terminal. Rather than replicate the Job path (separate reconciler writing dataflow-affinity.fluid.io/* annotations onto the pod, then reading them back), I'd like to compute the NodeAffinity directly in OnceHandler.GetOperationStatus() once we have the succeeded pod: read pod.Spec.NodeName, fetch the Node, and build result.NodeAffinity from its labels the same logic injectPodNodeLabelsToJob uses, just inlined instead of round-tripped through annotations. That removes the dependency on a second controller ever running, which seems like the right call for a Pod-backed operation that transitions to Complete in a single pass. Before I rewrite this, does that direction make sense to you, or would you rather see a pod-side reconciler added to dataflowaffinity for consistency with the Job path (accepting the ordering risk, or with some safeguard against it)? Want to make sure I'm not solving this a third way before you've had a chance to weigh in. Will also fold in the affinityErr local-var fix and extract the shared annotation/label-parsing helper regardless of which direction we land on, and add an e2e case under test/gha-e2e/ once the mechanism is settled. |
2 participants
Ⅰ. Describe what this PR does
DataLoad,DataMigrate, andDataProcessall populatestatus.nodeAffinityafter a successful operation usingdataflow.GenerateNodeAffinity(job). This allows downstreamrunAfteroperations to be scheduled on nodes where data is already cached, improving cache hit rates.DataBackupwas missing this because it uses a Pod instead of a Job, anddataflow.GenerateNodeAffinity()only accepts*batchv1.Job. As a result,DataBackup.Status.NodeAffinitywas never set — breaking the dataflow affinity chain for any operation that depends on a DataBackup viarunAfter.This PR:
dataflow.GenerateNodeAffinityFromPod(*corev1.Pod)inpkg/dataflow/helper.gousing the same annotation-based logic as the existingGenerateNodeAffinityDataBackup.OnceHandler.GetOperationStatus()after the backup pod succeeds, same as other handlers do for jobsV(1)and suppressed so the operation still completes successfullyⅡ. Does this pull request fix one issue?
Fixes #6107
Ⅲ. List the added test cases
TestGenerateNodeAffinityFromPodinpkg/dataflow/helper_test.gocovering: nil pod, missing inject annotation, inject annotation with no affinity labels (error), inject annotation with labels (success)status_handler_test.go: NodeAffinity injected when pod has dataflow annotations, NodeAffinity not injected when pod has no annotationsⅣ. Describe how to verify it