feat: add validating webhook for persistentvolumes (#928)

ref: juicedata/voc#246

Signed-off-by: Xuhui zhang <[email protected]>

.github/scripts/e2e-test.py

@@ -49,6 +49,7 @@
     test_dynamic_expand,
     test_multi_pvc,
     test_mountpod_recreated,
+    test_validate_pv,
 )
 from util import die, mount_on_host, umount, clean_juicefs_volume, deploy_secret_and_sc, check_do_test
 
@@ -146,6 +147,7 @@
                 test_static_mount_image_with_webhook()
                 test_deployment_static_patch_pv_with_webhook()
                 test_static_job_complete()
+                test_validate_pv()
                 if not IN_CCI:
                     test_delete_pvc()
                     test_job_complete_using_storage()

.github/scripts/test_case.py

@@ -2609,4 +2609,26 @@ def test_mountpod_recreated():
         raise Exception("Pods of deployment {} are not delete within 5 min.".format(deployment.name))
     LOG.info("Remove pvc {}".format(pvc.name))
     pvc.delete()
-    return
+    return
+
+def test_validate_pv():
+    LOG.info("[test case] Test validate pv begin..")
+    # deploy one pv
+    pv = PV(name="pv-with-duplicate-handle", access_mode="ReadWriteMany", volume_handle="pv-with-duplicate-handle", secret_name=SECRET_NAME)
+    LOG.info("Deploy pv {}".format(pv.name))
+    pv.create()
+
+    # deploy pv with duplicate handle
+    pv1 = PV(name="pv-with-duplicate-handle-1", access_mode="ReadWriteMany", volume_handle="pv-with-duplicate-handle", secret_name=SECRET_NAME)
+    LOG.info("Deploy pv {}".format(pv1.name))
+    try:
+        pv1.create()
+    except Exception as e:
+        LOG.info(e)
+        LOG.info("Test pass.")
+        # delete test resources
+        LOG.info("Remove pv {}".format(pv.name))
+        pv.delete()
+        return
+
+    raise Exception("PV with duplicate handle should not be created.")

deploy/kubernetes/base/resources.yaml

@@ -671,6 +671,23 @@ webhooks:
     failurePolicy: Ignore
     sideEffects: None
     admissionReviewVersions: ["v1"]
+  - name: validate.pv.juicefs.com
+    matchPolicy: Equivalent
+    rules:
+      - apiGroups: [""]
+        apiVersions: ["v1"]
+        operations: ["CREATE"]
+        resources: ["persistentvolumes"]
+    clientConfig:
+      service:
+        namespace: kube-system
+        name: juicefs-admission-webhook
+        path: "/juicefs/validate-pv"
+      caBundle: CA_BUNDLE
+    timeoutSeconds: 5
+    failurePolicy: Ignore
+    sideEffects: None
+    admissionReviewVersions: ["v1"]
 ---
 apiVersion: v1
 kind: Service

deploy/kubernetes/csi-ci/webhook/statefulset.yaml

@@ -15,3 +15,4 @@ spec:
             - --leader-election
             - --v=6
             - --webhook=true
+            - --validating-webhook=true

deploy/webhook-with-certmanager.yaml

@@ -682,3 +682,25 @@ webhooks:
     - secrets
   sideEffects: None
   timeoutSeconds: 5
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    caBundle: CA_BUNDLE
+    service:
+      name: juicefs-admission-webhook
+      namespace: kube-system
+      path: /juicefs/validate-pv
+  failurePolicy: Ignore
+  matchPolicy: Equivalent
+  name: validate.pv.juicefs.com
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumes
+  sideEffects: None
+  timeoutSeconds: 5

deploy/webhook.yaml

@@ -683,3 +683,25 @@ webhooks:
     - secrets
   sideEffects: None
   timeoutSeconds: 5
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    caBundle: CA_BUNDLE
+    service:
+      name: juicefs-admission-webhook
+      namespace: kube-system
+      path: /juicefs/validate-pv
+  failurePolicy: Ignore
+  matchPolicy: Equivalent
+  name: validate.pv.juicefs.com
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumes
+  sideEffects: None
+  timeoutSeconds: 5

pkg/driver/node.go

