refactor volume resize related changes

Author: ushabelgur

api/machine.go

@@ -84,12 +84,12 @@ type EmptyDiskSpec struct {
 }
 
 type VolumeConnection struct {
-	Driver                string            ` json:"driver,omitempty"`
-	Handle                string            ` json:"handle,omitempty"`
-	Attributes            map[string]string ` json:"attributes,omitempty"`
-	SecretData            map[string][]byte ` json:"secret_data,omitempty"`
-	EncryptionData        map[string][]byte ` json:"encryption_data,omitempty"`
-	EffectiveStorageBytes int64             ` json:"effective_storage_bytes,omitempty"`
+	Driver                string            `json:"driver,omitempty"`
+	Handle                string            `json:"handle,omitempty"`
+	Attributes            map[string]string `json:"attributes,omitempty"`
+	SecretData            map[string][]byte `json:"secret_data,omitempty"`
+	EncryptionData        map[string][]byte `json:"encryption_data,omitempty"`
+	EffectiveStorageBytes int64             `json:"effective_storage_bytes,omitempty"`
 }
 
 type VolumeState string

internal/controllers/machine_controller_volumes.go

@@ -26,6 +26,8 @@ import (
 	"libvirt.org/go/libvirtxml"
 )
 
+var ErrNotAssignedDomain = errors.New("not assigned to domain")
+
 func (r *MachineReconciler) deleteVolumes(ctx context.Context, log logr.Logger, machine *api.Machine) error {
 	mounter := r.machineVolumeMounter(machine)
 	var errs []error
@@ -66,20 +68,38 @@ func (r *MachineReconciler) machineVolumeMounter(machine *api.Machine) VolumeMou
 	}
 }
 
