Skip to content

Commit de2e56e

Browse files
k8s-ci-robotk8s-ci-robot
authored
Merge pull request #99 from gnufied/update-release-10-branch
Update release 1.0 branch
2 parents d017b1f + dddcb17 commit de2e56e

File tree

10 files changed

+247
-110
lines changed

10 files changed

+247
-110
lines changed

‎README.md

+1-1
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@ Note that the external-resizer does not scale with more replicas. Only one exter
4444

4545
* `--leader-election-namespace`: Namespace where the leader election resource lives. Defaults to the pod namespace if not set.
4646

47-
* `--csiTimeout <duration>`: Timeout of all calls to CSI driver. It should be set to value that accommodates majority of `ControllerExpandVolume` calls. 15 seconds is used by default.
47+
* `--timeout <duration>`: Timeout of all calls to CSI driver. It should be set to value that accommodates majority of `ControllerExpandVolume` calls. 10 seconds is used by default.
4848

4949
* `--retry-interval-start`: The starting value of the exponential backoff for failures. 1 second is used by default.
5050

‎cmd/csi-resizer/main.go

+7-5
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,12 @@ import (
2020
"context"
2121
"flag"
2222
"fmt"
23+
"os"
24+
"time"
25+
2326
"k8s.io/client-go/kubernetes"
2427
"k8s.io/client-go/rest"
2528
"k8s.io/client-go/tools/clientcmd"
26-
"os"
27-
"time"
2829

2930
"k8s.io/client-go/util/workqueue"
3031

@@ -44,8 +45,9 @@ var (
4445
resyncPeriod = flag.Duration("resync-period", time.Minute*10, "Resync period for cache")
4546
workers = flag.Int("workers", 10, "Concurrency to process multiple resize requests")
4647

47-
csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
48-
csiTimeout = flag.Duration("csiTimeout", 15*time.Second, "Timeout for waiting for CSI driver socket.")
48+
csiAddress = flag.String("csi-address", "/run/csi/socket", "Address of the CSI driver socket.")
49+
timeout = flag.Duration("timeout", 10*time.Second, "Timeout for waiting for CSI driver socket.")
50+
4951
showVersion = flag.Bool("version", false, "Show version")
5052

5153
retryIntervalStart = flag.Duration("retry-interval-start", time.Second, "Initial retry interval of failed volume resize. It exponentially increases with each failure, up to retry-interval-max.")
@@ -99,7 +101,7 @@ func main() {
99101

100102
csiResizer, err := resizer.NewResizer(
101103
*csiAddress,
102-
*csiTimeout,
104+
*timeout,
103105
kubeClient,
104106
informerFactory,
105107
*metricsAddress,

‎pkg/controller/controller.go

+114-50
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,9 @@ import (
2626
"google.golang.org/grpc/codes"
2727
"google.golang.org/grpc/status"
2828
v1 "k8s.io/api/core/v1"
29-
k8serrors "k8s.io/apimachinery/pkg/api/errors"
3029
"k8s.io/apimachinery/pkg/api/resource"
3130
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
31+
"k8s.io/apimachinery/pkg/types"
3232
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
3333
"k8s.io/apimachinery/pkg/util/wait"
3434
"k8s.io/client-go/informers"
@@ -55,15 +55,18 @@ type resizeController struct {
5555
kubeClient kubernetes.Interface
5656
claimQueue workqueue.RateLimitingInterface
5757
eventRecorder record.EventRecorder
58-
pvLister corelisters.PersistentVolumeLister
5958
pvSynced cache.InformerSynced
60-
pvcLister corelisters.PersistentVolumeClaimLister
6159
pvcSynced cache.InformerSynced
6260

6361
usedPVCs *inUsePVCStore
6462

65-
podLister corelisters.PodLister
66-
podListerSynced cache.InformerSynced
63+
podLister corelisters.PodLister
64+
podListerSynced cache.InformerSynced
65+
66+
// a cache to store PersistentVolume objects
67+
volumes cache.Store
68+
// a cache to store PersistentVolumeClaim objects
69+
claims cache.Store
6770
handleVolumeInUseError bool
6871
}
6972

@@ -91,11 +94,11 @@ func NewResizeController(
9194
name: name,
9295
resizer: resizer,
9396
kubeClient: kubeClient,
94-
pvLister: pvInformer.Lister(),
9597
pvSynced: pvInformer.Informer().HasSynced,
96-
pvcLister: pvcInformer.Lister(),
9798
pvcSynced: pvcInformer.Informer().HasSynced,
9899
claimQueue: claimQueue,
100+
volumes: pvInformer.Informer().GetStore(),
101+
claims: pvcInformer.Informer().GetStore(),
99102
eventRecorder: eventRecorder,
100103
usedPVCs: newUsedPVCStore(),
101104
handleVolumeInUseError: handleVolumeInUseError,
@@ -266,6 +269,7 @@ func (ctrl *resizeController) syncPVCs() {
266269

267270
if err := ctrl.syncPVC(key.(string)); err != nil {
268271
// Put PVC back to the queue so that we can retry later.
272+
klog.Errorf("Error syncing PVC: %v", err)
269273
ctrl.claimQueue.AddRateLimited(key)
270274
} else {
271275
ctrl.claimQueue.Forget(key)
@@ -278,33 +282,42 @@ func (ctrl *resizeController) syncPVC(key string) error {
278282

279283
namespace, name, err := cache.SplitMetaNamespaceKey(key)
280284
if err != nil {
281-
klog.Errorf("Split meta namespace key of pvc %s failed: %v", key, err)
282-
return err
285+
return fmt.Errorf("getting namespace and name from key %s failed: %v", key, err)
283286
}
284287

285-
pvc, err := ctrl.pvcLister.PersistentVolumeClaims(namespace).Get(name)
288+
pvcObject, exists, err := ctrl.claims.GetByKey(key)
286289
if err != nil {
287-
if k8serrors.IsNotFound(err) {
288-
klog.V(3).Infof("PVC %s/%s is deleted, no need to process it", namespace, name)
289-
return nil
290-
}
291-
klog.Errorf("Get PVC %s/%s failed: %v", namespace, name, err)
292-
return err
290+
return fmt.Errorf("getting PVC %s/%s failed: %v", namespace, name, err)
291+
}
292+
293+
if !exists {
294+
klog.V(3).Infof("PVC %s/%s is deleted or does not exist", namespace, name)
295+
return nil
296+
}
297+
298+
pvc, ok := pvcObject.(*v1.PersistentVolumeClaim)
299+
if !ok {
300+
return fmt.Errorf("expected PVC got: %v", pvcObject)
293301
}
294302

295303
if !ctrl.pvcNeedResize(pvc) {
296304
klog.V(4).Infof("No need to resize PVC %q", util.PVCKey(pvc))
297305
return nil
298306
}
299307

300-
pv, err := ctrl.pvLister.Get(pvc.Spec.VolumeName)
308+
volumeObj, exists, err := ctrl.volumes.GetByKey(pvc.Spec.VolumeName)
301309
if err != nil {
302-
if k8serrors.IsNotFound(err) {
303-
klog.V(3).Infof("PV %s is deleted, no need to process it", pvc.Spec.VolumeName)
304-
return nil
305-
}
306-
klog.Errorf("Get PV %q of pvc %q failed: %v", pvc.Spec.VolumeName, util.PVCKey(pvc), err)
307-
return err
310+
return fmt.Errorf("Get PV %q of pvc %q failed: %v", pvc.Spec.VolumeName, util.PVCKey(pvc), err)
311+
}
312+
313+
if !exists {
314+
klog.Warningf("PV %q bound to PVC %s not found", pvc.Spec.VolumeName, util.PVCKey(pvc))
315+
return nil
316+
}
317+
318+
pv, ok := volumeObj.(*v1.PersistentVolume)
319+
if !ok {
320+
return fmt.Errorf("expected volume but got %+v", volumeObj)
308321
}
309322

310323
if !ctrl.pvNeedResize(pvc, pv) {
@@ -369,8 +382,7 @@ func (ctrl *resizeController) pvNeedResize(pvc *v1.PersistentVolumeClaim, pv *v1
369382
// 3. Mark pvc as resizing finished(no error, no need to resize fs), need resizing fs or resize failed.
370383
func (ctrl *resizeController) resizePVC(pvc *v1.PersistentVolumeClaim, pv *v1.PersistentVolume) error {
371384
if updatedPVC, err := ctrl.markPVCResizeInProgress(pvc); err != nil {
372-
klog.Errorf("Mark pvc %q as resizing failed: %v", util.PVCKey(pvc), err)
373-
return err
385+
return fmt.Errorf("marking pvc %q as resizing failed: %v", util.PVCKey(pvc), err)
374386
} else if updatedPVC != nil {
375387
pvc = updatedPVC
376388
}
@@ -406,7 +418,6 @@ func (ctrl *resizeController) resizePVC(pvc *v1.PersistentVolumeClaim, pv *v1.Pe
406418
// Record an event to indicate that resize operation is failed.
407419
ctrl.eventRecorder.Eventf(pvc, v1.EventTypeWarning, util.VolumeResizeFailed, err.Error())
408420
}
409-
410421
return err
411422
}
412423

@@ -426,25 +437,48 @@ func (ctrl *resizeController) resizeVolume(
426437
newSize, fsResizeRequired, err := ctrl.resizer.Resize(pv, requestSize)
427438

428439
if err != nil {
429-
klog.Errorf("Resize volume %q by resizer %q failed: %v", pv.Name, ctrl.name, err)
430440
// if this error was a in-use error then it must be tracked so as we don't retry without
431441
// first verifying if volume is in-use
432442
if inUseError(err) {
433443
ctrl.usedPVCs.addPVCWithInUseError(pvc)
434444
}
435-
return newSize, fsResizeRequired, fmt.Errorf("resize volume %s failed: %v", pv.Name, err)
445+
return newSize, fsResizeRequired, fmt.Errorf("resize volume %q by resizer %q failed: %v", pv.Name, ctrl.name, err)
436446
}
437447
klog.V(4).Infof("Resize volume succeeded for volume %q, start to update PV's capacity", pv.Name)
438448

439-
if err := util.UpdatePVCapacity(pv, newSize, ctrl.kubeClient); err != nil {
440-
klog.Errorf("Update capacity of PV %q to %s failed: %v", pv.Name, newSize.String(), err)
449+
err = ctrl.updatePVCapacity(pv, newSize)
450+
if err != nil {
441451
return newSize, fsResizeRequired, err
442452
}
443453
klog.V(4).Infof("Update capacity of PV %q to %s succeeded", pv.Name, newSize.String())
444454

445455
return newSize, fsResizeRequired, nil
446456
}
447457

458+
func (ctrl *resizeController) markPVCAsFSResizeRequired(pvc *v1.PersistentVolumeClaim) error {
459+
pvcCondition := v1.PersistentVolumeClaimCondition{
460+
Type: v1.PersistentVolumeClaimFileSystemResizePending,
461+
Status: v1.ConditionTrue,
462+
LastTransitionTime: metav1.Now(),
463+
Message: "Waiting for user to (re-)start a pod to finish file system resize of volume on node.",
464+
}
465+
newPVC := pvc.DeepCopy()
466+
newPVC.Status.Conditions = util.MergeResizeConditionsOfPVC(newPVC.Status.Conditions,
467+
[]v1.PersistentVolumeClaimCondition{pvcCondition})
468+
469+
_, err := ctrl.patchClaim(pvc, newPVC)
470+
471+
if err != nil {
472+
return fmt.Errorf("Mark PVC %q as file system resize required failed: %v", util.PVCKey(pvc), err)
473+
}
474+
475+
klog.V(4).Infof("Mark PVC %q as file system resize required", util.PVCKey(pvc))
476+
ctrl.eventRecorder.Eventf(pvc, v1.EventTypeNormal,
477+
util.FileSystemResizeRequired, "Require file system resize of volume on node")
478+
479+
return nil
480+
}
481+
448482
func (ctrl *resizeController) markPVCResizeInProgress(pvc *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
449483
// Mark PVC as Resize Started
450484
progressCondition := v1.PersistentVolumeClaimCondition{
@@ -455,7 +489,12 @@ func (ctrl *resizeController) markPVCResizeInProgress(pvc *v1.PersistentVolumeCl
455489
newPVC := pvc.DeepCopy()
456490
newPVC.Status.Conditions = util.MergeResizeConditionsOfPVC(newPVC.Status.Conditions,
457491
[]v1.PersistentVolumeClaimCondition{progressCondition})
458-
return util.PatchPVCStatus(pvc, newPVC, ctrl.kubeClient)
492+
493+
updatedPVC, err := ctrl.patchClaim(pvc, newPVC)
494+
if err != nil {
495+
return nil, err
496+
}
497+
return updatedPVC, nil
459498
}
460499

461500
func (ctrl *resizeController) markPVCResizeFinished(
@@ -464,9 +503,10 @@ func (ctrl *resizeController) markPVCResizeFinished(
464503
newPVC := pvc.DeepCopy()
465504
newPVC.Status.Capacity[v1.ResourceStorage] = newSize
466505
newPVC.Status.Conditions = util.MergeResizeConditionsOfPVC(pvc.Status.Conditions, []v1.PersistentVolumeClaimCondition{})
467-
if _, err := util.PatchPVCStatus(pvc, newPVC, ctrl.kubeClient); err != nil {
468-
klog.Errorf("Mark PVC %q as resize finished failed: %v", util.PVCKey(pvc), err)
469-
return err
506+
507+
_, err := ctrl.patchClaim(pvc, newPVC)
508+
if err != nil {
509+
return fmt.Errorf("Mark PVC %q as resize finished failed: %v", util.PVCKey(pvc), err)
470510
}
471511

472512
klog.V(4).Infof("Resize PVC %q finished", util.PVCKey(pvc))
@@ -475,28 +515,52 @@ func (ctrl *resizeController) markPVCResizeFinished(
475515
return nil
476516
}
477517

478-
func (ctrl *resizeController) markPVCAsFSResizeRequired(pvc *v1.PersistentVolumeClaim) error {
479-
pvcCondition := v1.PersistentVolumeClaimCondition{
480-
Type: v1.PersistentVolumeClaimFileSystemResizePending,
481-
Status: v1.ConditionTrue,
482-
LastTransitionTime: metav1.Now(),
483-
Message: "Waiting for user to (re-)start a pod to finish file system resize of volume on node.",
518+
func (ctrl *resizeController) patchClaim(oldPVC, newPVC *v1.PersistentVolumeClaim) (*v1.PersistentVolumeClaim, error) {
519+
patchBytes, err := util.GetPVCPatchData(oldPVC, newPVC)
520+
if err != nil {
521+
return nil, fmt.Errorf("can't patch status of PVC %s as generate path data failed: %v", util.PVCKey(oldPVC), err)
484522
}
485-
newPVC := pvc.DeepCopy()
486-
newPVC.Status.Conditions = util.MergeResizeConditionsOfPVC(newPVC.Status.Conditions,
487-
[]v1.PersistentVolumeClaimCondition{pvcCondition})
488-
489-
if _, err := util.PatchPVCStatus(pvc, newPVC, ctrl.kubeClient); err != nil {
490-
klog.Errorf("Mark PVC %q as file system resize required failed: %v", util.PVCKey(pvc), err)
491-
return err
523+
updatedClaim, updateErr := ctrl.kubeClient.CoreV1().PersistentVolumeClaims(oldPVC.Namespace).
524+
Patch(context.TODO(), oldPVC.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
525+
if updateErr != nil {
526+
return nil, fmt.Errorf("can't patch status of PVC %s with %v", util.PVCKey(oldPVC), updateErr)
492527
}
493-
klog.V(4).Infof("Mark PVC %q as file system resize required", util.PVCKey(pvc))
494-
ctrl.eventRecorder.Eventf(pvc, v1.EventTypeNormal,
495-
util.FileSystemResizeRequired, "Require file system resize of volume on node")
528+
err = ctrl.claims.Update(updatedClaim)
529+
if err != nil {
530+
return nil, fmt.Errorf("error updating PVC %s in local cache: %v", util.PVCKey(newPVC), err)
531+
}
532+
533+
return updatedClaim, nil
534+
}
535+
536+
func (ctrl *resizeController) updatePVCapacity(pv *v1.PersistentVolume, newCapacity resource.Quantity) error {
537+
klog.V(4).Infof("Resize volume succeeded for volume %q, start to update PV's capacity", pv.Name)
538+
newPV := pv.DeepCopy()
539+
newPV.Spec.Capacity[v1.ResourceStorage] = newCapacity
496540

541+
_, err := ctrl.patchPersistentVolume(pv, newPV)
542+
if err != nil {
543+
return fmt.Errorf("updating capacity of PV %q to %s failed: %v", pv.Name, newCapacity.String(), err)
544+
}
497545
return nil
498546
}
499547

548+
func (ctrl *resizeController) patchPersistentVolume(oldPV, newPV *v1.PersistentVolume) (*v1.PersistentVolume, error) {
549+
patchBytes, err := util.GetPatchData(oldPV, newPV)
550+
if err != nil {
551+
return nil, fmt.Errorf("can't update capacity of PV %s as generate path data failed: %v", newPV.Name, err)
552+
}
553+
updatedPV, updateErr := ctrl.kubeClient.CoreV1().PersistentVolumes().Patch(context.TODO(), newPV.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{})
554+
if updateErr != nil {
555+
return nil, fmt.Errorf("update capacity of PV %s failed: %v", newPV.Name, updateErr)
556+
}
557+
err = ctrl.volumes.Update(updatedPV)
558+
if err != nil {
559+
return nil, fmt.Errorf("error updating PV %s in local cache: %v", newPV.Name, err)
560+
}
561+
return updatedPV, nil
562+
}
563+
500564
func parsePod(obj interface{}) *v1.Pod {
501565
if obj == nil {
502566
return nil

0 commit comments

Comments
 (0)