Support to specify block mode volumes in kube.PodOptions. (#1756)

* Added support to specify block mode volumes in kube.PodOptions.

* Added UT to test mounting block volumes in a pod.

pkg/kube/job.go

@@ -73,7 +73,7 @@ func NewJob(clientset kubernetes.Interface, jobName string, namespace string, se
 // Create creates the Job in Kubernetes.
 func (job *Job) Create() error {
 	falseVal := false
-	volumeMounts, podVolumes, err := createVolumeSpecs(job.vols)
+	volumeMounts, podVolumes, err := createFilesystemModeVolumeSpecs(job.vols)
 	if err != nil {
 		return errors.Wrapf(err, "Failed to create volume spec for job %s", job.name)
 	}
@@ -125,8 +125,8 @@ func (job *Job) Create() error {
 	return nil
 }
 
-func createVolumeSpecs(vols map[string]string) (volumeMounts []v1.VolumeMount, podVolumes []v1.Volume, error error) {
-	// Build volume specs
+func createFilesystemModeVolumeSpecs(vols map[string]string) (volumeMounts []v1.VolumeMount, podVolumes []v1.Volume, error error) {
+	// Build filesystem mode volume specs
 	for pvc, mountPath := range vols {
 		id, err := uuid.NewV1()
 		if err != nil {
@@ -148,6 +148,29 @@ func createVolumeSpecs(vols map[string]string) (volumeMounts []v1.VolumeMount, p
 	return volumeMounts, podVolumes, nil
 }
 
+func createBlockModeVolumeSpecs(blockVols map[string]string) (volumeDevices []v1.VolumeDevice, podVolumes []v1.Volume, error error) {
+	// Build block mode volume specs
+	for pvc, devicePath := range blockVols {
+		id, err := uuid.NewV1()
+		if err != nil {
+			return nil, nil, err
+		}
+		podBlockVolName := fmt.Sprintf("block-%s", id.String())
+		volumeDevices = append(volumeDevices, v1.VolumeDevice{Name: podBlockVolName, DevicePath: devicePath})
+		podVolumes = append(podVolumes,
+			v1.Volume{
+				Name: podBlockVolName,
+				VolumeSource: v1.VolumeSource{
+					PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+						ClaimName: pvc,
+					},
+				},
+			},
+		)
+	}
+	return volumeDevices, podVolumes, nil
+}
+
 // WaitForCompletion waits for the job to run to completion.
 func (job *Job) WaitForCompletion(ctx context.Context) error {
 	batchClient := job.clientset.BatchV1()

pkg/kube/pod.go

@@ -58,6 +58,7 @@ type PodOptions struct {
 	Namespace          string
 	ServiceAccountName string
 	Volumes            map[string]string
+	BlockVolumes       map[string]string
 	// PodSecurityContext and ContainerSecurityContext can be used to set the security context
 	// at the pod level and container level respectively.
 	// You can still use podOverride to set the pod security context, but these fields will take precedence.
@@ -97,10 +98,15 @@ func GetPodObjectFromPodOptions(cli kubernetes.Interface, opts *PodOptions) (*v1
 		opts.RestartPolicy = v1.RestartPolicyNever
 	}
 
-	volumeMounts, podVolumes, err := createVolumeSpecs(opts.Volumes)
+	volumeMounts, podVolumes, err := createFilesystemModeVolumeSpecs(opts.Volumes)
 	if err != nil {
 		return nil, errors.Wrapf(err, "Failed to create volume spec")
 	}
+	volumeDevices, blockVolumes, err := createBlockModeVolumeSpecs(opts.BlockVolumes)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Failed to create raw block volume spec")
+	}
+	podVolumes = append(podVolumes, blockVolumes...)
 	defaultSpecs := v1.PodSpec{
 		Containers: []v1.Container{
 			{
@@ -109,6 +115,7 @@ func GetPodObjectFromPodOptions(cli kubernetes.Interface, opts *PodOptions) (*v1
 				Command:         opts.Command,
 				ImagePullPolicy: v1.PullPolicy(v1.PullIfNotPresent),
 				VolumeMounts:    volumeMounts,
+				VolumeDevices:   volumeDevices,
 				Resources:       opts.Resources,
 			},
 		},

pkg/kube/pod_test.go

@@ -267,7 +267,7 @@ func (s *PodSuite) createServiceAccount(name, ns string) error {
 	return nil
 }
 
-func (s *PodSuite) TestPodWithVolumes(c *C) {
+func (s *PodSuite) TestPodWithFilesystemModeVolumes(c *C) {
 	cli := fake.NewSimpleClientset()
 	pvcName := "prometheus-ibm-monitoring-prometheus-db-prometheus-ibm-monitoring-prometheus-0"
 	pvc := &v1.PersistentVolumeClaim{
@@ -313,6 +313,58 @@ func (s *PodSuite) TestPodWithVolumes(c *C) {
 	c.Assert(pod.Spec.Volumes, HasLen, 1)
 	c.Assert(pod.Spec.Volumes[0].VolumeSource.PersistentVolumeClaim.ClaimName, Equals, pvcName)
 	c.Assert(pod.Spec.Containers[0].VolumeMounts[0].MountPath, Equals, "/mnt/data1")
+	c.Assert(len(pod.Spec.Containers[0].VolumeDevices), Equals, 0)
+}
+
+func (s *PodSuite) TestPodWithBlockModeVolumes(c *C) {
+	cli := fake.NewSimpleClientset()
+	pvcName := "block-mode-volume"
+	blockMode := v1.PersistentVolumeBlock
+	pvc := &v1.PersistentVolumeClaim{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: pvcName,
+		},
+		Spec: v1.PersistentVolumeClaimSpec{
+			AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+			VolumeMode:  &blockMode,
+			Resources: v1.ResourceRequirements{
+				Requests: v1.ResourceList{
+					v1.ResourceName(v1.ResourceStorage): resource.MustParse("1Gi"),
+				},
+			},
+		},
+	}
+	pvc, err := cli.CoreV1().PersistentVolumeClaims(s.namespace).Create(context.TODO(), pvc, metav1.CreateOptions{})
+	c.Assert(err, IsNil)
+	vols := map[string]string{pvc.Name: "/mnt/data1"}
+	ctx := context.Background()
+	var p *v1.Pod
+	cli.PrependReactor("create", "pods", func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+		fmt.Println("found pod")
+		ca := action.(testing.CreateAction)
+		p = ca.GetObject().(*v1.Pod)
+		if len(p.Spec.Volumes[0].Name) > 63 {
+			return true, nil, errors.New("spec.volumes[0].name must be no more than 63 characters")
+		}
+		return false, nil, nil
+	})
+	cli.PrependReactor("get", "pods", func(action testing.Action) (handled bool, ret runtime.Object, err error) {
+		p.Status.Phase = v1.PodRunning
+		return true, p, nil
+	})
+	pod, err := CreatePod(ctx, cli, &PodOptions{
+		Namespace:    s.namespace,
+		GenerateName: "test-",
+		Image:        consts.LatestKanisterToolsImage,
+		Command:      []string{"sh", "-c", "tail -f /dev/null"},
+		BlockVolumes: vols,
+	})
+	c.Assert(err, IsNil)
+	c.Assert(WaitForPodReady(ctx, cli, s.namespace, pod.Name), IsNil)
+	c.Assert(pod.Spec.Volumes, HasLen, 1)
+	c.Assert(pod.Spec.Volumes[0].VolumeSource.PersistentVolumeClaim.ClaimName, Equals, pvcName)
+	c.Assert(len(pod.Spec.Containers[0].VolumeMounts), Equals, 0)
+	c.Assert(pod.Spec.Containers[0].VolumeDevices[0].DevicePath, Equals, "/mnt/data1")
 }
 
 func (s *PodSuite) TestGetPodLogs(c *C) {