-func getVolumeStatus(machine *api.Machine, volumeID string) *api.VolumeStatus {
-	for _, volumeStatus := range machine.Status.VolumeStatus {
-		if volumeID == volumeStatus.Handle {
-			return &volumeStatus
+func (r *MachineReconciler) getLastVolumeSize(machineID, deviceName string) (int64, error) {
+	domain, err := r.host.Libvirt().DomainLookupByUUID(libvirtutils.UUIDStringToBytes(machineID))
+	if err != nil {
+		if !libvirt.IsNotFound(err) {
+			return 0, fmt.Errorf("error getting blockInfo from domain for machine %s device %s: %w", machineID, deviceName, err)
 		}
+		return 0, nil
 	}
-	return nil
+
+	virtDeviceName := computeVirtioDiskTargetDeviceName(deviceName)
+	_, capacity, _, err := r.host.Libvirt().DomainGetBlockInfo(domain, virtDeviceName, 0)
+
+	if err != nil {
+		if IsNotAssignedDomainErr(err) {
+			return 0, nil
+		}
+		return 0, fmt.Errorf("error getting blockInfo from domain for machine %s device %s: %w", machineID, deviceName, err)
+	}
+
+	return int64(capacity), nil
 }
 
-func getLastVolumeSize(machine *api.Machine, volumeID string) int64 {
-	if status := getVolumeStatus(machine, volumeID); status != nil && status.Size != 0 {
-		return status.Size
+func IsNotAssignedDomainErr(err error) bool {
+	var lvErr libvirt.Error
+	if !errors.As(err, &lvErr) {
+		return false
 	}
-	return 0
+	if lvErr.Code == uint32(libvirt.ErrInvalidArg) && strings.Contains(err.Error(), ErrNotAssignedDomain.Error()) {
+		return true
+	}
+
+	return false
 }
 
 func (r *MachineReconciler) attachDetachVolumes(ctx context.Context, log logr.Logger, machine *api.Machine, attacher VolumeAttacher) ([]api.VolumeStatus, error) {
@@ -635,6 +655,18 @@ func (r *MachineReconciler) applyVolume(
 		return "", 0, fmt.Errorf("error applying volume mount: %w", err)
 	}
 
+	lastVolumeSize, err := r.getLastVolumeSize(machine.ID, desiredVolume.Device)
+	if err != nil {
+		return volumeID, 0, fmt.Errorf("error getting last volume size: %w", err)
+	}
+
+	var volumeSize int64
+	if providerVolume != nil {
+		volumeSize = providerVolume.EffectiveStorageBytesSize
+	} else {
+		volumeSize = lastVolumeSize
+	}
+
 	log.V(1).Info("Ensuring volume is attached")
 	if err := attacher.AttachVolume(&AttachVolume{
 		Name:   desiredVolume.Name,
@@ -645,8 +677,8 @@ func (r *MachineReconciler) applyVolume(
 	}
 
 	//TODO do epsilon comparison
-	if lastVolumeSize := getLastVolumeSize(machine, volumeID); lastVolumeSize != 0 && providerVolume.EffectiveStorageBytesSize != lastVolumeSize {
-		log.V(1).Info("Resize volume", "volumeID", volumeID, "lastSize", lastVolumeSize, "volumeSize", providerVolume.EffectiveStorageBytesSize)
+	if lastVolumeSize != 0 && volumeSize != lastVolumeSize {
+		log.V(1).Info("Resize volume", "volumeID", volumeID, "lastSize", lastVolumeSize, "volumeSize", volumeSize)
 		if err := attacher.ResizeVolume(&AttachVolume{
 			Name:   desiredVolume.Name,
 			Device: desiredVolume.Device,

internal/plugins/volume/ceph/ceph.go

@@ -174,6 +174,9 @@ func (p *plugin) getVolumeData(spec *api.VolumeSpec) (vData *volumeData, err err
 	if connection.Handle == "" {
 		return nil, fmt.Errorf("volume connection does not specify handle")
 	}
+	if connection.EffectiveStorageBytes <= 0 {
+		return nil, fmt.Errorf("volume connection does not specify effective storage bytes")
+	}
 	vData.handle = connection.Handle
 
 	vData.monitors, vData.image, err = readVolumeAttributes(connection.Attributes)
@@ -193,9 +196,8 @@ func (p *plugin) getVolumeData(spec *api.VolumeSpec) (vData *volumeData, err err
 		}
 	}
 
-	if effectiveStorageBytes := spec.Connection.EffectiveStorageBytes; effectiveStorageBytes != 0 {
-		vData.effectiveStorageBytes = effectiveStorageBytes
-	}
+	vData.effectiveStorageBytes = spec.Connection.EffectiveStorageBytes
+
 	return vData, nil
 }
 

internal/server/machine_volume_attach_test.go

@@ -12,6 +12,7 @@ import (
 	libvirtutils "github.com/ironcore-dev/libvirt-provider/internal/libvirt/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"libvirt.org/go/libvirtxml"
 )
 
@@ -115,6 +116,7 @@ var _ = Describe("AttachVolume", func() {
 						"userID":  []byte(cephUsername),
 						"userKey": []byte(cephUserkey),
 					},
+					EffectiveStorageBytes: resource.NewQuantity(1*1024*1024*1024, resource.BinarySI).Value(),
 				},
 			},
 		})

internal/server/machine_volume_detach_test.go

@@ -12,6 +12,7 @@ import (
 	libvirtutils "github.com/ironcore-dev/libvirt-provider/internal/libvirt/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"libvirt.org/go/libvirtxml"
 )
 
@@ -60,6 +61,7 @@ var _ = Describe("DetachVolume", func() {
 									"userID":  []byte(cephUsername),
 									"userKey": []byte(cephUserkey),
 								},
+								EffectiveStorageBytes: resource.NewQuantity(1*1024*1024*1024, resource.BinarySI).Value(),
 							},
 						},
 					},

internal/server/machine_volume_update.go

@@ -20,20 +20,19 @@ func (s *Server) UpdateVolume(ctx context.Context, req *iri.UpdateVolumeRequest)
 		return nil, convertInternalErrorToGRPC(ErrInvalidRequest)
 	}
 
-	iriVolume := req.Volume
 	apiMachine, err := s.machineStore.Get(ctx, req.MachineId)
 	if err != nil {
 		return nil, convertInternalErrorToGRPC(fmt.Errorf("failed to get machine '%s': %w", req.MachineId, err))
 	}
 
 	apiVolumeIndex := apiMachineVolumeIndex(apiMachine, req.Volume.Name)
-	if apiVolumeIndex < 0 {
+	if apiVolumeIndex == -1 {
 		return nil, convertInternalErrorToGRPC(fmt.Errorf("volume '%s' not found in machine '%s': %w", req.Volume.Name, req.MachineId, ErrVolumeNotFound))
 	}
 
 	apiBaseVolume := apiMachine.Spec.Volumes[apiVolumeIndex]
-	apiBaseVolume.Device = iriVolume.Device
-	if volumeConnection := iriVolume.Connection; volumeConnection != nil {
+	apiBaseVolume.Device = req.Volume.Device
+	if volumeConnection := req.Volume.Connection; volumeConnection != nil {
 		apiBaseVolume.Connection = &api.VolumeConnection{
 			Driver:                volumeConnection.Driver,
 			Handle:                volumeConnection.Handle,

internal/server/machine_volume_update_test.go

@@ -119,9 +119,6 @@ var _ = Describe("UpdateVolume", func() {
 		}).Should(Equal(2))
 		Expect(disks[0].Serial).To(HavePrefix("oda"))
 
-		// wait to complete machine reconciliation
-		time.Sleep(20 * time.Second)
-
 		By("updating machine volume")
 		updateVolumeResp, err := machineClient.UpdateVolume(ctx, &iri.UpdateVolumeRequest{
 			MachineId: createResp.Machine.Metadata.Id,
@@ -146,9 +143,6 @@ var _ = Describe("UpdateVolume", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(updateVolumeResp).NotTo(BeNil())
 
-		// wait to complete machine reconciliation
-		time.Sleep(20 * time.Second)
-
 		By("ensuring volume has been resized and updated in machine spec field")
 		Eventually(func(g Gomega) *iri.Volume {
 			listResp, err := machineClient.ListMachines(ctx, &iri.ListMachinesRequest{