Merge pull request #32 from vshn/add/inplace-resize

Add inplace pvc resize controller

.gitignore

@@ -1,5 +1,7 @@
 cover.out
 .github/release-notes.md
+.vscode
+__debug_bin*
 
 # Binaries for programs and plugins
 *.exe

Makefile

@@ -9,19 +9,19 @@ ENVTEST_ASSETS_DIR=$(shell pwd)/testbin
 all: fmt vet build
 
 .PHONY: build
-build: 
+build:
 	CGO_ENABLED=0 go build
 
 
 run: fmt vet ## Run against the configured Kubernetes cluster in ~/.kube/config
 	go run ./main.go
 
 .PHONY: test
-test: fmt  ## Run tests 
+test: fmt  ## Run tests
 	go test -tags="" ./... -coverprofile cover.out
 
 .PHONY: integration-test
-integration-test: export ENVTEST_K8S_VERSION = 1.19.x
+integration-test: export ENVTEST_K8S_VERSION = 1.24.x
 integration-test: ## Run integration tests with envtest
 	mkdir -p ${ENVTEST_ASSETS_DIR}
 	$(setup-envtest) use '$(ENVTEST_K8S_VERSION)!'

config/rbac/role.yaml

@@ -1,9 +1,7 @@
-
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
 metadata:
-  creationTimestamp: null
   name: controller-manager
 rules:
 - apiGroups:

controllers/controller_util_test.go

@@ -320,6 +320,16 @@ func rbNotExists(ctx context.Context, c client.Client, other *rbacv1.RoleBinding
 	return apierrors.IsNotFound(err) || (err == nil && rb.DeletionTimestamp != nil)
 }
 
+func pvcEqualSize(ctx context.Context, c client.Client, other *corev1.PersistentVolumeClaim, newSize string) bool {
+	pvc := &corev1.PersistentVolumeClaim{}
+	key := client.ObjectKeyFromObject(other)
+	err := c.Get(ctx, key, pvc)
+	if err != nil {
+		return false
+	}
+	return newSize == pvc.Spec.Resources.Requests.Storage().String()
+}
+
 // Only succeeds if the condition is valid for `waitFor` time.
 // Checks the condition every `tick`
 func consistently(t assert.TestingT, condition func() bool, waitFor time.Duration, tick time.Duration, msgAndArgs ...interface{}) bool {

controllers/inplace_controller.go

@@ -0,0 +1,87 @@
+package controllers
+
+import (
+	"context"
+	"time"
+
+	"github.com/vshn/statefulset-resize-controller/statefulset"
+	appsv1 "k8s.io/api/apps/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+// StatefulSetController is an interface for various implementations
+// of the StatefulSet controller.
+type StatefulSetController interface {
+	SetupWithManager(ctrl.Manager) error
+}
+
+// InplaceReconciler reconciles a StatefulSet object
+// It will resize the PVCs according to the sts template.
+type InplaceReconciler struct {
+	client.Client
+	Scheme   *runtime.Scheme
+	Recorder record.EventRecorder
+
+	RequeueAfter time.Duration
+	LabelName    string
+}
+
+// Reconcile is the main work loop, reacting to changes in statefulsets and initiating resizing of StatefulSets.
+func (r *InplaceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	l := log.FromContext(ctx).WithValues("statefulset", req.NamespacedName)
+	ctx = log.IntoContext(ctx, l)
+
+	sts := &appsv1.StatefulSet{}
+	err := r.Client.Get(ctx, req.NamespacedName, sts)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			return ctrl.Result{}, nil
+		}
+		return ctrl.Result{}, err
+	}
+
+	l.V(1).Info("Checking label for sts", "labelName", r.LabelName)
+	if sts.GetLabels()[r.LabelName] != "true" {
+		l.V(1).Info("Label not found, skipping sts")
+		return ctrl.Result{}, nil
+	}
+
+	l.Info("Found sts with label", "labelName", r.LabelName)
+
+	stsEntity, err := statefulset.NewEntity(sts)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	stsEntity.Pvcs, err = fetchResizablePVCs(ctx, r.Client, *stsEntity)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	if len(stsEntity.Pvcs) == 0 {
+		l.Info("All PVCs have the right size")
+		return ctrl.Result{}, nil
+	}
+
+	err = resizePVCsInplace(ctx, r.Client, stsEntity.Pvcs)
+	if err != nil {
+		r.Recorder.Event(sts, "Warning", "ResizeFailed", "There was an error during the PVC resize")
+		return ctrl.Result{}, err
+	}
+
+	r.Recorder.Event(sts, "Normal", "ResizeSuccessful", "All PVCs have been resized successfully")
+
+	return ctrl.Result{}, nil
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *InplaceReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&appsv1.StatefulSet{}).
+		Complete(r)
+}

