Add support for dm-integrity on lvm raids (#106)

.github/workflows/docker.yaml

@@ -43,6 +43,9 @@ jobs:
     - name: Checkout
       uses: actions/checkout@v4
 
+    - name: load modules for dm-raid and dm-integrity
+      run: lsmod && sudo modprobe dm-raid && sudo modprobe dm-integrity && lsmod
+
     - name: Set up Go 1.23
       uses: actions/setup-go@v5
       with:

cmd/provisioner/createlv.go

@@ -32,6 +32,10 @@ func createLVCmd() *cli.Command {
 				Name:  flagDevicesPattern,
 				Usage: "Required. comma-separated grok patterns of the physical volumes to use.",
 			},
+			&cli.BoolFlag{
+				Name:  flagIntegrity,
+				Usage: "Optional. if set type must be mirrored. Inserts a dm-integrity layer.",
+			},
 		},
 		Action: func(c *cli.Context) error {
 			if err := createLV(c); err != nil {
@@ -64,6 +68,7 @@ func createLV(c *cli.Context) error {
 	if devicesPattern == "" {
 		return fmt.Errorf("invalid empty flag %v", flagDevicesPattern)
 	}
+	integrity := c.Bool(flagIntegrity)
 
 	klog.Infof("create lv %s size:%d vg:%s devicespattern:%s  type:%s", lvName, lvSize, vgName, devicesPattern, lvmType)
 
@@ -72,7 +77,7 @@ func createLV(c *cli.Context) error {
 		return fmt.Errorf("unable to create vg: %w output:%s", err, output)
 	}
 
-	output, err = lvm.CreateLVS(vgName, lvName, lvSize, lvmType)
+	output, err = lvm.CreateLVS(vgName, lvName, lvSize, lvmType, integrity)
 	if err != nil {
 		return fmt.Errorf("unable to create lv: %w output:%s", err, output)
 	}

cmd/provisioner/main.go

@@ -14,6 +14,7 @@ const (
 	flagVGName         = "vgname"
 	flagDevicesPattern = "devices"
 	flagLVMType        = "lvmtype"
+	flagIntegrity      = "integrity"
 )
 
 func cmdNotFound(c *cli.Context, command string) {

pkg/lvm/controllerserver.go

@@ -114,6 +114,10 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 	if !(lvmType == "linear" || lvmType == "mirror" || lvmType == "striped") {
 		return nil, status.Errorf(codes.Internal, "lvmType is incorrect: %s", lvmType)
 	}
+	integrity, err := strconv.ParseBool(req.GetParameters()["integrity"])
+	if err != nil {
+		klog.Warningf("Could not parse 'integrity' request parameter, assuming false: %s", err)
+	}
 
 	volumeContext := req.GetParameters()
 	size := strconv.FormatInt(req.GetCapacityRange().GetRequiredBytes(), 10)
@@ -140,6 +144,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 		namespace:        cs.namespace,
 		vgName:           cs.vgName,
 		hostWritePath:    cs.hostWritePath,
+		integrity:        integrity,
 	}
 	if err := createProvisionerPod(ctx, va); err != nil {
 		klog.Errorf("error creating provisioner pod :%v", err)

pkg/lvm/lvm.go

@@ -78,6 +78,7 @@ type volumeAction struct {
 	namespace        string
 	vgName           string
 	hostWritePath    string
+	integrity        bool
 }
 
 const (
@@ -264,6 +265,9 @@ func createProvisionerPod(ctx context.Context, va volumeAction) (err error) {
 	args := []string{}
 	if va.action == actionTypeCreate {
 		args = append(args, "createlv", "--lvsize", fmt.Sprintf("%d", va.size), "--devices", va.devicesPattern, "--lvmtype", va.lvmType)
+		if va.integrity {
+			args = append(args, "--integrity")
+		}
 	}
 	if va.action == actionTypeDelete {
 		args = append(args, "deletelv")
@@ -511,7 +515,7 @@ func CreateVG(name string, devicesPattern string) (string, error) {
 
 // CreateLVS creates the new volume
 // used by lvcreate provisioner pod and by nodeserver for ephemeral volumes
-func CreateLVS(vg string, name string, size uint64, lvmType string) (string, error) {
+func CreateLVS(vg string, name string, size uint64, lvmType string, integrity bool) (string, error) {
 
 	if lvExists(vg, name) {
 		klog.Infof("logicalvolume: %s already exists\n", name)
@@ -550,6 +554,15 @@ func CreateLVS(vg string, name string, size uint64, lvmType string) (string, err
 		return "", fmt.Errorf("unsupported lvmtype: %s", lvmType)
 	}
 
+	if integrity {
+		switch lvmType {
+		case mirrorType:
+			args = append(args, "--raidintegrity", "y")
+		default:
+			return "", fmt.Errorf("integrity is only supported if type is mirror")
+		}
+	}
+
 	tags := []string{"lv.metal-stack.io/csi-lvm-driver"}
 	for _, tag := range tags {
 		args = append(args, "--addtag", tag)

pkg/lvm/nodeserver.go

@@ -122,7 +122,7 @@ func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 			return nil, fmt.Errorf("unable to create vg: %w output:%s", err, output)
 		}
 
-		output, err = CreateLVS(ns.vgName, volID, size, req.GetVolumeContext()["type"])
+		output, err = CreateLVS(ns.vgName, volID, size, req.GetVolumeContext()["type"], false)
 		if err != nil {
 			return nil, fmt.Errorf("unable to create lv: %w output:%s", err, output)
 		}

tests/bats/test.bats

@@ -109,6 +109,51 @@
     [ "$status" -eq 0 ]
 }
 
+@test "create storageclass mirror-integrity" {
+    # Requires kernel modules:
+    # modprobe dm-raid && modprobe dm-integrity
+    run kubectl apply -f files/storageclass.mirror-integrity.yaml --wait --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "create pvc mirror-integrity" {
+    run kubectl apply -f files/pvc.mirror-integrity.yaml --wait --timeout=10s
+    [ "$status" -eq 0 ]
+
+    run kubectl wait --for=jsonpath='{.status.phase}'=Pending -f files/pvc.mirror-integrity.yaml --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "deploy mirror-integrity pod" {
+    run kubectl apply -f files/pod.mirror-integrity.vol.yaml --wait --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "mirror-integrity pod running" {
+    run kubectl wait --for=jsonpath='{.status.phase}'=Running -f files/pod.mirror-integrity.vol.yaml --timeout=30s
+    [ "$status" -eq 0 ]
+}
+
+@test "pvc mirror-integrity bound" {
+    run kubectl wait --for=jsonpath='{.status.phase}'=Bound -f files/pvc.mirror-integrity.yaml --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "delete mirror-integrity pod" {
+    run kubectl delete -f files/pod.mirror-integrity.vol.yaml --grace-period=0 --wait --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "delete mirror-integrity pvc" {
+    run kubectl delete -f files/pvc.mirror-integrity.yaml --grace-period=0 --wait --timeout=10s
+    [ "$status" -eq 0 ]
+}
+
+@test "delete storageclass mirror-integrity" {
+    run kubectl delete -f files/storageclass.mirror-integrity.yaml --wait --timeout=20s
+    [ "$status" -eq 0 ]
+}
+
 @test "deploy inline xfs pod with ephemeral volume" {
     run kubectl apply -f files/pod.inline.vol.xfs.yaml --wait --timeout=20s
     [ "$status" -eq 0 ]

tests/files/pod.mirror-integrity.vol.yaml

@@ -0,0 +1,33 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: volume-test
+spec:
+  containers:
+  - name: volume-test
+    image: alpine
+    imagePullPolicy: IfNotPresent
+    command:
+      - tail
+      - -f
+      - /etc/hosts
+    securityContext:
+      allowPrivilegeEscalation: false
+      runAsNonRoot: true
+      runAsUser: 10014
+      seccompProfile:
+        type: RuntimeDefault
+      capabilities:
+        drop:
+          - ALL
+    volumeMounts:
+    - name: mirror-integrity
+      mountPath: /mirror-integrity
+    resources:
+      limits:
+        cpu: 100m
+        memory: 100M
+  volumes:
+  - name: mirror-integrity
+    persistentVolumeClaim:
+      claimName: lvm-pvc-mirror-integrity

tests/files/pvc.mirror-integrity.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: lvm-pvc-mirror-integrity
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: csi-driver-lvm-mirror-integrity
+  resources:
+    requests:
+      storage: 100Mi

tests/files/storageclass.mirror-integrity.yaml

@@ -0,0 +1,11 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: csi-driver-lvm-mirror-integrity
+parameters:
+  type: mirror
+  integrity: "true"
+provisioner: lvm.csi.metal-stack.io
+allowVolumeExpansion: true
+reclaimPolicy: Delete
+volumeBindingMode: WaitForFirstConsumer