api/handler/service_vm_hotplug.go

Rama main · réplica de lectura del repositorio original.

package handler

import (
	"context"
	"fmt"
	"strings"

	dbmodel "github.com/goodrain/rainbond/db/model"
	"github.com/goodrain/rainbond/pkg/component/k8s"
	appmvolume "github.com/goodrain/rainbond/worker/appm/volume"
	corev1 "k8s.io/api/core/v1"
	k8serrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/util/retry"
	v1 "kubevirt.io/api/core/v1"
)

var dataVolumeGVR = schema.GroupVersionResource{
	Group:    "cdi.kubevirt.io",
	Version:  "v1beta1",
	Resource: "datavolumes",
}

var vmHotplugDynamicClient = func() dynamic.Interface {
	if k8s.Default() == nil {
		return nil
	}
	return k8s.Default().DynamicClient
}

func (s *ServiceAction) hotplugVMDataDisk(tenantID string, volume *dbmodel.TenantServiceVolume) error {
	if s != nil && s.hotplugVMDataDiskHook != nil {
		return s.hotplugVMDataDiskHook(tenantID, volume)
	}
	if volume == nil {
		return nil
	}
	if volume.VolumeName == "disk" || volume.VolumeType == dbmodel.ConfigFileVolumeType.String() {
		return nil
	}

	service, err := s.getDBManager().TenantServiceDao().GetServiceByID(volume.ServiceID)
	if err != nil || service == nil || !service.IsVM() {
		return err
	}
	if resolveVMHotplugDeviceType(volume.VolumePath) != "disk" {
		return s.syncVirtualMachineSpecAfterResourceUpdate(volume.ServiceID)
	}
	vm, err := s.getVirtualMachineByServiceID(volume.ServiceID)
	if err != nil || vm == nil {
		return err
	}
	if vm.Status.PrintableStatus != v1.VirtualMachineStatusRunning {
		return s.syncVirtualMachineSpecAfterResourceUpdate(volume.ServiceID)
	}
	backingName := vmHotplugBackingName(volume)
	if err := ensureVMHotplugDataVolume(vm.Namespace, tenantID, volume, backingName); err != nil {
		return err
	}

	opts := buildVMHotplugAddVolumeOptions(backingName, volume.VolumePath)
	return s.performVMHotplugAddVolume(volume.ServiceID, opts)
}

func buildVMHotplugAddVolumeOptions(backingName, volumePath string) *v1.AddVolumeOptions {
	return &v1.AddVolumeOptions{
		Name: backingName,
		Disk: &v1.Disk{
			Serial: backingName,
			DiskDevice: v1.DiskDevice{
				Disk: &v1.DiskTarget{
					Bus: appmvolume.VMVolumeDiskBus(volumePath),
				},
			},
		},
		VolumeSource: &v1.HotplugVolumeSource{
			DataVolume: &v1.DataVolumeSource{
				Name: backingName,
			},
		},
	}
}

func buildVMHotplugRemoveVolumeOptions(backingName string) *v1.RemoveVolumeOptions {
	return &v1.RemoveVolumeOptions{Name: backingName}
}

func vmHotplugBackingName(volume *dbmodel.TenantServiceVolume) string {
	if volume == nil {
		return ""
	}
	return fmt.Sprintf("manual%d", volume.ID)
}

func (s *ServiceAction) hotunplugVMDataDisk(volume *dbmodel.TenantServiceVolume) error {
	if s != nil && s.hotunplugVMDataDiskHook != nil {
		return s.hotunplugVMDataDiskHook(volume)
	}
	if volume == nil {
		return nil
	}
	if volume.VolumeName == "disk" || volume.VolumeType == dbmodel.ConfigFileVolumeType.String() {
		return s.syncVirtualMachineSpecAfterResourceUpdate(volume.ServiceID)
	}

	service, err := s.getDBManager().TenantServiceDao().GetServiceByID(volume.ServiceID)
	if err != nil || service == nil || !service.IsVM() {
		return err
	}
	if resolveVMHotplugDeviceType(volume.VolumePath) != "disk" {
		if err := s.syncVirtualMachineSpecAfterResourceUpdate(volume.ServiceID); err != nil {
			return err
		}
		return s.cleanupVMHotplugBackingStorage(volume.ServiceID, vmHotplugBackingName(volume))
	}

	opts := buildVMHotplugRemoveVolumeOptions(vmHotplugBackingName(volume))
	if err := s.performVMHotplugRemoveVolume(volume.ServiceID, opts); err != nil {
		return err
	}
	return s.cleanupVMHotplugBackingStorage(volume.ServiceID, opts.Name)
}

func (s *ServiceAction) performVMHotplugAddVolume(serviceID string, opts *v1.AddVolumeOptions) error {
	return s.retryVMHotplugVolumeRequest(serviceID, func(vm *v1.VirtualMachine) error {
		if vm == nil || vm.Status.PrintableStatus != v1.VirtualMachineStatusRunning {
			return s.syncVirtualMachineSpecAfterResourceUpdate(serviceID)
		}
		return s.kubevirtClient.VirtualMachine(vm.Namespace).AddVolume(context.Background(), vm.Name, opts)
	})
}

func (s *ServiceAction) performVMHotplugRemoveVolume(serviceID string, opts *v1.RemoveVolumeOptions) error {
	return s.retryVMHotplugVolumeRequest(serviceID, func(vm *v1.VirtualMachine) error {
		if vm == nil || vm.Status.PrintableStatus != v1.VirtualMachineStatusRunning {
			return s.syncVirtualMachineSpecAfterResourceUpdate(serviceID)
		}
		return s.kubevirtClient.VirtualMachine(vm.Namespace).RemoveVolume(context.Background(), vm.Name, opts)
	})
}

