mirror of https://github.com/k3s-io/k3s
Add preconditions to pod/rs update retry func
parent
2a4479fa60
commit
aae2073a78
|
@ -187,27 +187,30 @@ func addHashKeyToRSAndPods(deployment *extensions.Deployment, c clientset.Interf
|
|||
}))
|
||||
rsUpdated := false
|
||||
// 1. Add hash template label to the rs. This ensures that any newly created pods will have the new label.
|
||||
if len(updatedRS.Spec.Template.Labels[extensions.DefaultDeploymentUniqueLabelKey]) == 0 {
|
||||
updatedRS, rsUpdated, err = rsutil.UpdateRSWithRetries(c.Extensions().ReplicaSets(namespace), updatedRS, func(updated *extensions.ReplicaSet) {
|
||||
updatedRS, rsUpdated, err = rsutil.UpdateRSWithRetries(c.Extensions().ReplicaSets(namespace), updatedRS,
|
||||
func(updated *extensions.ReplicaSet) bool {
|
||||
return updated.Spec.Template.Labels[extensions.DefaultDeploymentUniqueLabelKey] != hash
|
||||
},
|
||||
func(updated *extensions.ReplicaSet) {
|
||||
updated.Spec.Template.Labels = labelsutil.AddLabel(updated.Spec.Template.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error updating %s %s/%s pod template label with template hash: %v", updatedRS.Kind, updatedRS.Namespace, updatedRS.Name, err)
|
||||
}
|
||||
if rsUpdated {
|
||||
// Make sure rs pod template is updated so that it won't create pods without the new label (orphaned pods).
|
||||
if updatedRS.Generation > updatedRS.Status.ObservedGeneration {
|
||||
if err = waitForReplicaSetUpdated(c, updatedRS.Generation, namespace, updatedRS.Name); err != nil {
|
||||
return nil, fmt.Errorf("error waiting for %s %s/%s generation %d observed by controller: %v", updatedRS.Kind, updatedRS.Namespace, updatedRS.Name, updatedRS.Generation, err)
|
||||
}
|
||||
}
|
||||
glog.V(4).Infof("Observed the update of %s %s/%s's pod template with hash %s.", rs.Kind, rs.Namespace, rs.Name, hash)
|
||||
} else {
|
||||
// If RS wasn't updated but didn't return error in step 1, we've hit a RS not found error.
|
||||
// Return here and retry in the next sync loop.
|
||||
return &rs, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error updating %s %s/%s pod template label with template hash: %v", updatedRS.Kind, updatedRS.Namespace, updatedRS.Name, err)
|
||||
}
|
||||
if rsUpdated {
|
||||
// Make sure rs pod template is updated so that it won't create pods without the new label (orphaned pods).
|
||||
if updatedRS.Generation > updatedRS.Status.ObservedGeneration {
|
||||
if err = waitForReplicaSetUpdated(c, updatedRS.Generation, namespace, updatedRS.Name); err != nil {
|
||||
return nil, fmt.Errorf("error waiting for %s %s/%s generation %d observed by controller: %v", updatedRS.Kind, updatedRS.Namespace, updatedRS.Name, updatedRS.Generation, err)
|
||||
}
|
||||
}
|
||||
glog.V(4).Infof("Observed the update of %s %s/%s's pod template with hash %s.", rs.Kind, rs.Namespace, rs.Name, hash)
|
||||
} else {
|
||||
// If RS wasn't updated but didn't return error in step 1, we've hit a RS not found error.
|
||||
// Return here and retry in the next sync loop.
|
||||
return &rs, nil
|
||||
}
|
||||
glog.V(4).Infof("Observed the update of rs %s's pod template with hash %s.", rs.Name, hash)
|
||||
|
||||
// 2. Update all pods managed by the rs to have the new hash label, so they will be correctly adopted.
|
||||
selector, err := unversioned.LabelSelectorAsSelector(updatedRS.Spec.Selector)
|
||||
|
@ -231,10 +234,14 @@ func addHashKeyToRSAndPods(deployment *extensions.Deployment, c clientset.Interf
|
|||
|
||||
// 3. Update rs label and selector to include the new hash label
|
||||
// Copy the old selector, so that we can scrub out any orphaned pods
|
||||
if updatedRS, rsUpdated, err = rsutil.UpdateRSWithRetries(c.Extensions().ReplicaSets(namespace), updatedRS, func(updated *extensions.ReplicaSet) {
|
||||
updated.Labels = labelsutil.AddLabel(updated.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
updated.Spec.Selector = labelsutil.AddLabelToSelector(updated.Spec.Selector, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
}); err != nil {
|
||||
if updatedRS, rsUpdated, err = rsutil.UpdateRSWithRetries(c.Extensions().ReplicaSets(namespace), updatedRS,
|
||||
func(updated *extensions.ReplicaSet) bool {
|
||||
return updated.Labels[extensions.DefaultDeploymentUniqueLabelKey] != hash || updated.Spec.Selector.MatchLabels[extensions.DefaultDeploymentUniqueLabelKey] != hash
|
||||
},
|
||||
func(updated *extensions.ReplicaSet) {
|
||||
updated.Labels = labelsutil.AddLabel(updated.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
updated.Spec.Selector = labelsutil.AddLabelToSelector(updated.Spec.Selector, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
}); err != nil {
|
||||
return nil, fmt.Errorf("error updating %s %s/%s label and selector with template hash: %v", updatedRS.Kind, updatedRS.Namespace, updatedRS.Name, err)
|
||||
}
|
||||
if rsUpdated {
|
||||
|
@ -264,9 +271,13 @@ func labelPodsWithHash(podList *api.PodList, rs *extensions.ReplicaSet, c client
|
|||
for _, pod := range podList.Items {
|
||||
// Only label the pod that doesn't already have the new hash
|
||||
if pod.Labels[extensions.DefaultDeploymentUniqueLabelKey] != hash {
|
||||
if _, podUpdated, err := podutil.UpdatePodWithRetries(c.Core().Pods(namespace), &pod, func(podToUpdate *api.Pod) {
|
||||
podToUpdate.Labels = labelsutil.AddLabel(podToUpdate.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
}); err != nil {
|
||||
if _, podUpdated, err := podutil.UpdatePodWithRetries(c.Core().Pods(namespace), &pod,
|
||||
func(podToUpdate *api.Pod) bool {
|
||||
return podToUpdate.Labels[extensions.DefaultDeploymentUniqueLabelKey] != hash
|
||||
},
|
||||
func(podToUpdate *api.Pod) {
|
||||
podToUpdate.Labels = labelsutil.AddLabel(podToUpdate.Labels, extensions.DefaultDeploymentUniqueLabelKey, hash)
|
||||
}); err != nil {
|
||||
return false, fmt.Errorf("error in adding template hash label %s to pod %+v: %s", hash, pod, err)
|
||||
} else if podUpdated {
|
||||
glog.V(4).Infof("Labeled %s %s/%s of %s %s/%s with hash %s.", pod.Kind, pod.Namespace, pod.Name, rs.Kind, rs.Namespace, rs.Name, hash)
|
||||
|
|
|
@ -39,10 +39,11 @@ func GetPodTemplateSpecHash(template api.PodTemplateSpec) uint32 {
|
|||
// TODO: use client library instead when it starts to support update retries
|
||||
// see https://github.com/kubernetes/kubernetes/issues/21479
|
||||
type updatePodFunc func(pod *api.Pod)
|
||||
type preconditionFunc func(pod *api.Pod) bool
|
||||
|
||||
// UpdatePodWithRetries updates a pod with given applyUpdate function. Note that pod not found error is ignored.
|
||||
// UpdatePodWithRetries updates a pod with given applyUpdate function, when the given precondition holds. Note that pod not found error is ignored.
|
||||
// The returned bool value can be used to tell if the pod is actually updated.
|
||||
func UpdatePodWithRetries(podClient unversionedcore.PodInterface, pod *api.Pod, applyUpdate updatePodFunc) (*api.Pod, bool, error) {
|
||||
func UpdatePodWithRetries(podClient unversionedcore.PodInterface, pod *api.Pod, preconditionHold preconditionFunc, applyUpdate updatePodFunc) (*api.Pod, bool, error) {
|
||||
var err error
|
||||
var podUpdated bool
|
||||
oldPod := pod
|
||||
|
@ -51,8 +52,11 @@ func UpdatePodWithRetries(podClient unversionedcore.PodInterface, pod *api.Pod,
|
|||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if !preconditionHold(pod) {
|
||||
glog.V(4).Infof("pod %s precondition doesn't hold, skip updating it.", pod.Name)
|
||||
return true, nil
|
||||
}
|
||||
// Apply the update, then attempt to push it to the apiserver.
|
||||
// TODO: add precondition for update
|
||||
applyUpdate(pod)
|
||||
if pod, err = podClient.Update(pod); err == nil {
|
||||
// Update successful.
|
||||
|
|
|
@ -30,10 +30,11 @@ import (
|
|||
// TODO: use client library instead when it starts to support update retries
|
||||
// see https://github.com/kubernetes/kubernetes/issues/21479
|
||||
type updateRSFunc func(rs *extensions.ReplicaSet)
|
||||
type preconditionFunc func(rs *extensions.ReplicaSet) bool
|
||||
|
||||
// UpdateRSWithRetries updates a RS with given applyUpdate function. Note that RS not found error is ignored.
|
||||
// UpdateRSWithRetries updates a RS with given applyUpdate function, when the given precondition holds. Note that RS not found error is ignored.
|
||||
// The returned bool value can be used to tell if the RS is actually updated.
|
||||
func UpdateRSWithRetries(rsClient unversionedextensions.ReplicaSetInterface, rs *extensions.ReplicaSet, applyUpdate updateRSFunc) (*extensions.ReplicaSet, bool, error) {
|
||||
func UpdateRSWithRetries(rsClient unversionedextensions.ReplicaSetInterface, rs *extensions.ReplicaSet, preconditionHold preconditionFunc, applyUpdate updateRSFunc) (*extensions.ReplicaSet, bool, error) {
|
||||
var err error
|
||||
var rsUpdated bool
|
||||
oldRs := rs
|
||||
|
@ -42,8 +43,11 @@ func UpdateRSWithRetries(rsClient unversionedextensions.ReplicaSetInterface, rs
|
|||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if !preconditionHold(rs) {
|
||||
glog.V(4).Infof("rs %s precondition doesn't hold, skip updating it.", rs.Name)
|
||||
return true, nil
|
||||
}
|
||||
// Apply the update, then attempt to push it to the apiserver.
|
||||
// TODO: add precondition for update
|
||||
applyUpdate(rs)
|
||||
if rs, err = rsClient.Update(rs); err == nil {
|
||||
// Update successful.
|
||||
|
|
Loading…
Reference in New Issue