// Copyright 2020-2026 Project Capsule Authors // SPDX-License-Identifier: Apache-2.0 package pv import ( "context" "fmt" corev1 "k8s.io/api/core/v1" apiequality "k8s.io/apimachinery/pkg/api/equality" "k8s.io/apimachinery/pkg/api/errors" "k8s.io/client-go/util/retry" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" log2 "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" capsulev1beta2 "github.com/projectcapsule/capsule/api/v1beta2" "github.com/projectcapsule/capsule/internal/controllers/utils" "github.com/projectcapsule/capsule/pkg/tenant" capsuleutils "github.com/projectcapsule/capsule/pkg/utils" ) type Controller struct { client client.Client reader client.Reader label string } func (c *Controller) SetupWithManager(mgr ctrl.Manager, ctrlConfig utils.ControllerOptions) error { label, err := capsuleutils.GetTypeLabel(&capsulev1beta2.Tenant{}) if err != nil { return err } c.client = mgr.GetClient() c.reader = mgr.GetAPIReader() c.label = label return ctrl.NewControllerManagedBy(mgr). Named("capsule/persistentvolumes"). For(&corev1.PersistentVolume{}, builder.WithPredicates(persistentVolumePredicate(c.label))). WithOptions(ctrlConfig.Runtime.ToControllerOptions()). Complete(c) } func persistentVolumePredicate(label string) predicate.Predicate { return predicate.Funcs{ CreateFunc: func(e event.CreateEvent) bool { pv, ok := e.Object.(*corev1.PersistentVolume) return ok && pv.Spec.ClaimRef != nil }, DeleteFunc: func(event.DeleteEvent) bool { return false }, UpdateFunc: func(e event.UpdateEvent) bool { oldPV, oldOK := e.ObjectOld.(*corev1.PersistentVolume) newPV, newOK := e.ObjectNew.(*corev1.PersistentVolume) if !oldOK || !newOK || newPV.Spec.ClaimRef == nil { return false } return !apiequality.Semantic.DeepEqual(oldPV.Spec.ClaimRef, newPV.Spec.ClaimRef) || persistentVolumeLabelChanged(oldPV, newPV, label) }, GenericFunc: func(event.GenericEvent) bool { return false }, } } func persistentVolumeLabelChanged(oldPV, newPV *corev1.PersistentVolume, label string) bool { oldValue, oldExists := oldPV.GetLabels()[label] newValue, newExists := newPV.GetLabels()[label] return oldExists != newExists || oldValue != newValue } func (c *Controller) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { log := log2.FromContext(ctx) persistentVolume := corev1.PersistentVolume{} if err := c.client.Get(ctx, request.NamespacedName, &persistentVolume); err != nil { if errors.IsNotFound(err) { log.V(5).Info("skipping reconciliation, resource may have been deleted") return reconcile.Result{}, nil } log.Error(err, "cannot retrieve corev1.PersistentVolume") return reconcile.Result{}, err } if persistentVolume.Spec.ClaimRef == nil { log.Info("skipping reconciliation, missing claimRef") return reconcile.Result{}, nil } tnt, err := c.resolveNamespaceTenant(ctx, persistentVolume.Spec.ClaimRef.Namespace) if err != nil { log.Error(err, "unable to retrieve Tenant from the claimRef") return reconcile.Result{}, err } if tnt == nil { log.V(4).Info("skipping reconciliation, PV is claimed by a PVC not managed in a Tenant") return reconcile.Result{}, nil } if persistentVolume.GetLabels()[c.label] == tnt.GetName() { return reconcile.Result{}, nil } retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error { pv := persistentVolume if err = c.client.Get(ctx, request.NamespacedName, &pv); err != nil { return err } if pv.Spec.ClaimRef == nil || pv.Spec.ClaimRef.Namespace != persistentVolume.Spec.ClaimRef.Namespace { return nil } labels := pv.GetLabels() if labels == nil { labels = map[string]string{} } if labels[c.label] == tnt.GetName() { return nil } labels[c.label] = tnt.GetName() pv.SetLabels(labels) return c.client.Update(ctx, &pv) }) if retryErr != nil { log.Error(retryErr, "unable to update PersistentVolume with Capsule label") return reconcile.Result{}, retryErr } return reconcile.Result{}, nil } func (c *Controller) resolveNamespaceTenant( ctx context.Context, namespace string, ) (*capsulev1beta2.Tenant, error) { reader := c.reader if reader == nil { reader = c.client } ns := &corev1.Namespace{} if err := reader.Get(ctx, client.ObjectKey{Name: namespace}, ns); err != nil { if errors.IsNotFound(err) { return nil, nil } return nil, fmt.Errorf("get claim namespace %q: %w", namespace, err) } tnt, err := tenant.ResolveNamespaceTenant(ctx, reader, ns) if err != nil { return nil, fmt.Errorf("resolve claim namespace %q: %w", namespace, err) } return tnt, nil }