KubeVirt 数据卷热插实现

$ kubectl get po -n demo
NAME                            READY   STATUS    RESTARTS   AGE
virt-launcher-ecs-test1-n4pl8   1/1     Running   0          6m21s

$ kubectl exec -it virt-launcher-ecs-test1-n4pl8 -n demo -- virsh dumpxml 1
  <devices>
    <disk type='block' device='disk'>
      <driver name='qemu' type='raw' error_policy='stop' discard='unmap'/>
      <source dev='/var/run/kubevirt/hotplug-disks/datavol-hotplug' index='4'/>
      <backingStore/>
      <target dev='sda' bus='scsi'/>
      <serial>datavol-hotplug</serial>
      <alias name='ua-datavol-hotplug'/>
      <address type='drive' controller='0' bus='0' target='0' unit='0'/>
    </disk>
  </devices>

$ kubectl exec -it virt-launcher-ecs-test1-n4pl8 -n demo -- ls -al /var/run/kubevirt/hotplug-disks/datavol-hotplug
brw-rw---- 1 qemu qemu 252, 240 Jul  4 07:29 /var/run/kubevirt/hotplug-disks/datavol-hotplug

$ kubectl get po virt-launcher-ecs-test1-n4pl8 -n demo -o yaml | grep datavol-hotplug

$ kubectl get vm ecs-test1 -n demo -ojsonpath='{.spec.template.spec.volumes}' | jq
[
  {
    "name": "bootdisk",
    "persistentVolumeClaim": {
      "claimName": "ecs-test1-bootpvc-bgzrlh"
    }
  },
  {
    "cloudInitNoCloud": {
      "userData": "#cloud-config\npassword: atomic\nssh_pwauth: True\nchpasswd: { expire: False }\nbootcmd:\n- \"setenforce 0\""
    },
    "name": "cloudinitdisk"
  },
  {
    "name": "datavol-hotplug",
    "persistentVolumeClaim": {
      "claimName": "datavol-hotplug",
      "hotpluggable": true
    }
  }
]


$ kubectl get vmi ecs-test1 -n demo -ojsonpath='{.spec.volumes}' | jq
[
  {
    "name": "bootdisk",
    "persistentVolumeClaim": {
      "claimName": "ecs-test1-bootpvc-bgzrlh"
    }
  },
  {
    "cloudInitNoCloud": {
      "userData": "#cloud-config\npassword: atomic\nssh_pwauth: True\nchpasswd: { expire: False }\nbootcmd:\n- \"setenforce 0\""
    },
    "name": "cloudinitdisk"
  },
  {
    "name": "datavol-hotplug",
    "persistentVolumeClaim": {
      "claimName": "datavol-hotplug",
      "hotpluggable": true
    }
  }
]

$ kubectl get po -n demo
NAME                            READY   STATUS    RESTARTS   AGE
hp-volume-kdpgl                 1/1     Running   0          17m
virt-launcher-ecs-test1-n4pl8   1/1     Running   0          20m

$ kubectl get po hp-volume-kdpgl -n demo -o yaml
apiVersion: v1
kind: Pod
metadata:
  ownerReferences:
  - apiVersion: v1
    blockOwnerDeletion: true
    controller: true
    kind: Pod
    name: virt-launcher-ecs-test1-n4pl8
    uid: 7bf1ee01-f81f-4406-a500-b351a6115dd9
  resourceVersion: "70797056"
  uid: 90f44088-5dd0-4d4d-9ca4-d9fee4ef736a
spec:
  affinity:
    nodeAffinity:
      requiredDuringSchedulingIgnoredDuringExecution:
        nodeSelectorTerms:
        - matchExpressions:
          - key: kubernetes.io/hostname
            operator: In
            values:
            - bj-node171
  containers:
  - command:
    - /bin/sh
    - -c
    - /usr/bin/container-disk --copy-path /path/hp
    image: registry-1.ict-mec.net:18443/kubevirt/virt-launcher:v0.51.0-dev
    name: hotplug-disk
    volumeDevices:
    - devicePath: /path/datavol-hotplug/368836ba-6f6a-4d07-b7ad-914b856642f6
      name: datavol-hotplug
    volumeMounts:
    - mountPath: /path
      mountPropagation: HostToContainer
      name: hotplug-disks
    - mountPath: /var/run/secrets/kubernetes.io/serviceaccount
      name: kube-api-access-drrz6
      readOnly: true
  volumes:
  - emptyDir: {}
    name: hotplug-disks
  - name: datavol-hotplug
    persistentVolumeClaim:
      claimName: datavol-hotplug