controllers/inplace_controller_test.go

@@ -0,0 +1,169 @@
+//go:build integration
+// +build integration
+
+package controllers
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	appsv1 "k8s.io/api/apps/v1"
+	batchv1 "k8s.io/api/batch/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	storagev1 "k8s.io/api/storage/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/utils/pointer"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+)
+
+const (
+	testLabelName = "mylabel"
+)
+
+func TestInplaceController(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c, stop := startInplaceTestReconciler(t, ctx, "")
+	defer stop()
+
+	t.Run("InplaceE2E", func(t *testing.T) {
+
+		sc := &storagev1.StorageClass{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "mysc",
+				Annotations: map[string]string{
+					"storageclass.kubernetes.io/is-default-class": "true",
+				},
+			},
+			Provisioner:          "mysc",
+			AllowVolumeExpansion: pointer.Bool(true),
+		}
+
+		require.NoError(t, c.Create(ctx, sc))
+
+		t.Run("Don't touch correct PVCs", func(t *testing.T) {
+			t.Parallel()
+			ctx := context.Background()
+			ns := "e2e1"
+			require := require.New(t)
+			require.NoError(c.Create(ctx, &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: ns,
+				},
+			}))
+			pvcSize := "2G"
+			sts := newTestStatefulSet(ns, "test", 1, pvcSize)
+			sts.Labels = map[string]string{
+				testLabelName: "true",
+			}
+
+			pvc := applyResizablePVC(ctx, "data-test-0", ns, pvcSize, sts, c, require)
+
+			require.NoError(c.Create(ctx, sts))
+
+			consistently(t, func() bool {
+				return pvcEqualSize(ctx, c, pvc, pvcSize)
+			}, duration, interval, "PVCs equal size")
+
+		})
+		t.Run("Ignore STS without the label", func(t *testing.T) {
+			t.Parallel()
+			ctx := context.Background()
+			ns := "e2e2"
+			require := require.New(t)
+			require.NoError(c.Create(ctx, &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: ns,
+				},
+			}))
+			sts := newTestStatefulSet(ns, "test", 1, "2G")
+
+			pvc := applyResizablePVC(ctx, "data-test-0", ns, "1G", sts, c, require)
+
+			require.NoError(c.Create(ctx, sts))
+
+			consistently(t, func() bool {
+				return pvcEqualSize(ctx, c, pvc, "1G")
+			}, duration, interval, "PVCs equal size")
+		})
+		t.Run("Change PVCs if they not match", func(t *testing.T) {
+			t.Parallel()
+			ctx := context.Background()
+			ns := "e2e3"
+			require := require.New(t)
+			require.NoError(c.Create(ctx, &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: ns,
+				},
+			}))
+			sts := newTestStatefulSet(ns, "test", 1, "2G")
+			sts.Labels = map[string]string{
+				testLabelName: "true",
+			}
+
+			pvc := applyResizablePVC(ctx, "data-test-0", ns, "1G", sts, c, require)
+
+			require.NoError(c.Create(ctx, sts))
+
+			consistently(t, func() bool {
+				return pvcEqualSize(ctx, c, pvc, "2G")
+			}, duration, interval, "PVCs equal size")
+		})
+	})
+
+}
+
+// startInplaceTestReconciler sets up a separate test env and starts the controller
+func startInplaceTestReconciler(t *testing.T, ctx context.Context, crname string) (client.Client, func() error) {
+	req := require.New(t)
+
+	testEnv := &envtest.Environment{}
+	conf, err := testEnv.Start()
+	req.NoError(err)
+
+	s := runtime.NewScheme()
+	req.NoError(appsv1.AddToScheme(s))
+	req.NoError(corev1.AddToScheme(s))
+	req.NoError(batchv1.AddToScheme(s))
+	req.NoError(rbacv1.AddToScheme(s))
+	req.NoError(storagev1.AddToScheme(s))
+
+	mgr, err := ctrl.NewManager(conf, ctrl.Options{
+		Scheme: s,
+	})
+	req.NoError(err)
+	req.NoError((&InplaceReconciler{
+		Client:       mgr.GetClient(),
+		Scheme:       mgr.GetScheme(),
+		Recorder:     mgr.GetEventRecorderFor("statefulset-resize-controller"),
+		RequeueAfter: time.Second,
+		LabelName:    testLabelName,
+	}).SetupWithManager(mgr))
+	go func() {
+		req.NoError(mgr.Start(ctx))
+	}()
+
+	return mgr.GetClient(), testEnv.Stop
+}
+
+func applyResizablePVC(ctx context.Context, name, ns, size string, sts *appsv1.StatefulSet, c client.Client, require *require.Assertions) *corev1.PersistentVolumeClaim {
+	pvc := newSource(ns, name, size,
+		func(pvc *corev1.PersistentVolumeClaim) *corev1.PersistentVolumeClaim {
+			pvc.Labels = sts.Spec.Selector.MatchLabels
+			return pvc
+		})
+
+	pvc.Spec.StorageClassName = pointer.String("mysc")
+	require.NoError(c.Create(ctx, pvc))
+
+	// we need to set the PVC to bound in order for the resize to work
+	pvc.Status.Phase = corev1.ClaimBound
+	require.NoError(c.Status().Update(ctx, pvc))
+	return pvc
+}

