Modify volumebroker to filter volumes before aggregation (#1407)

Author: kasabe28

broker/volumebroker/cmd/volumebroker/app/app.go

@@ -140,5 +140,9 @@ func Run(ctx context.Context, opts Options) error {
 	if err := grpcSrv.Serve(l); err != nil {
 		return fmt.Errorf("error serving: %w", err)
 	}
+
+	if err := srv.SetVolumeUIDLabelToAllVolumes(ctx, log); err != nil {
+		return fmt.Errorf("failed to set volume uid label to all brokered volumes: %w", err)
+	}
 	return nil
 }

broker/volumebroker/server/event_list.go

@@ -40,8 +40,8 @@ func (s *Server) listEvents(ctx context.Context) ([]*irievent.Event, error) {
 
 	var iriEvents []*irievent.Event
 	for _, volumeEvent := range volumeEventList.Items {
-		ironcoreVolume, err := s.getIronCoreVolume(ctx, volumeEvent.InvolvedObject.Name)
-		if err != nil {
+		ironcoreVolume := &storagev1alpha1.Volume{}
+		if err := s.getManagedAndCreated(ctx, volumeEvent.InvolvedObject.Name, ironcoreVolume); err != nil {
 			log.V(1).Info("Unable to get ironcore volume", "VolumeName", volumeEvent.InvolvedObject.Name)
 			continue
 		}

broker/volumebroker/server/server.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 
 	"github.com/go-logr/logr"
+	"github.com/ironcore-dev/controller-utils/metautils"
 	computev1alpha1 "github.com/ironcore-dev/ironcore/api/compute/v1alpha1"
 	ipamv1alpha1 "github.com/ironcore-dev/ironcore/api/ipam/v1alpha1"
 	networkingv1alpha1 "github.com/ironcore-dev/ironcore/api/networking/v1alpha1"
@@ -17,6 +18,7 @@ import (
 	volumebrokerv1alpha1 "github.com/ironcore-dev/ironcore/broker/volumebroker/api/v1alpha1"
 	"github.com/ironcore-dev/ironcore/broker/volumebroker/apiutils"
 	iri "github.com/ironcore-dev/ironcore/iri/apis/volume/v1alpha1"
+	volumepoolletv1alpha1 "github.com/ironcore-dev/ironcore/poollet/volumepoollet/api/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -140,3 +142,36 @@ func (s *Server) getManagedAndCreated(ctx context.Context, name string, obj clie
 	}
 	return nil
 }
+
+func (s *Server) SetVolumeUIDLabelToAllVolumes(ctx context.Context, log logr.Logger) error {
+	volumeList := &storagev1alpha1.VolumeList{}
+	log.V(2).Info("Listing brokered volumes")
+	if err := s.listManagedAndCreated(ctx, volumeList, nil); err != nil {
+		return fmt.Errorf("error listing ironcore volumes: %w", err)
+	}
+
+	for i := range volumeList.Items {
+		volume := &volumeList.Items[i]
+		labels, err := apiutils.GetLabelsAnnotation(volume)
+		if err != nil {
+			return fmt.Errorf("failed to get labels annotation: %w", err)
+		}
+		if labels == nil {
+			log.V(2).Info("Labels are nil", "name", volume.Name, "namespace", volume.Namespace)
+			continue
+		}
+
+		volumeUid := labels[volumepoolletv1alpha1.VolumeUIDLabel]
+		if volumeUid == "" {
+			log.V(2).Info("Volume uid label is empty", "name", volume.Name, "namespace", volume.Namespace)
+			continue
+		}
+		log.V(2).Info("Setting volume uid label for", "name", volume.Name, "namespace", volume.Namespace, "labelKey", volumepoolletv1alpha1.VolumeUIDLabel, "labelValue", volumeUid)
+		base := volume.DeepCopy()
+		metautils.SetLabel(volume, volumepoolletv1alpha1.VolumeUIDLabel, volumeUid)
+		if err := s.client.Patch(ctx, volume, client.MergeFrom(base)); err != nil {
+			return fmt.Errorf("error patching volume uid label: %w", err)
+		}
+	}
+	return nil
+}

broker/volumebroker/server/volume_create.go

@@ -65,6 +65,7 @@ func (s *Server) getIronCoreVolumeConfig(_ context.Context, volume *iri.Volume)
 		s.brokerDownwardAPILabels,
 		volumepoolletv1alpha1.VolumeDownwardAPIPrefix,
 	)
+	labels[volumepoolletv1alpha1.VolumeUIDLabel] = volume.GetMetadata().GetLabels()[volumepoolletv1alpha1.VolumeUIDLabel]
 
 	var image string
 	var volumeSnapshotRef *corev1.LocalObjectReference

broker/volumebroker/server/volume_create_test.go

@@ -52,8 +52,9 @@ var _ = Describe("CreateVolume", func() {
 		By("inspecting the ironcore volume")
 		Expect(ironcoreVolume.Labels).To(Equal(map[string]string{
 			poolletutils.DownwardAPILabel(volumepoolletv1alpha1.VolumeDownwardAPIPrefix, "root-volume-uid"): "foobar",
-			volumebrokerv1alpha1.CreatedLabel: "true",
-			volumebrokerv1alpha1.ManagerLabel: volumebrokerv1alpha1.VolumeBrokerManager,
+			volumebrokerv1alpha1.CreatedLabel:    "true",
+			volumebrokerv1alpha1.ManagerLabel:    volumebrokerv1alpha1.VolumeBrokerManager,
+			volumepoolletv1alpha1.VolumeUIDLabel: "foobar",
 		}))
 		encodedIRIAnnotations, err := apiutils.EncodeAnnotationsAnnotation(nil)
 		Expect(err).NotTo(HaveOccurred())

broker/volumebroker/server/volume_list.go

@@ -10,29 +10,35 @@ import (
 	storagev1alpha1 "github.com/ironcore-dev/ironcore/api/storage/v1alpha1"
 	"github.com/ironcore-dev/ironcore/broker/common"
 	volumebrokerv1alpha1 "github.com/ironcore-dev/ironcore/broker/volumebroker/api/v1alpha1"
-	"github.com/ironcore-dev/ironcore/broker/volumebroker/apiutils"
 	iri "github.com/ironcore-dev/ironcore/iri/apis/volume/v1alpha1"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/labels"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func (s *Server) listManagedAndCreated(ctx context.Context, list client.ObjectList) error {
-	return s.client.List(ctx, list,
+func (s *Server) listManagedAndCreated(ctx context.Context, ironcoreVolumeList *storagev1alpha1.VolumeList, filter *iri.VolumeFilter) error {
+	matchingLabels := client.MatchingLabels{
+		volumebrokerv1alpha1.ManagerLabel: volumebrokerv1alpha1.VolumeBrokerManager,
+		volumebrokerv1alpha1.CreatedLabel: "true",
+	}
+
+	if filter != nil && filter.LabelSelector != nil {
+		for k := range filter.LabelSelector {
+			matchingLabels[k] = filter.LabelSelector[k]
+		}
+	}
+
+	return s.client.List(ctx, ironcoreVolumeList,
 		client.InNamespace(s.namespace),
-		client.MatchingLabels{
-			volumebrokerv1alpha1.ManagerLabel: volumebrokerv1alpha1.VolumeBrokerManager,
-			volumebrokerv1alpha1.CreatedLabel: "true",
-		},
+		matchingLabels,
 	)
 }
 
-func (s *Server) listAggregateIronCoreVolumes(ctx context.Context) ([]AggregateIronCoreVolume, error) {
+func (s *Server) listAggregateIronCoreVolumes(ctx context.Context, filter *iri.VolumeFilter) ([]AggregateIronCoreVolume, error) {
 	ironcoreVolumeList := &storagev1alpha1.VolumeList{}
-	if err := s.listManagedAndCreated(ctx, ironcoreVolumeList); err != nil {
+	if err := s.listManagedAndCreated(ctx, ironcoreVolumeList, filter); err != nil {
 		return nil, fmt.Errorf("error listing ironcore volumes: %w", err)
 	}
 
@@ -132,21 +138,6 @@ func (s *Server) aggregateIronCoreVolume(
 	}, nil
 }
 
-func (s *Server) getIronCoreVolume(ctx context.Context, id string) (*storagev1alpha1.Volume, error) {
-	ironcoreVolume := &storagev1alpha1.Volume{}
-	ironcoreVolumeKey := client.ObjectKey{Namespace: s.namespace, Name: id}
-	if err := s.client.Get(ctx, ironcoreVolumeKey, ironcoreVolume); err != nil {
-		if !apierrors.IsNotFound(err) {
-			return nil, fmt.Errorf("error getting ironcore volume %s: %w", id, err)
-		}
-		return nil, status.Errorf(codes.NotFound, "volume %s not found", id)
-	}
-	if !apiutils.IsManagedBy(ironcoreVolume, volumebrokerv1alpha1.VolumeBrokerManager) || !apiutils.IsCreated(ironcoreVolume) {
-		return nil, status.Errorf(codes.NotFound, "volume %s not found", id)
-	}
-	return ironcoreVolume, nil
-}
-
 func (s *Server) getAggregateIronCoreVolume(ctx context.Context, id string) (*AggregateIronCoreVolume, error) {
 	ironcoreVolume := &storagev1alpha1.Volume{}
 	if err := s.getManagedAndCreated(ctx, id, ironcoreVolume); err != nil {
@@ -159,8 +150,8 @@ func (s *Server) getAggregateIronCoreVolume(ctx context.Context, id string) (*Ag
 	return s.aggregateIronCoreVolume(ironcoreVolume, s.clientGetSecretFunc(ctx))
 }
 
-func (s *Server) listVolumes(ctx context.Context) ([]*iri.Volume, error) {
-	ironcoreVolumes, err := s.listAggregateIronCoreVolumes(ctx)
+func (s *Server) listVolumes(ctx context.Context, filter *iri.VolumeFilter) ([]*iri.Volume, error) {
+	ironcoreVolumes, err := s.listAggregateIronCoreVolumes(ctx, filter)
 	if err != nil {
 		return nil, fmt.Errorf("error listing volumes: %w", err)
 	}
@@ -177,25 +168,6 @@ func (s *Server) listVolumes(ctx context.Context) ([]*iri.Volume, error) {
 	return res, nil
 }
 
-func (s *Server) filterVolumes(volumes []*iri.Volume, filter *iri.VolumeFilter) []*iri.Volume {
-	if filter == nil {
-		return volumes
-	}
-
-	var (
-		res []*iri.Volume
-		sel = labels.SelectorFromSet(filter.LabelSelector)
-	)
-	for _, iriVolume := range volumes {
-		if !sel.Matches(labels.Set(iriVolume.Metadata.Labels)) {
-			continue
-		}
-
-		res = append(res, iriVolume)
-	}
-	return res
-}
-
 func (s *Server) getVolume(ctx context.Context, id string) (*iri.Volume, error) {
 	ironcoreVolume, err := s.getAggregateIronCoreVolume(ctx, id)
 	if err != nil {
@@ -222,13 +194,11 @@ func (s *Server) ListVolumes(ctx context.Context, req *iri.ListVolumesRequest) (
 		}, nil
 	}
 
-	volumes, err := s.listVolumes(ctx)
+	volumes, err := s.listVolumes(ctx, req.Filter)
 	if err != nil {
 		return nil, err
 	}
 
-	volumes = s.filterVolumes(volumes, req.Filter)
-
 	return &iri.ListVolumesResponse{
 		Volumes: volumes,
 	}, nil

broker/volumebroker/server/volume_list_test.go

@@ -4,15 +4,20 @@
 package server_test
 
 import (
+	storagev1alpha1 "github.com/ironcore-dev/ironcore/api/storage/v1alpha1"
+	volumebrokerv1alpha1 "github.com/ironcore-dev/ironcore/broker/volumebroker/api/v1alpha1"
 	irimeta "github.com/ironcore-dev/ironcore/iri/apis/meta/v1alpha1"
 	iri "github.com/ironcore-dev/ironcore/iri/apis/volume/v1alpha1"
+	poolletutils "github.com/ironcore-dev/ironcore/poollet/common/utils"
 	volumepoolletv1alpha1 "github.com/ironcore-dev/ironcore/poollet/volumepoollet/api/v1alpha1"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 var _ = Describe("ListVolumes", func() {
-	_, srv := SetupTest()
+	ns, srv := SetupTest()
 	volumeClass := SetupVolumeClass()
 
 	It("should correctly list volumes", func(ctx SpecContext) {
@@ -44,4 +49,59 @@ var _ = Describe("ListVolumes", func() {
 		By("listing the Volumes")
 		Expect(srv.ListVolumes(ctx, &iri.ListVolumesRequest{})).To(HaveField("Volumes", ConsistOf(Volumes...)))
 	})
+
+	It("should set volume uid label to all existing volumes", func(ctx SpecContext) {
+		By("creating multiple volumes")
+		res, err := srv.CreateVolume(ctx, &iri.CreateVolumeRequest{
+			Volume: &iri.Volume{
+				Metadata: &irimeta.ObjectMetadata{
+					Labels: map[string]string{
+						volumepoolletv1alpha1.VolumeUIDLabel: "foobar",
+					},
+				},
+				Spec: &iri.VolumeSpec{
+					Class: volumeClass.Name,
+					Resources: &iri.VolumeResources{
+						StorageBytes: 100,
+					},
+				},
+			},
+		})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(res).NotTo(BeNil())
+
+		ironcoreVolume := &storagev1alpha1.Volume{}
+		ironcoreVolumeKey := client.ObjectKey{Namespace: ns.Name, Name: res.Volume.Metadata.Id}
+		Expect(k8sClient.Get(ctx, ironcoreVolumeKey, ironcoreVolume)).To(Succeed())
+		Expect(ironcoreVolume.Labels).To(Equal(map[string]string{
+			poolletutils.DownwardAPILabel(volumepoolletv1alpha1.VolumeDownwardAPIPrefix, "root-volume-uid"): "foobar",
+			volumebrokerv1alpha1.CreatedLabel:    "true",
+			volumebrokerv1alpha1.ManagerLabel:    volumebrokerv1alpha1.VolumeBrokerManager,
+			volumepoolletv1alpha1.VolumeUIDLabel: "foobar",
+		}))
+
+		base := ironcoreVolume.DeepCopy()
+		delete(ironcoreVolume.Labels, volumepoolletv1alpha1.VolumeUIDLabel)
+		Expect(k8sClient.Patch(ctx, ironcoreVolume, client.MergeFrom(base))).To(Succeed())
+		Expect(k8sClient.Get(ctx, ironcoreVolumeKey, ironcoreVolume)).To(Succeed())
+		Expect(ironcoreVolume.Labels).To(Equal(map[string]string{
+			poolletutils.DownwardAPILabel(volumepoolletv1alpha1.VolumeDownwardAPIPrefix, "root-volume-uid"): "foobar",
+			volumebrokerv1alpha1.CreatedLabel: "true",
+			volumebrokerv1alpha1.ManagerLabel: volumebrokerv1alpha1.VolumeBrokerManager,
+		}))
+
+		log := ctrl.LoggerFrom(ctx)
+		Expect(srv.SetVolumeUIDLabelToAllVolumes(ctx, log)).NotTo(HaveOccurred())
+
+		By("getting the ironcore volume")
+		Expect(k8sClient.Get(ctx, ironcoreVolumeKey, ironcoreVolume)).To(Succeed())
+
+		By("inspecting the ironcore volume")
+		Expect(ironcoreVolume.Labels).To(Equal(map[string]string{
+			poolletutils.DownwardAPILabel(volumepoolletv1alpha1.VolumeDownwardAPIPrefix, "root-volume-uid"): "foobar",
+			volumebrokerv1alpha1.CreatedLabel:    "true",
+			volumebrokerv1alpha1.ManagerLabel:    volumebrokerv1alpha1.VolumeBrokerManager,
+			volumepoolletv1alpha1.VolumeUIDLabel: "foobar",
+		}))
+	})
 })

poollet/volumepoollet/controllers/volume_controller.go

@@ -484,22 +484,16 @@ func (r *VolumeReconciler) reconcile(ctx context.Context, log logr.Logger, volum
 	}
 
 	log.V(1).Info("Listing volumes")
-	res, err := r.VolumeRuntime.ListVolumes(ctx, &iri.ListVolumesRequest{
-		Filter: &iri.VolumeFilter{
-			LabelSelector: map[string]string{
-				volumepoolletv1alpha1.VolumeUIDLabel: string(volume.UID),
-			},
-		},
-	})
+	volumes, err := r.listIRIVolumesByUID(ctx, volume.UID)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("error listing volumes: %w", err)
 	}
 
-	switch len(res.Volumes) {
+	switch len(volumes) {
 	case 0:
 		return r.create(ctx, log, volume)
 	case 1:
-		iriVolume := res.Volumes[0]
+		iriVolume := volumes[0]
 		if err := r.update(ctx, log, volume, iriVolume); err != nil {
 			return ctrl.Result{}, fmt.Errorf("error updating volume: %w", err)
 		}