func (c *VMIController) createAttachmentPodTemplate(vmi *virtv1.VirtualMachineInstance, virtlauncherPod *k8sv1.Pod, volumes []*virtv1.Volume) (*k8sv1.Pod, error) {
    logger := log.Log.Object(vmi)
    var pod *k8sv1.Pod
    var err error

    volumeNamesPVCMap, err := kubevirttypes.VirtVolumesToPVCMap(volumes, c.pvcInformer.GetStore(), virtlauncherPod.Namespace)
    if err != nil {
        return nil, fmt.Errorf("failed to get PVC map: %v", err)
    }
    // a lot of code here
    if len(volumeNamesPVCMap) > 0 {
        pod, err = c.templateService.RenderHotplugAttachmentPodTemplate(volumes, virtlauncherPod, vmi, volumeNamesPVCMap, false)
    }
    return pod, err
}

func (c *VMIController) createAttachmentPod(vmi *virtv1.VirtualMachineInstance, virtLauncherPod *k8sv1.Pod, volumes []*virtv1.Volume) syncError {
    attachmentPodTemplate, _ := c.createAttachmentPodTemplate(vmi, virtLauncherPod, volumes)
    if attachmentPodTemplate == nil {
        return nil
    }
    vmiKey := controller.VirtualMachineInstanceKey(vmi)
    c.podExpectations.ExpectCreations(vmiKey, 1)

    pod, err := c.clientset.CoreV1().Pods(vmi.GetNamespace()).Create(context.Background(), attachmentPodTemplate, v1.CreateOptions{})
    if err != nil {
        c.podExpectations.CreationObserved(vmiKey)
        c.recorder.Eventf(vmi, k8sv1.EventTypeWarning, FailedCreatePodReason, "Error creating attachment pod: %v", err)
        return &syncErrorImpl{fmt.Errorf("Error creating attachment pod %v", err), FailedCreatePodReason}
    }
    c.recorder.Eventf(vmi, k8sv1.EventTypeNormal, SuccessfulCreatePodReason, "Created attachment pod %s", pod.Name)
    return nil
}

$ kubectl exec -it hp-volume-kdpgl -n demo -- ls -al /path
total 0
drwxrwxrwx 3 root root 44 Jul  6 03:06 .
drwxr-xr-x 1 root root 63 Jul  6 03:06 ..
drwxr-xr-x 2 root root 50 Jul  6 03:06 datavol-hotplug
srwxr-xr-x 1 root root  0 Jul  6 03:06 hp.sock

$ ls -al /var/lib/kubelet/pods/$(kubectl get po hp-volume-kdpgl -n demo -o jsonpath='{.metadata.uid}')/volumes/kubernetes.io~empty-dir/hotplug-disks/hp.sock
srwxr-xr-x 1 root root 0 Jul  6 11:06 /var/lib/kubelet/pods/90f44088-5dd0-4d4d-9ca4-d9fee4ef736a/volumes/kubernetes.io~empty-dir/hotplug-disks/hp.sock

func (d *VirtualMachineController) processVmUpdate(vmi *v1.VirtualMachineInstance, domainExists bool) error {
    isUnresponsive, isInitialized, err := d.isLauncherClientUnresponsive(vmi)
    if err != nil {
        return err
    }
    if !isInitialized {
        d.Queue.AddAfter(controller.VirtualMachineInstanceKey(vmi), time.Second*1)
        return nil
    } else if isUnresponsive {
        return goerror.New(fmt.Sprintf("Can not update a VirtualMachineInstance with unresponsive command server."))
    }

    d.handlePostMigrationProxyCleanup(vmi)

    if d.isPreMigrationTarget(vmi) {
        return d.vmUpdateHelperMigrationTarget(vmi)
    } else if d.isMigrationSource(vmi) {
        return d.vmUpdateHelperMigrationSource(vmi)
    } else {
        return d.vmUpdateHelperDefault(vmi, domainExists)
    }
}

func (d *VirtualMachineController) vmUpdateHelperDefault(origVMI *v1.VirtualMachineInstance, domainExists bool) error {
    // a lot of code here
        // Try to mount hotplug volume if there is any during startup.
        if err := d.hotplugVolumeMounter.Mount(vmi); err != nil {
            return err
        }
}

func (m *volumeMounter) Mount(vmi *v1.VirtualMachineInstance) error {
    record, err := m.getMountTargetRecord(vmi)
    if err != nil {
        return err
    }
    for _, volumeStatus := range vmi.Status.VolumeStatus {
        if volumeStatus.HotplugVolume == nil {
            // Skip non hotplug volumes
            continue
        }
        sourceUID := volumeStatus.HotplugVolume.AttachPodUID
        if err := m.mountHotplugVolume(vmi, volumeStatus.Name, sourceUID, record); err != nil {
            return err
        }
    }
    return nil
}

