feat: implement CreateVolume with loopback image creation via Job

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: tryuuu

charts/csi-driver/templates/controller-deployment.yaml

@@ -23,6 +23,8 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: spec.nodeName
+            - name: CSI_IMAGE
+              value: {{ .Values.image.repository }}:{{ .Values.image.tag }}
           volumeMounts:
             - name: socket-dir
               mountPath: /csi

internal/driver/controller.go

@@ -22,6 +22,7 @@ const (
 	topologyKey      = "kubernetes.io/hostname"
 	deleteJobNS      = "kube-system"
 	deleteJobTimeout = 2 * time.Minute
+	createJobTimeout = 5 * time.Minute
 )
 
 var errNodeNotFound = errors.New("node not found")
@@ -30,28 +31,44 @@ var errNoNodeAffinity = errors.New("no nodeAffinity on PV")
 type ControllerServer struct {
 	csi.UnimplementedControllerServer
 	kubeClient kubernetes.Interface
+	jobImage   string
 }
 
-func (s *ControllerServer) CreateVolume(_ context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+func (s *ControllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	if req.GetName() == "" {
 		return nil, status.Error(codes.InvalidArgument, "volume name is required")
 	}
 
+	sizeBytes := req.GetCapacityRange().GetRequiredBytes()
+	if sizeBytes == 0 {
+		return nil, status.Error(codes.InvalidArgument, "required bytes must be greater than 0")
+	}
+
 	// Preferred → Requisite の順でトポロジーを選ぶ。
 	// external-provisioner が WaitForFirstConsumer で決定したノードに PV を固定
 	var topology []*csi.Topology
+	var nodeName string
 	if topo := req.GetAccessibilityRequirements(); topo != nil {
 		if len(topo.GetPreferred()) > 0 {
 			topology = []*csi.Topology{topo.GetPreferred()[0]}
+			nodeName = topo.GetPreferred()[0].GetSegments()[topologyKey]
 		} else if len(topo.GetRequisite()) > 0 {
 			topology = []*csi.Topology{topo.GetRequisite()[0]}
+			nodeName = topo.GetRequisite()[0].GetSegments()[topologyKey]
 		}
 	}
+	if nodeName == "" {
+		return nil, status.Error(codes.InvalidArgument, "cannot determine target node from topology")
+	}
+
+	if err := s.runCreateJob(ctx, req.GetName(), nodeName, sizeBytes); err != nil {
+		return nil, status.Errorf(codes.Internal, "create job failed: %v", err)
+	}
 
 	return &csi.CreateVolumeResponse{
 		Volume: &csi.Volume{
 			VolumeId:           req.GetName(),
-			CapacityBytes:      req.GetCapacityRange().GetRequiredBytes(),
+			CapacityBytes:      sizeBytes,
 			AccessibleTopology: topology,
 		},
 	}, nil
@@ -107,7 +124,104 @@ func (s *ControllerServer) nodeFromPV(ctx context.Context, volumeID string) (str
 	return "", errNodeNotFound
 }
 
-// は対象ノードで `rm -rf <volumeBasePath>/<volumeID>` を実行する Job を作成し、完了を待つ。
+// 対象ノードでイメージファイルの作成・ループデバイス割り当て・フォーマットを行う Job を実行する。
+func (s *ControllerServer) runCreateJob(ctx context.Context, volumeID, nodeName string, sizeBytes int64) error {
+	jobName := createJobName(volumeID)
+	imgPath := fmt.Sprintf("%s/%s.img", volumeBasePath, volumeID)
+
+	script := fmt.Sprintf(`set -e
+IMG='%s'
+if [ ! -f "$IMG" ]; then
+  fallocate -l %d "$IMG"
+fi
+DEV=$(losetup -j "$IMG" | cut -d: -f1)
+if [ -z "$DEV" ]; then
+  DEV=$(losetup -f --show "$IMG")
+fi
+if ! blkid "$DEV" >/dev/null 2>&1; then
+  mkfs.ext4 "$DEV"
+fi`, imgPath, sizeBytes)
+
+	privileged := true
+	job := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      jobName,
+			Namespace: deleteJobNS,
+		},
+		Spec: batchv1.JobSpec{
+			Template: corev1.PodTemplateSpec{
+				Spec: corev1.PodSpec{
+					NodeName:      nodeName,
+					RestartPolicy: corev1.RestartPolicyNever,
+					Containers: []corev1.Container{
+						{
+							Name:    "create",
+							Image:   s.jobImage,
+							Command: []string{"/bin/sh", "-c", script},
+							SecurityContext: &corev1.SecurityContext{
+								Privileged: &privileged,
+							},
+							VolumeMounts: []corev1.VolumeMount{
+								{Name: "data", MountPath: volumeBasePath},
+								{Name: "dev", MountPath: "/dev"},
+							},
+						},
+					},
+					Volumes: []corev1.Volume{
+						{
+							Name: "data",
+							VolumeSource: corev1.VolumeSource{
+								HostPath: &corev1.HostPathVolumeSource{Path: volumeBasePath},
+							},
+						},
+						{
+							Name: "dev",
+							VolumeSource: corev1.VolumeSource{
+								HostPath: &corev1.HostPathVolumeSource{Path: "/dev"},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	_, err := s.kubeClient.BatchV1().Jobs(deleteJobNS).Create(ctx, job, metav1.CreateOptions{})
+	if err != nil && !k8serrors.IsAlreadyExists(err) {
+		return err
+	}
+
+	deadline := time.Now().Add(createJobTimeout)
+	for time.Now().Before(deadline) {
+		j, err := s.kubeClient.BatchV1().Jobs(deleteJobNS).Get(ctx, jobName, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		if j.Status.Succeeded > 0 {
+			_ = s.kubeClient.BatchV1().Jobs(deleteJobNS).Delete(ctx, jobName, metav1.DeleteOptions{})
+			return nil
+		}
+		if j.Status.Failed > 3 {
+			_ = s.kubeClient.BatchV1().Jobs(deleteJobNS).Delete(ctx, jobName, metav1.DeleteOptions{})
+			return fmt.Errorf("create job failed after retries")
+		}
+		time.Sleep(3 * time.Second)
+	}
+
+	return fmt.Errorf("create job timed out after %v", createJobTimeout)
+}
+
+func createJobName(volumeID string) string {
+	const prefix = "csi-create-"
+	name := prefix + volumeID
+	if len(name) <= 63 {
+		return name
+	}
+	h := sha256.Sum256([]byte(volumeID))
+	return fmt.Sprintf("%s%x", prefix, h[:])[:63]
+}
+
+// 対象ノードで `rm -rf <volumeBasePath>/<volumeID>` を実行する Job を作成し、完了を待つ。
 func (s *ControllerServer) runDeleteJob(ctx context.Context, volumeID, nodeName string) error {
 	jobName := deleteJobName(volumeID)
 

internal/driver/server.go

@@ -34,7 +34,10 @@ func Run(socketPath string) error {
 
 	srv := grpc.NewServer()
 	csi.RegisterIdentityServer(srv, &IdentityServer{})
-	csi.RegisterControllerServer(srv, &ControllerServer{kubeClient: kubeClient})
+	csi.RegisterControllerServer(srv, &ControllerServer{
+		kubeClient: kubeClient,
+		jobImage:   os.Getenv("CSI_IMAGE"),
+	})
 	csi.RegisterNodeServer(srv, nodeServer)
 
 	return srv.Serve(lis)