controllers/pvc.go

@@ -16,14 +16,14 @@ import (
 )
 
 // getResizablePVCs fetches the information of all PVCs that are smaller than the request of the statefulset
-func (r StatefulSetReconciler) fetchResizablePVCs(ctx context.Context, si statefulset.Entity) ([]pvc.Entity, error) {
+func fetchResizablePVCs(ctx context.Context, cl client.Client, si statefulset.Entity) ([]pvc.Entity, error) {
 	// NOTE(glrf) This will get _all_ PVCs that belonged to the sts. Even the ones not used anymore (i.e. if scaled up and down).
 	sts, err := si.StatefulSet()
 	if err != nil {
 		return nil, err
 	}
 	pvcs := corev1.PersistentVolumeClaimList{}
-	if err := r.List(ctx, &pvcs, client.InNamespace(sts.Namespace), client.MatchingLabels(sts.Spec.Selector.MatchLabels)); err != nil {
+	if err := cl.List(ctx, &pvcs, client.InNamespace(sts.Namespace), client.MatchingLabels(sts.Spec.Selector.MatchLabels)); err != nil {
 		return nil, err
 	}
 	pis := filterResizablePVCs(ctx, *sts, pvcs.Items)
@@ -109,3 +109,22 @@ func (r *StatefulSetReconciler) resizePVCs(ctx context.Context, oldPIs []pvc.Ent
 	}
 	return pis, nil
 }
+
+func resizePVCsInplace(ctx context.Context, cl client.Client, PVCs []pvc.Entity) error {
+	l := log.FromContext(ctx)
+
+	for _, pvc := range PVCs {
+		l.Info("Updating PVC", "PVCName", pvc.SourceName)
+
+		resizedPVC := pvc.GetResizedSource()
+		resizedPVC.Spec.StorageClassName = pvc.SourceStorageClass
+		resizedPVC.Spec.VolumeName = pvc.Spec.VolumeName
+
+		err := cl.Update(ctx, resizedPVC)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

controllers/statefulset.go

@@ -50,7 +50,7 @@ func (r StatefulSetReconciler) fetchStatefulSet(ctx context.Context, namespacedN
 	}
 
 	if !sts.Resizing() {
-		sts.Pvcs, err = r.fetchResizablePVCs(ctx, *sts)
+		sts.Pvcs, err = fetchResizablePVCs(ctx, r.Client, *sts)
 		return sts, err
 	}
 	return sts, nil