func (m *volumeMounter) mountHotplugVolume(vmi *v1.VirtualMachineInstance, volumeName string, sourceUID types.UID, record *vmiMountTargetRecord) error {
    logger := log.DefaultLogger()
    logger.V(4).Infof("Hotplug check volume name: %s", volumeName)
    if sourceUID != types.UID("") {
        if m.isBlockVolume(&vmi.Status, volumeName) {
            logger.V(4).Infof("Mounting block volume: %s", volumeName)
            if err := m.mountBlockHotplugVolume(vmi, volumeName, sourceUID, record); err != nil {
                return err
            }
        }
    }
    return nil
}

func (m *volumeMounter) mountBlockHotplugVolume(vmi *v1.VirtualMachineInstance, volume string, sourceUID types.UID, record *vmiMountTargetRecord) error {
    virtlauncherUID := m.findVirtlauncherUID(vmi)
    if virtlauncherUID == "" {
        // This is not the node the pod is running on.
        return nil
    }
    targetPath, err := m.hotplugDiskManager.GetHotplugTargetPodPathOnHost(virtlauncherUID)
    if err != nil {
        return err
    }
    // a lot of core here
    if isBlockExists, _ := isBlockDevice(deviceName); !isBlockExists {
        sourceMajor, sourceMinor, permissions, err := m.getSourceMajorMinor(sourceUID, volume)
        if err != nil {
            return err
        }
        if err := m.writePathToMountRecord(deviceName, vmi, record); err != nil {
            return err
        }
        // allow block devices
        if err := m.allowBlockMajorMinor(sourceMajor, sourceMinor, cgroupsManager); err != nil {
            return err
        }
        if _, err = m.createBlockDeviceFile(deviceName, sourceMajor, sourceMinor, permissions); err != nil {
            return err
        }
    }
}

var mknodCommand = func(deviceName string, major, minor int64, blockDevicePermissions string) ([]byte, error) {
    output, err := exec.Command("/usr/bin/mknod", "--mode", fmt.Sprintf("0%s", blockDevicePermissions), deviceName, "b", strconv.FormatInt(major, 10), strconv.FormatInt(minor, 10)).CombinedOutput()
    log.Log.V(3).Infof("running mknod. err: %v, output: %s", err, string(output))
    return output, err
}

func (m *volumeMounter) createBlockDeviceFile(deviceName string, major, minor int64, blockDevicePermissions string) (string, error) {
    exists, err := diskutils.FileExists(deviceName)
    if err != nil {
        return "", err
    }
    if !exists {
        out, err := mknodCommand(deviceName, major, minor, blockDevicePermissions)
        if err != nil {
            log.DefaultLogger().Errorf("Error creating block device file: %s, %v", out, err)
            return "", err
        }
    }
    return deviceName, nil
}

$ kubectl exec -it hp-volume-kdpgl -n demo -- ls -al /path
total 0
drwxr-xr-x 2 root root 50 Jul  6 03:06 datavol-hotplug
srwxr-xr-x 1 root root  0 Jul  6 03:06 hp.sock

var socketPath = func(podUID types.UID) string {
    return fmt.Sprintf("pods/%s/volumes/kubernetes.io~empty-dir/hotplug-disks/hp.sock", string(podUID))
}

func (m *volumeMounter) getSourcePodFilePath(sourceUID types.UID, vmi *v1.VirtualMachineInstance, volume string) (string, error) {
    iso := isolationDetector("/path")
    isoRes, err := iso.DetectForSocket(vmi, socketPath(sourceUID))
    if err != nil {
        return "", err
    }
    findmounts, err := LookupFindmntInfoByVolume(volume, isoRes.Pid())
    if err != nil {
        return "", err
    }
    // a lot of code here
    // Did not find the disk image file, return error
    return "", fmt.Errorf("unable to find source disk image path for pod %s", sourceUID)
}

func (s *socketBasedIsolationDetector) DetectForSocket(vm *v1.VirtualMachineInstance, socket string) (IsolationResult, error) {
    var pid int
    var ppid int
    var err error

    if pid, err = s.getPid(socket); err != nil {
        log.Log.Object(vm).Reason(err).Errorf("Could not get owner Pid of socket %s", socket)
        return nil, err
    }

    // a lot of code here
}

func (s *socketBasedIsolationDetector) getPid(socket string) (int, error) {
    sock, err := net.DialTimeout("unix", socket, time.Duration(isolationDialTimeout)*time.Second)
    if err != nil {
        return -1, err
    }
    defer sock.Close()

    ufile, err := sock.(*net.UnixConn).File()
    if err != nil {
        return -1, err
    }
    // This is the tricky part, which will give us the PID of the owning socket
    ucreds, err := syscall.GetsockoptUcred(int(ufile.Fd()), syscall.SOL_SOCKET, syscall.SO_PEERCRED)
    if err != nil {
        return -1, err
    }

    if int(ucreds.Pid) == 0 {
        return -1, fmt.Errorf("the detected PID is 0. Is the isolation detector running in the host PID namespace?")
    }

    return int(ucreds.Pid), nil
}