feat: volume replication

Add the capability to replicate a ZFS disk to another zone (proxmox node).

Signed-off-by: Serge Logvinov <[email protected]>

Author: sergelogvinov

docker-compose.yml

@@ -52,7 +52,7 @@ services:
       - "--csi-address=unix:///csi/csi.sock"
       - "--leader-election=false"
       - "--kubeconfig=/etc/kubernetes/kubeconfig"
-      - "--feature-gates=VolumeAttributesClass=true"
+      # - "--feature-gates=VolumeAttributesClass=true"
     volumes:
       - type: volume
         source: socket-dir
@@ -69,7 +69,7 @@ services:
       - "--csi-address=unix:///csi/csi.sock"
       - "--leader-election=false"
       - "--default-fstype=ext4"
-      - "--feature-gates=VolumeAttributesClass=true"
+      # - "--feature-gates=VolumeAttributesClass=true"
       - "--enable-capacity"
       - "--capacity-ownerref-level=-1"
       - "--capacity-poll-interval=2m"

docs/options.md

@@ -38,6 +38,11 @@ parameters:
   diskIOPS: "4000"
   diskMBps: "1000"
 
+  ## Optional: Zone replication
+  replicate: "true"
+  replicateSchedule: "*/15"
+  replicateZones: "pve-1,pve-3"
+
 # Optional: This field allows you to specify additional mount options to be applied when the volume is mounted on the node
 mountOptions:
   # Common for ssd
@@ -77,6 +82,10 @@ parameters:
 
   ## Optional: Backup disk with VM
   backup: "true"
