Add support for dm-integrity on mirrors

This adds support for automated raid integrity checking and can thereby
help preventing bit rot. It however costs some read and write
performance.

Details can be found in `man lvmraid`.

To run these tests we need the dm-raid and dm-integrity kernel modules
loaded.

.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,7 @@ 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, _ := strconv.ParseBool(req.GetParameters()["integrity"])
 
 	volumeContext := req.GetParameters()
 	size := strconv.FormatInt(req.GetCapacityRange().GetRequiredBytes(), 10)
@@ -140,6 +141,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,34 @@
     [ "$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=10s
+    [ "$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 "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: 10Mi

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