func (s *ServiceAction) retryVMHotplugVolumeRequest(serviceID string, request func(vm *v1.VirtualMachine) error) error {
	return retry.RetryOnConflict(retry.DefaultRetry, func() error {
		vm, err := s.getVirtualMachineByServiceID(serviceID)
		if err != nil {
			return err
		}
		return request(vm)
	})
}

func (s *ServiceAction) cleanupVMHotplugBackingStorage(serviceID, backingName string) error {
	if strings.TrimSpace(backingName) == "" {
		return nil
	}
	vm, err := s.getVirtualMachineByServiceID(serviceID)
	if err != nil || vm == nil {
		return err
	}
	return s.deleteVMHotplugBackingStorage(vm.Namespace, backingName)
}

func (s *ServiceAction) deleteVMHotplugBackingStorage(namespace, backingName string) error {
	if strings.TrimSpace(namespace) == "" || strings.TrimSpace(backingName) == "" {
		return nil
	}
	if dynamicClient := vmHotplugDynamicClient(); dynamicClient != nil {
		err := dynamicClient.Resource(dataVolumeGVR).Namespace(namespace).Delete(context.Background(), backingName, metav1.DeleteOptions{})
		if err != nil && !k8serrors.IsNotFound(err) {
			return fmt.Errorf("delete vm backing datavolume %s/%s: %w", namespace, backingName, err)
		}
	}
	if s.kubeClient != nil {
		err := s.kubeClient.CoreV1().PersistentVolumeClaims(namespace).Delete(context.Background(), backingName, metav1.DeleteOptions{})
		if err != nil && !k8serrors.IsNotFound(err) {
			return fmt.Errorf("delete vm backing pvc %s/%s: %w", namespace, backingName, err)
		}
	}
	return nil
}

func ensureVMHotplugDataVolume(namespace, tenantID string, volume *dbmodel.TenantServiceVolume, backingName string) error {
	dynamicClient := vmHotplugDynamicClient()
	if dynamicClient == nil {
		return fmt.Errorf("dynamic client is required for vm hotplug volume")
	}
	resourceIf := dynamicClient.Resource(dataVolumeGVR).Namespace(namespace)
	if _, err := resourceIf.Get(context.Background(), backingName, metav1.GetOptions{}); err == nil {
		return nil
	}

	obj := buildVMHotplugDataVolumeObject(namespace, tenantID, volume, backingName)
	_, err := resourceIf.Create(context.Background(), obj, metav1.CreateOptions{})
	return err
}

func buildVMHotplugDataVolumeObject(namespace, tenantID string, volume *dbmodel.TenantServiceVolume, backingName string) *unstructured.Unstructured {
	storageQty := resource.MustParse(fmt.Sprintf("%dGi", volume.VolumeCapacity))
	accessModes := resolveVMHotplugAccessModes(volume)
	accessModeValues := make([]any, 0, len(accessModes))
	for _, mode := range accessModes {
		accessModeValues = append(accessModeValues, string(mode))
	}

	return &unstructured.Unstructured{
		Object: map[string]any{
			"apiVersion": "cdi.kubevirt.io/v1beta1",
			"kind":       "DataVolume",
			"metadata": map[string]any{
				"name":      backingName,
				"namespace": namespace,
				"labels": map[string]any{
					"service_id":  volume.ServiceID,
					"tenant_id":   tenantID,
					"volume_name": backingName,
					"stateless":   "",
				},
				"annotations": map[string]any{
					"volume_name": volume.VolumeName,
				},
			},
			"spec": map[string]any{
				"source": map[string]any{
					"blank": map[string]any{},
				},
				"storage": map[string]any{
					"accessModes": accessModeValues,
					"resources": map[string]any{
						"requests": map[string]any{
							"storage": storageQty.String(),
						},
					},
					"storageClassName": resolveVMHotplugStorageClassName(volume),
					"volumeMode":       string(corev1.PersistentVolumeFilesystem),
				},
			},
		},
	}
}

func resolveVMHotplugAccessModes(volume *dbmodel.TenantServiceVolume) []corev1.PersistentVolumeAccessMode {
	defaultModes := []corev1.PersistentVolumeAccessMode{corev1.ReadWriteMany}
	if volume == nil {
		return defaultModes
	}

	seen := make(map[corev1.PersistentVolumeAccessMode]struct{})
	modes := make([]corev1.PersistentVolumeAccessMode, 0, 1)
	for _, item := range strings.Split(volume.AccessMode, ",") {
		mode, ok := normalizeVMHotplugAccessMode(item)
		if !ok {
			continue
		}
		if _, exists := seen[mode]; exists {
			continue
		}
		seen[mode] = struct{}{}
		modes = append(modes, mode)
	}
	if len(modes) == 0 {
		return defaultModes
	}
	return modes
}

func normalizeVMHotplugAccessMode(mode string) (corev1.PersistentVolumeAccessMode, bool) {
	switch strings.ToUpper(strings.TrimSpace(mode)) {
	case "RWO", "READWRITEONCE":
		return corev1.ReadWriteOnce, true
	case "ROX", "READONLYMANY":
		return corev1.ReadOnlyMany, true
	case "RWX", "READWRITEMANY":
		return corev1.ReadWriteMany, true
	case "READWRITEONCEPOD":
		return corev1.ReadWriteOncePod, true
	default:
		return "", false
	}
}

func resolveVMHotplugDeviceType(volumePath string) string {
	return appmvolume.VMVolumeDeviceType(volumePath)
}

func resolveVMHotplugStorageClassName(volume *dbmodel.TenantServiceVolume) string {
	if volume == nil || volume.VolumeType == "" || volume.VolumeType == dbmodel.VMVolumeType.String() {
		return "local-path"
	}
	return volume.VolumeType
}