+
+  ## Optional: Zone replication
+  replicateSchedule: "*/30"
+  replicateZones: "rnd-1,rnd-2"
 ```
 
 ## Parameters:
@@ -105,6 +114,10 @@ metadata:
 
 * `backup` - set true if you want to backup the disk with VM. Dangerous option! Do not use it unless you fully understand how to use it in the recovery process.
 
+* `replicate` - set true if you want to replicate the disk to another zone
+* `replicateSchedule` - replication schedule [in systemd calendar format](https://pve.proxmox.com/pve-docs/pve-admin-guide.html#pvesr_schedule_time_format) (default: `*/15`)
+* `replicateZones` - zones where the disk will be replicated, separated by commas
+
 ## AllowVolumeExpansion
 
 Allow you to resize (expand) the PVC in future.

docs/proxmox-zfs.yaml

@@ -0,0 +1,16 @@
+allowVolumeExpansion: true
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: proxmox-zfs
+parameters:
+  csi.storage.k8s.io/fstype: xfs
+  #
+  storage: zfs
+  #
+  replicate: "true"
+  replicateSchedule: "*/15"
+  replicateZones: "pve-1,pve-2"
+provisioner: csi.proxmox.sinextra.dev
+reclaimPolicy: Delete
+volumeBindingMode: WaitForFirstConsumer

pkg/csi/controller.go

@@ -30,9 +30,10 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	proxmox "github.com/sergelogvinov/proxmox-cloud-controller-manager/pkg/cluster"
+	cluster "github.com/sergelogvinov/proxmox-cloud-controller-manager/pkg/cluster"
 	"github.com/sergelogvinov/proxmox-csi-plugin/pkg/helpers/ptr"
 	"github.com/sergelogvinov/proxmox-csi-plugin/pkg/metrics"
+	"github.com/sergelogvinov/proxmox-csi-plugin/pkg/proxmox"
 	"github.com/sergelogvinov/proxmox-csi-plugin/pkg/tools"
 	volume "github.com/sergelogvinov/proxmox-csi-plugin/pkg/volume"
 
@@ -60,22 +61,22 @@ var controllerCaps = []csi.ControllerServiceCapability_RPC_Type{
 
 // ControllerService is the controller service for the CSI driver
 type ControllerService struct {
-	Cluster     *proxmox.Cluster
+	Cluster     *cluster.Cluster
 	Kclient     clientkubernetes.Interface
-	Provider    proxmox.Provider
+	Provider    cluster.Provider
 	volumeLocks sync.Mutex
 
 	csi.UnimplementedControllerServer
 }
 
 // NewControllerService returns a new controller service
 func NewControllerService(kclient *clientkubernetes.Clientset, cloudConfig string) (*ControllerService, error) {
-	cfg, err := proxmox.ReadCloudConfigFromFile(cloudConfig)
+	cfg, err := cluster.ReadCloudConfigFromFile(cloudConfig)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read config: %v", err)
 	}
 
-	cluster, err := proxmox.NewCluster(&cfg, nil)
+	cluster, err := cluster.NewCluster(&cfg, nil)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create proxmox cluster client: %v", err)
 	}
@@ -110,7 +111,7 @@ func (d *ControllerService) CreateVolume(_ context.Context, request *csi.CreateV
 
 	klog.V(5).InfoS("CreateVolume: parameters", "parameters", params)
 
-	_, err := ExtractAndDefaultParameters(params)
+	paramsSC, err := ExtractAndDefaultParameters(params)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -192,6 +193,33 @@ func (d *ControllerService) CreateVolume(_ context.Context, request *csi.CreateV
 	vmr.SetNode(zone)
 	vmr.SetVmType("qemu")
 
+	if paramsSC.Replicate != nil && *paramsSC.Replicate {
+		if storageConfig["type"].(string) != "zfspool" { //nolint:errcheck
+			return nil, status.Error(codes.Internal, "error: storage type is not zfs in replication mode")
+		}
+
+		vmr, err = cl.GetVmRefByName(pvc)
+		if err != nil {
+			id, err := cl.GetNextID(vmID + 1)
+			if err != nil {
+				klog.ErrorS(err, "CreateVolume: failed to get next id", "cluster", region)
+
+				return nil, status.Error(codes.Internal, err.Error())
+			}
+
+			vmr = pxapi.NewVmRef(id)
+			vmr.SetNode(zone)
+			vmr.SetVmType("qemu")
+
+			mc := metrics.NewMetricContext("CreateVm")
+			if err := proxmox.CreateQemuVM(cl, vmr, pvc); mc.ObserveRequest(err) != nil {
+				klog.ErrorS(err, "CreateVolume: failed to create vm", "cluster", region)
+
+				return nil, status.Error(codes.Internal, err.Error())
+			}
+		}
+	}
+
 	vol := volume.NewVolume(region, zone, params[StorageIDKey], fmt.Sprintf("vm-%d-%s", vmr.VmId(), pvc))
 	if storageConfig["path"] != nil && storageConfig["path"].(string) != "" { //nolint:errcheck
 		vol = volume.NewVolume(region, zone, params[StorageIDKey], fmt.Sprintf("%d/vm-%d-%s.raw", vmr.VmId(), vmr.VmId(), pvc))
@@ -218,6 +246,35 @@ func (d *ControllerService) CreateVolume(_ context.Context, request *csi.CreateV
 		return nil, status.Error(codes.AlreadyExists, "volume already exists with same name and different capacity")
 	}
 
+	if paramsSC.Replicate != nil && *paramsSC.Replicate {
+		_, err := attachVolume(cl, vmr, vol.Storage(), vol.Disk(), paramsSC.ToMap())
+		if err != nil {
+			klog.ErrorS(err, "CreateVolume: failed to attach volume", "cluster", region, "volumeID", vol.VolumeID(), "vmID", vmr.VmId())
+
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+
+		if paramsSC.ReplicateZones != "" {
+			var replicaZone string
+
+			for _, z := range strings.Split(paramsSC.ReplicateZones, ",") {
+				if z != zone {
+					replicaZone = z
+
+					break
+				}
+			}
+
+			if replicaZone != "" {
+				if err := proxmox.SetQemuVMReplication(cl, vmr, replicaZone, paramsSC.ReplicateSchedule); err != nil {
+					klog.ErrorS(err, "CreateVolume: failed to set replication", "cluster", region, "zone", replicaZone, "volumeID", vol.VolumeID(), "vmID", vmr.VmId())
+
+					return nil, status.Error(codes.Internal, err.Error())
+				}
+			}
+		}
+	}
+
 	volID := vol.VolumeID()
 	if storageConfig["shared"] != nil && int(storageConfig["shared"].(float64)) == 1 { //nolint:errcheck
 		volID = vol.VolumeSharedID()
@@ -279,6 +336,25 @@ func (d *ControllerService) DeleteVolume(_ context.Context, request *csi.DeleteV
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	if vmr.VmId() != vmID {
+		config, err := cl.GetVmConfig(vmr)
+		if err != nil {
+			klog.ErrorS(err, "DeleteVolume: failed to get vm config", "cluster", vol.Cluster(), "volumeName", vol.Disk())
+		}
+
+		if config != nil {
+			vmName := config["name"].(string) //nolint:errcheck
+			if vmName != "" && strings.HasSuffix(vol.Disk(), vmName) {
+				mc := metrics.NewMetricContext("deleteVm")
+				if err := proxmox.DeleteQemuVM(cl, vmr); mc.ObserveRequest(err) != nil {
+					klog.ErrorS(err, "DeleteVolume: failed to delete vm", "cluster", vol.Cluster(), "volumeName", vol.Disk())
+
+					return nil, status.Error(codes.Internal, fmt.Sprintf("failed to delete volume: %s", vol.Disk()))
+				}
+			}
+		}
+	}
+
 	mc := metrics.NewMetricContext("deleteVolume")
 	if _, err := cl.DeleteVolume(vmr, vol.Storage(), vol.Disk()); mc.ObserveRequest(err) != nil {
 		klog.ErrorS(err, "DeleteVolume: failed to delete volume", "cluster", vol.Cluster(), "volumeName", vol.Disk())
@@ -729,7 +805,7 @@ func (d *ControllerService) getVMRefbyNodeID(ctx context.Context, cl *pxapi.Clie
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	if d.Provider == proxmox.ProviderCapmox {
+	if d.Provider == cluster.ProviderCapmox {
 		vmr, _, err = d.Cluster.FindVMByUUID(node.Status.NodeInfo.SystemUUID)
 		if err != nil {
 			return nil, status.Error(codes.Internal, err.Error())

pkg/csi/driver.go

@@ -21,9 +21,9 @@ const (
 	// DriverName is the name of the CSI driver
 	DriverName = "csi.proxmox.sinextra.dev"
 	// DriverVersion is the version of the CSI driver
-	DriverVersion = "0.4.0"
+	DriverVersion = "0.5.0"
 	// DriverSpecVersion CSI spec version
-	DriverSpecVersion = "1.9.0"
+	DriverSpecVersion = "1.11.0"
 
 	// MaxVolumesPerNode is the maximum number of volumes that can be attached to a node
 	MaxVolumesPerNode = 24

pkg/csi/parameters.go

@@ -58,14 +58,17 @@ type StorageParameters struct {
 	IopsWrite      *int   `json:"iops_wr,omitempty"`
 	ReadSpeedMbps  *int   `json:"mbps_rd,omitempty"`
 	WriteSpeedMbps *int   `json:"mbps_wr,omitempty"`
-	Replicate      *bool  `json:"replicate,omitempty"`
 	SSD            *bool  `json:"ssd,omitempty"            cfg:"ssd"`
 	ReadOnly       *bool  `json:"ro,omitempty"`
 
 	Iops      *int `cfg:"diskIOPS"`
 	SpeedMbps *int `cfg:"diskMBps"`
 	BlockSize *int `cfg:"blockSize"`
 	InodeSize *int `cfg:"inodeSize"`
+
+	Replicate         *bool  `json:"replicate,omitempty"      cfg:"replicate"`
+	ReplicateSchedule string `cfg:"replicateSchedule"`
+	ReplicateZones    string `cfg:"replicateZones"`
 }
 
 // ModifyVolumeParameters contains parameters to modify a volume
@@ -80,6 +83,9 @@ type ModifyVolumeParameters struct {
 
 	Iops      *int `cfg:"diskIOPS"`
 	SpeedMbps *int `cfg:"diskMBps"`
+
+	ReplicateSchedule string `cfg:"replicateSchedule"`
+	ReplicateZones    string `cfg:"replicateZones"`
 }
 
 // ExtractAndDefaultParameters extracts storage parameters from a map and sets default values.
@@ -122,6 +128,20 @@ func ExtractAndDefaultParameters(parameters map[string]string) (StorageParameter
 
 					f.Set(reflect.ValueOf(ptr.Ptr(i)))
 				}
+			} else {
+				switch f.Kind() { //nolint:exhaustive
+				case reflect.String:
+					f.Set(reflect.ValueOf(v))
+				case reflect.Bool:
+					f.Set(reflect.ValueOf(v == "true"))
+				case reflect.Int:
+					i, err := strconv.Atoi(v)
+					if err != nil {
+						return p, fmt.Errorf("parameters %s must be a number", fieldName)
+					}
+
+					f.Set(reflect.ValueOf(i))
+				}
 			}
 		}
 	}
@@ -154,6 +174,12 @@ func ExtractAndDefaultParameters(parameters map[string]string) (StorageParameter
 		p.WriteSpeedMbps = ptr.Ptr(*p.SpeedMbps)
 	}
 
+	if p.Replicate != nil && *p.Replicate {
+		if p.ReplicateZones == "" {
+			return p, fmt.Errorf("parameters %s must be provided", "replicateZones")
+		}
+	}
+
 	return p, nil
 }
 

pkg/csi/parameters_test.go

@@ -94,6 +94,20 @@ func Test_ExtractAndDefaultParameters(t *testing.T) {
 				IopsWrite: ptr.Ptr(100),
 			},
 		},
+		{
+			msg: "replication disk",
+			params: map[string]string{
+				csi.StorageIDKey: "local-lvm",
+				"replicate":      "true",
+				"replicateZones": "zone1,zone2",
+			},
+			storage: csi.StorageParameters{
+				Backup:         ptr.Ptr(false),
+				IOThread:       true,
+				Replicate:      ptr.Ptr(true),
+				ReplicateZones: "zone1,zone2",
+			},
+		},
 	}
 
 	for _, testCase := range tests {
@@ -140,6 +154,21 @@ func Test_ToMap(t *testing.T) {
 				"iops_wr":  "100",
 			},
 		},
+		{
+			msg: "Params with replication",
+			storage: csi.StorageParameters{
+				Cache:             "directsync",
+				IOThread:          true,
+				Replicate:         ptr.Ptr(true),
+				ReplicateZones:    "zone1,zone2",
+				ReplicateSchedule: "*/30",
+			},
+			params: map[string]string{
+				"cache":     "directsync",
+				"iothread":  "1",
+				"replicate": "1",
+			},
+		},
 	}
 
 	for _, testCase := range tests {

pkg/proxmox/doc.go

@@ -0,0 +1,18 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package proxmox implements tools to work with Proxmox VM.
+package proxmox