Skip to content

Commit

Permalink
Merge pull request #947 from rishabh-11/remove-shadow
Browse files Browse the repository at this point in the history 
Fix context declaration for vol detachment and reattachment
  • Loading branch information
@unmarshall
unmarshall authored Oct 21, 2024
2 parents 109a923 + 3a07275 commit e91182a
Show file tree
Hide file tree
Showing 5 changed files with 39 additions and 122 deletions.
17 changes: 2 additions & 15 deletions pkg/util/provider/app/app.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,6 @@ import (
machineinformers "github.com/gardener/machine-controller-manager/pkg/client/informers/externalversions"
coreclientbuilder "github.com/gardener/machine-controller-manager/pkg/util/clientbuilder/core"
machineclientbuilder "github.com/gardener/machine-controller-manager/pkg/util/clientbuilder/machine"
"github.com/gardener/machine-controller-manager/pkg/util/k8sutils"
machinecontroller "github.com/gardener/machine-controller-manager/pkg/util/provider/machinecontroller"
coreinformers "k8s.io/client-go/informers"
kubescheme "k8s.io/client-go/kubernetes/scheme"
Expand All @@ -52,8 +51,6 @@ import (
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/discovery"
policyv1informers "k8s.io/client-go/informers/policy/v1"
policyv1beta1informers "k8s.io/client-go/informers/policy/v1beta1"
"k8s.io/client-go/kubernetes"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
"k8s.io/client-go/pkg/version"
Expand Down Expand Up @@ -262,16 +259,7 @@ func StartControllers(s *options.MCServer,
s.MinResyncPeriod.Duration,
)

var (
pdbV1Informer policyv1informers.PodDisruptionBudgetInformer
pdbV1beta1Informer policyv1beta1informers.PodDisruptionBudgetInformer
)

if k8sutils.ConstraintK8sGreaterEqual121.Check(targetKubernetesVersion) {
pdbV1Informer = targetCoreInformerFactory.Policy().V1().PodDisruptionBudgets()
} else {
pdbV1beta1Informer = targetCoreInformerFactory.Policy().V1beta1().PodDisruptionBudgets()
}
pdbInformer := targetCoreInformerFactory.Policy().V1().PodDisruptionBudgets()

// All shared informers are v1alpha1 API level
machineSharedInformers := controlMachineInformerFactory.Machine().V1alpha1()
Expand All @@ -287,8 +275,7 @@ func StartControllers(s *options.MCServer,
targetCoreInformerFactory.Core().V1().PersistentVolumes(),
controlCoreInformerFactory.Core().V1().Secrets(),
targetCoreInformerFactory.Core().V1().Nodes(),
pdbV1beta1Informer,
pdbV1Informer,
pdbInformer,
targetCoreInformerFactory.Storage().V1().VolumeAttachments(),
machineSharedInformers.MachineClasses(),
machineSharedInformers.Machines(),
Expand Down
114 changes: 28 additions & 86 deletions pkg/util/provider/drain/drain.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,10 +45,8 @@ import (
"k8s.io/client-go/kubernetes"
corelisters "k8s.io/client-go/listers/core/v1"
policyv1listers "k8s.io/client-go/listers/policy/v1"
policyv1beta1listers "k8s.io/client-go/listers/policy/v1beta1"
"k8s.io/klog/v2"

"github.com/gardener/machine-controller-manager/pkg/util/k8sutils"
"github.com/gardener/machine-controller-manager/pkg/util/provider/driver"
)

Expand All @@ -72,8 +70,7 @@ type Options struct {
Out io.Writer
pvcLister corelisters.PersistentVolumeClaimLister
pvLister corelisters.PersistentVolumeLister
pdbV1beta1Lister policyv1beta1listers.PodDisruptionBudgetLister
pdbV1Lister policyv1listers.PodDisruptionBudgetLister
pdbLister policyv1listers.PodDisruptionBudgetLister
nodeLister corelisters.NodeLister
volumeAttachmentHandler *VolumeAttachmentHandler
Timeout time.Duration
Expand Down Expand Up @@ -182,8 +179,7 @@ func NewDrainOptions(
driver driver.Driver,
pvcLister corelisters.PersistentVolumeClaimLister,
pvLister corelisters.PersistentVolumeLister,
pdbV1beta1Lister policyv1beta1listers.PodDisruptionBudgetLister,
pdbV1Lister policyv1listers.PodDisruptionBudgetLister,
pdbLister policyv1listers.PodDisruptionBudgetLister,
nodeLister corelisters.NodeLister,
volumeAttachmentHandler *VolumeAttachmentHandler,
) *Options {
Expand All @@ -205,8 +201,7 @@ func NewDrainOptions(
Driver: driver,
pvcLister: pvcLister,
pvLister: pvLister,
pdbV1beta1Lister: pdbV1beta1Lister,
pdbV1Lister: pdbV1Lister,
pdbLister: pdbLister,
nodeLister: nodeLister,
volumeAttachmentHandler: volumeAttachmentHandler,
}
Expand Down Expand Up @@ -502,17 +497,17 @@ func (o *Options) evictPods(ctx context.Context, attemptEvict bool, pods []corev
}

doneCount := 0
var errors []error
var evictErrors []error

numPods := len(pods)
for doneCount < numPods {
err := <-returnCh
doneCount++
if err != nil {
errors = append(errors, err)
evictErrors = append(evictErrors, err)
}
}
return utilerrors.NewAggregate(errors)
return utilerrors.NewAggregate(evictErrors)
}

func (o *Options) evictPodsWithoutPv(ctx context.Context, attemptEvict bool, pods []*corev1.Pod,
Expand Down Expand Up @@ -722,27 +717,14 @@ func (o *Options) evictPodsWithPVInternal(
// Pod eviction failed because of PDB violation, we will retry one we are done with this list.
klog.V(3).Infof("Pod %s/%s couldn't be evicted from node %s. This may also occur due to PDB violation. Will be retried. Error: %v", pod.Namespace, pod.Name, pod.Spec.NodeName, err)

if k8sutils.ConstraintK8sGreaterEqual121.Check(o.kubernetesVersion) {
pdb := getPdbV1ForPod(o.pdbV1Lister, pod)
if pdb != nil {
if isMisconfiguredPdbV1(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
o.checkAndDeleteWorker(volumeAttachmentEventCh)
continue
}
}
} else {
pdb := getPdbV1beta1ForPod(o.pdbV1beta1Lister, pod)
if pdb != nil {
if isMisconfiguredPdbV1beta1(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
o.checkAndDeleteWorker(volumeAttachmentEventCh)
continue
}
pdb := getPdbForPod(o.pdbLister, pod)
if pdb != nil {
if isMisconfiguredPdb(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
o.checkAndDeleteWorker(volumeAttachmentEventCh)
continue
}
}

Expand Down Expand Up @@ -771,8 +753,8 @@ func (o *Options) evictPodsWithPVInternal(
)

podVolumeInfo := podVolumeInfoMap[getPodKey(pod)]
ctx, cancelFn := context.WithTimeout(ctx, o.getTerminationGracePeriod(pod)+o.PvDetachTimeout)
err = o.waitForDetach(ctx, podVolumeInfo, o.nodeName)
volDetachCtx, cancelFn := context.WithTimeout(ctx, o.getTerminationGracePeriod(pod)+o.PvDetachTimeout)
err = o.waitForDetach(volDetachCtx, podVolumeInfo, o.nodeName)
cancelFn()

if apierrors.IsNotFound(err) {
Expand All @@ -794,8 +776,8 @@ func (o *Options) evictPodsWithPVInternal(
time.Since(podEvictionStartTime),
)

ctx, cancelFn = context.WithTimeout(ctx, o.PvReattachTimeout)
err = o.waitForReattach(ctx, podVolumeInfo, o.nodeName, volumeAttachmentEventCh)
volReattachCtx, cancelFn := context.WithTimeout(ctx, o.PvReattachTimeout)
err = o.waitForReattach(volReattachCtx, podVolumeInfo, o.nodeName, volumeAttachmentEventCh)
cancelFn()

if err != nil {
Expand Down Expand Up @@ -1059,25 +1041,13 @@ func (o *Options) evictPodWithoutPVInternal(ctx context.Context, attemptEvict bo
// Pod couldn't be evicted because of PDB violation
klog.V(3).Infof("Pod %s/%s couldn't be evicted from node %s. This may also occur due to PDB violation. Will be retried. Error: %v", pod.Namespace, pod.Name, pod.Spec.NodeName, err)

if k8sutils.ConstraintK8sGreaterEqual121.Check(o.kubernetesVersion) {
pdb := getPdbV1ForPod(o.pdbV1Lister, pod)
if pdb != nil {
if isMisconfiguredPdbV1(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
return
}
}
} else {
pdb := getPdbV1beta1ForPod(o.pdbV1beta1Lister, pod)
if pdb != nil {
if isMisconfiguredPdbV1beta1(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
return
}
pdb := getPdbForPod(o.pdbLister, pod)
if pdb != nil {
if isMisconfiguredPdb(pdb) {
pdbErr := fmt.Errorf("error while evicting pod %q: pod disruption budget %s/%s is misconfigured and requires zero voluntary evictions",
pod.Name, pdb.Namespace, pdb.Name)
returnCh <- pdbErr
return
}
}

Expand All @@ -1086,11 +1056,7 @@ func (o *Options) evictPodWithoutPVInternal(ctx context.Context, attemptEvict bo

if o.ForceDeletePods {
// Skip waiting for pod termination in case of forced drain
if err == nil {
returnCh <- nil
} else {
returnCh <- err
}
returnCh <- nil
return
}

Expand Down Expand Up @@ -1194,23 +1160,7 @@ func (o *Options) RunCordonOrUncordon(ctx context.Context, desired bool) error {
return nil
}

func getPdbV1ForPod(pdbLister policyv1listers.PodDisruptionBudgetLister, pod *corev1.Pod) *policyv1.PodDisruptionBudget {
// GetPodPodDisruptionBudgets returns an error only if no PodDisruptionBudgets are found.
// We don't return that as an error to the caller.
pdbs, err := pdbLister.GetPodPodDisruptionBudgets(pod)
if err != nil {
klog.V(4).Infof("No PodDisruptionBudgets found for pod %s/%s.", pod.Namespace, pod.Name)
return nil
}

if len(pdbs) > 1 {
klog.Warningf("Pod %s/%s matches multiple PodDisruptionBudgets. Chose %q arbitrarily.", pod.Namespace, pod.Name, pdbs[0].Name)
}

return pdbs[0]
}

func getPdbV1beta1ForPod(pdbLister policyv1beta1listers.PodDisruptionBudgetLister, pod *corev1.Pod) *policyv1beta1.PodDisruptionBudget {
func getPdbForPod(pdbLister policyv1listers.PodDisruptionBudgetLister, pod *corev1.Pod) *policyv1.PodDisruptionBudget {
// GetPodPodDisruptionBudgets returns an error only if no PodDisruptionBudgets are found.
// We don't return that as an error to the caller.
pdbs, err := pdbLister.GetPodPodDisruptionBudgets(pod)
Expand All @@ -1226,15 +1176,7 @@ func getPdbV1beta1ForPod(pdbLister policyv1beta1listers.PodDisruptionBudgetListe
return pdbs[0]
}

func isMisconfiguredPdbV1(pdb *policyv1.PodDisruptionBudget) bool {
if pdb.ObjectMeta.Generation != pdb.Status.ObservedGeneration {
return false
}

return pdb.Status.ExpectedPods > 0 && pdb.Status.CurrentHealthy >= pdb.Status.ExpectedPods && pdb.Status.DisruptionsAllowed == 0
}

func isMisconfiguredPdbV1beta1(pdb *policyv1beta1.PodDisruptionBudget) bool {
func isMisconfiguredPdb(pdb *policyv1.PodDisruptionBudget) bool {
if pdb.ObjectMeta.Generation != pdb.Status.ObservedGeneration {
return false
}
Expand Down
3 changes: 1 addition & 2 deletions pkg/util/provider/drain/drain_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -157,8 +157,7 @@ var _ = Describe("drain", func() {
Out: GinkgoWriter,
pvcLister: fakePVCLister,
pvLister: fakePVLister,
pdbV1beta1Lister: nil,
pdbV1Lister: nil,
pdbLister: nil,
nodeLister: fakeNodeLister,
Timeout: 2 * time.Minute,
volumeAttachmentHandler: volumeAttachmentHandler,
Expand Down
24 changes: 7 additions & 17 deletions pkg/util/provider/machinecontroller/controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,11 @@ import (
runtimeutil "k8s.io/apimachinery/pkg/util/runtime"
coreinformers "k8s.io/client-go/informers/core/v1"
policyv1informers "k8s.io/client-go/informers/policy/v1"
policyv1beta1informers "k8s.io/client-go/informers/policy/v1beta1"
storageinformers "k8s.io/client-go/informers/storage/v1"
"k8s.io/client-go/kubernetes"
typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
corelisters "k8s.io/client-go/listers/core/v1"
policyv1listers "k8s.io/client-go/listers/policy/v1"
policyv1beta1listers "k8s.io/client-go/listers/policy/v1beta1"
storagelisters "k8s.io/client-go/listers/storage/v1"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
Expand Down Expand Up @@ -67,8 +65,7 @@ func NewController(
pvInformer coreinformers.PersistentVolumeInformer,
secretInformer coreinformers.SecretInformer,
nodeInformer coreinformers.NodeInformer,
pdbV1beta1Informer policyv1beta1informers.PodDisruptionBudgetInformer,
pdbV1Informer policyv1informers.PodDisruptionBudgetInformer,
pdbInformer policyv1informers.PodDisruptionBudgetInformer,
volumeAttachmentInformer storageinformers.VolumeAttachmentInformer,
machineClassInformer machineinformers.MachineClassInformer,
machineInformer machineinformers.MachineInformer,
Expand Down Expand Up @@ -137,13 +134,8 @@ func NewController(
controller.nodeSynced = nodeInformer.Informer().HasSynced
controller.machineSynced = machineInformer.Informer().HasSynced

if k8sutils.ConstraintK8sGreaterEqual121.Check(targetKubernetesVersion) {
controller.pdbV1Lister = pdbV1Informer.Lister()
controller.pdbV1Synced = pdbV1Informer.Informer().HasSynced
} else {
controller.pdbV1beta1Lister = pdbV1beta1Informer.Lister()
controller.pdbV1beta1Synced = pdbV1beta1Informer.Informer().HasSynced
}
controller.pdbLister = pdbInformer.Lister()
controller.pdbSynced = pdbInformer.Informer().HasSynced

// Secret Controller's Informers
_, _ = secretInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
Expand Down Expand Up @@ -248,8 +240,7 @@ type controller struct {
pvLister corelisters.PersistentVolumeLister
secretLister corelisters.SecretLister
nodeLister corelisters.NodeLister
pdbV1beta1Lister policyv1beta1listers.PodDisruptionBudgetLister
pdbV1Lister policyv1listers.PodDisruptionBudgetLister
pdbLister policyv1listers.PodDisruptionBudgetLister
volumeAttachementLister storagelisters.VolumeAttachmentLister
machineClassLister machinelisters.MachineClassLister
machineLister machinelisters.MachineLister
Expand All @@ -264,8 +255,7 @@ type controller struct {
pvcSynced cache.InformerSynced
pvSynced cache.InformerSynced
secretSynced cache.InformerSynced
pdbV1beta1Synced cache.InformerSynced
pdbV1Synced cache.InformerSynced
pdbSynced cache.InformerSynced
volumeAttachementSynced cache.InformerSynced
nodeSynced cache.InformerSynced
machineClassSynced cache.InformerSynced
Expand All @@ -288,12 +278,12 @@ func (c *controller) Run(workers int, stopCh <-chan struct{}) {
defer c.machineSafetyAPIServerQueue.ShutDown()

if k8sutils.ConstraintK8sGreaterEqual121.Check(c.targetKubernetesVersion) {
if !cache.WaitForCacheSync(stopCh, c.secretSynced, c.pvcSynced, c.pvSynced, c.pdbV1Synced, c.volumeAttachementSynced, c.nodeSynced, c.machineClassSynced, c.machineSynced) {
if !cache.WaitForCacheSync(stopCh, c.secretSynced, c.pvcSynced, c.pvSynced, c.pdbSynced, c.volumeAttachementSynced, c.nodeSynced, c.machineClassSynced, c.machineSynced) {
runtimeutil.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
return
}
} else {
if !cache.WaitForCacheSync(stopCh, c.secretSynced, c.pvcSynced, c.pvSynced, c.pdbV1beta1Synced, c.volumeAttachementSynced, c.nodeSynced, c.machineClassSynced, c.machineSynced) {
if !cache.WaitForCacheSync(stopCh, c.secretSynced, c.pvcSynced, c.pvSynced, c.volumeAttachementSynced, c.nodeSynced, c.machineClassSynced, c.machineSynced) {
runtimeutil.HandleError(fmt.Errorf("Timed out waiting for caches to sync"))
return
}
Expand Down
3 changes: 1 addition & 2 deletions pkg/util/provider/machinecontroller/machine_util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1173,8 +1173,7 @@ func (c *controller) drainNode(ctx context.Context, deleteMachineRequest *driver
c.driver,
c.pvcLister,
c.pvLister,
c.pdbV1beta1Lister,
c.pdbV1Lister,
c.pdbLister,
c.nodeLister,
c.volumeAttachmentHandler,
)
Expand Down

0 comments on commit e91182a

Please sign in to comment.