@@ -27,6 +27,7 @@ import (
 	"github.com/container-storage-interface/spec/lib/go/csi"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/klog"
 	k8sexec "k8s.io/utils/exec"
@@ -121,6 +122,15 @@ func (d *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 		return nil, status.Error(codes.InvalidArgument, "Volume capability not supported")
 	}
 
+	var pv *corev1.PersistentVolume
+	var err error
+	if d.k8sClient != nil {
+		pv, err = d.k8sClient.GetPersistentVolume(ctx, volumeID)
+		if k8serrors.IsNotFound(err) {
+			klog.Warningf("volumeHandle %s is not the same as pv name, mount pod might not be created", volumeID)
+		}
+	}
+
 	klog.V(5).Infof("NodePublishVolume: creating dir %s", target)
 	if err := d.juicefs.CreateTarget(ctx, target); err != nil {
 		return nil, status.Errorf(codes.Internal, "Could not create dir %q: %v", target, err)
@@ -169,14 +179,8 @@ func (d *nodeService) NodePublishVolume(ctx context.Context, req *csi.NodePublis
 		if err != nil {
 			return nil, status.Errorf(codes.Internal, "invalid capacity %s: %v", cap, err)
 		}
-		if d.k8sClient != nil {
-			pv, err := d.k8sClient.GetPersistentVolume(ctx, volumeID)
-			if err != nil && !k8serrors.IsNotFound(err) {
-				return nil, status.Errorf(codes.Internal, "get pv %s: %v", volumeID, err)
-			}
-			if err == nil && pv != nil {
-				capacity = pv.Spec.Capacity.Storage().Value()
-			}
+		if pv != nil {
+			capacity = pv.Spec.Capacity.Storage().Value()
 		}
 		settings, err := d.juicefs.Settings(ctx, volumeID, secrets, volCtx, options)
 		if err != nil {

pkg/k8sclient/client.go

@@ -354,6 +354,21 @@ func (k *K8sClient) ListPersistentVolumes(ctx context.Context, labelSelector *me
 	return pvList.Items, nil
 }
 
+func (k *K8sClient) ListPersistentVolumesByVolumeHandle(ctx context.Context, volumeHandle string) ([]corev1.PersistentVolume, error) {
+	pvs, err := k.ListPersistentVolumes(ctx, nil, nil)
+	if err != nil {
+		return nil, err
+	}
+	var result []corev1.PersistentVolume
+	for _, pv := range pvs {
+		pv := pv
+		if pv.Spec.CSI != nil && pv.Spec.CSI.VolumeHandle == volumeHandle {
+			result = append(result, pv)
+		}
+	}
+	return result, nil
+}
+
 func (k *K8sClient) ListStorageClasses(ctx context.Context) ([]storagev1.StorageClass, error) {
 	klog.V(6).Info("List storageclass")
 	scList, err := k.StorageV1().StorageClasses().List(ctx, metav1.ListOptions{})

pkg/webhook/handler/handler.go

@@ -19,6 +19,7 @@ package handler
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 
 	corev1 "k8s.io/api/core/v1"
@@ -139,3 +140,48 @@ func (s *SecretHandler) Handle(ctx context.Context, request admission.Request) a
 	}
 	return admission.Allowed("")
 }
+
+type PVHandler struct {
+	Client *k8sclient.K8sClient
+	// A decoder will be automatically injected
+	decoder *admission.Decoder
+}
+
+func NewPVHandler(client *k8sclient.K8sClient) *PVHandler {
+	return &PVHandler{
+		Client: client,
+	}
+}
+
+// InjectDecoder injects the decoder.
+func (s *PVHandler) InjectDecoder(d *admission.Decoder) error {
+	s.decoder = d
+	return nil
+}
+
+func (s *PVHandler) Handle(ctx context.Context, request admission.Request) admission.Response {
+	pv := &corev1.PersistentVolume{}
+	err := s.decoder.Decode(request, pv)
+	if err != nil {
+		klog.Errorf("unable to decoder pv from req, %v", err)
+		return admission.Errored(http.StatusBadRequest, err)
+	}
+
+	if pv.Spec.CSI == nil || pv.Spec.CSI.Driver != config.DriverName {
+		return admission.Allowed("")
+	}
+	if pv.Spec.StorageClassName != "" {
+		return admission.Allowed("")
+	}
+
+	volumeHandle := pv.Spec.CSI.VolumeHandle
+	existPvs, err := s.Client.ListPersistentVolumesByVolumeHandle(ctx, volumeHandle)
+	if err != nil {
+		klog.Errorf("list pv by volume handle %s failed, err: %v", volumeHandle, err)
+		return admission.Errored(http.StatusBadRequest, err)
+	}
+	if len(existPvs) > 0 {
+		return admission.Denied(fmt.Sprintf("pv %s with volume handle %s already exists", pv.Name, volumeHandle))
+	}
+	return admission.Allowed("")
+}

pkg/webhook/handler/register.go

@@ -29,6 +29,7 @@ const (
 	SidecarPath    = "/juicefs/inject-v1-pod"
 	ServerlessPath = "/juicefs/serverless/inject-v1-pod"
 	SecretPath     = "/juicefs/validate-secret"
+	PVPath         = "/juicefs/validate-pv"
 )
 
 // Register registers the handlers to the manager
@@ -40,5 +41,6 @@ func Register(mgr manager.Manager, client *k8sclient.K8sClient) {
 	klog.Infof("Registered webhook handler path %s for serverless", ServerlessPath)
 	if config.ValidatingWebhook {
 		server.Register(SecretPath, &webhook.Admission{Handler: NewSecretHandler(client)})
+		server.Register(PVPath, &webhook.Admission{Handler: NewPVHandler(client)})
 	}
 }

scripts/juicefs-csi-webhook-install.sh

@@ -754,6 +754,28 @@ webhooks:
     - secrets
   sideEffects: None
   timeoutSeconds: 5
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    caBundle: CA_BUNDLE
+    service:
+      name: juicefs-admission-webhook
+      namespace: kube-system
+      path: /juicefs/validate-pv
+  failurePolicy: Ignore
+  matchPolicy: Equivalent
+  name: validate.pv.juicefs.com
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumes
+  sideEffects: None
+  timeoutSeconds: 5
 EOF
   # webhook.yaml end
 
@@ -1448,6 +1470,28 @@ webhooks:
     - secrets
   sideEffects: None
   timeoutSeconds: 5
+- admissionReviewVersions:
+  - v1
+  clientConfig:
+    caBundle: CA_BUNDLE
+    service:
+      name: juicefs-admission-webhook
+      namespace: kube-system
+      path: /juicefs/validate-pv
+  failurePolicy: Ignore
+  matchPolicy: Equivalent
+  name: validate.pv.juicefs.com
+  rules:
+  - apiGroups:
+    - ""
+    apiVersions:
+    - v1
+    operations:
+    - CREATE
+    resources:
+    - persistentvolumes
+  sideEffects: None
+  timeoutSeconds: 5
 EOF
   # webhook-with-certmanager.yaml end