azurefile: kata/kata-cc: add kata node conditional

charts/latest/azurefile-csi-driver/templates/rbac-csi-azurefile-node.yaml

@@ -38,6 +38,9 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get"]
   - apiGroups: ["node.k8s.io"]
     resources: ["runtimeclasses"]
     verbs: ["get", "list"]

deploy/example/kata-cc/statefulset.yaml

@@ -37,7 +37,7 @@ spec:
     - metadata:
         name: persistent-storage
       spec:
-        storageClassName: azurefile-csi-kata-cc
+        storageClassName: azurefile-csi
         accessModes: ["ReadWriteMany"]
         resources:
           requests:

deploy/rbac-csi-azurefile-node.yaml

@@ -37,6 +37,9 @@ rules:
   - apiGroups: [""]
     resources: ["pods"]
     verbs: ["get"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get"]
   - apiGroups: ["node.k8s.io"]
     resources: ["runtimeclasses"]
     verbs: ["get", "list"]

pkg/azurefile/azurefile.go

@@ -48,6 +48,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
+	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
 	mount "k8s.io/mount-utils"
 	"k8s.io/utils/ptr"
@@ -195,7 +196,6 @@ const (
 	FSGroupChangeNone = "None"
 	// define tag value delimiter and default is comma
 	tagValueDelimiterField = "tagvaluedelimiter"
-	enableKataCCMountField = "enablekataccmount"
 )
 
 var (
@@ -289,6 +289,7 @@ type Driver struct {
 	endpoint     string
 	resolver     Resolver
 	directVolume DirectVolume
+	isKataNode   bool
 }
 
 // NewDriver Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
@@ -330,6 +331,7 @@ func NewDriver(options *DriverOptions) *Driver {
 	driver.endpoint = options.Endpoint
 	driver.resolver = new(NetResolver)
 	driver.directVolume = new(directVolume)
+	driver.isKataNode = false
 
 	var err error
 	getter := func(_ context.Context, _ string) (interface{}, error) { return nil, nil }
@@ -453,6 +455,7 @@ func (d *Driver) Run(ctx context.Context) error {
 	csi.RegisterControllerServer(server, d)
 	csi.RegisterNodeServer(server, d)
 	d.server = server
+	d.isKataNode = isKataNode(ctx, d.NodeID, d.kubeClient)
 
 	listener, err := csicommon.ListenEndpoint(d.endpoint)
 	if err != nil {
@@ -1282,3 +1285,36 @@ func (d *Driver) getFileShareClientForSub(subscriptionID string) (fileshareclien
 	}
 	return d.cloud.ComputeClientFactory.GetFileShareClientForSub(subscriptionID)
 }
+
+func getNodeInfoFromLabels(ctx context.Context, nodeID string, kubeClient clientset.Interface) (string, string, error) {
+	if kubeClient == nil || kubeClient.CoreV1() == nil {
+		return "", "", fmt.Errorf("kubeClient is nil")
+	}
+
+	node, err := kubeClient.CoreV1().Nodes().Get(ctx, nodeID, metav1.GetOptions{})
+	if err != nil {
+		return "", "", fmt.Errorf("get node(%s) failed with %v", nodeID, err)
+	}
+
+	if len(node.Labels) == 0 {
+		return "", "", fmt.Errorf("node(%s) label is empty", nodeID)
+	}
+	return node.Labels["kubernetes.azure.com/kata-mshv-vm-isolation"], node.Labels["katacontainers.io/kata-runtime"], nil
+}
+
+func isKataNode(ctx context.Context, nodeID string, kubeClient clientset.Interface) bool {
+	if nodeID == "" {
+		return false
+	}
+
+	kataVMIsolationLabel, kataRuntimeLabel, err := getNodeInfoFromLabels(ctx, nodeID, kubeClient)
+
+	if err != nil {
+		klog.Warningf("failed to get node info from labels: %v", err)
+		return false
+	}
+
+	klog.V(4).Infof("node(%s) labels: kataVMIsolationLabel(%s), kataRuntimeLabel(%s)", nodeID, kataVMIsolationLabel, kataRuntimeLabel)
+
+	return strings.EqualFold(kataVMIsolationLabel, "true") || strings.EqualFold(kataRuntimeLabel, "true")
+}

pkg/azurefile/azurefile_test.go

@@ -39,6 +39,7 @@ import (
 	"go.uber.org/mock/gomock"
 	v1api "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
 
 	"sigs.k8s.io/cloud-provider-azure/pkg/azclient"
@@ -1632,3 +1633,136 @@ func TestGetFileShareClientForSub(t *testing.T) {
 		assert.Equal(t, tc.expectedError, err)
 	}
 }
+
+func TestGetNodeInfoFromLabels(t *testing.T) {
+	testCases := []struct {
+		name         string
+		nodeName     string
+		labels       map[string]string
+		setupClient  bool
+		expectedVals [2]string
+		expectedErr  error
+	}{
+		{
+			name:        "Error when kubeClient is nil",
+			nodeName:    "test-node",
+			setupClient: false,
+			expectedErr: fmt.Errorf("kubeClient is nil"),
+		},
+		{
+			name:        "Error when node does not exist",
+			nodeName:    "nonexistent-node",
+			setupClient: true,
+			expectedErr: fmt.Errorf("get node(nonexistent-node) failed with nodes \"nonexistent-node\" not found"),
+		},
+		{
+			name:        "Error when node has no labels",
+			nodeName:    "test-node",
+			setupClient: true,
+			labels:      map[string]string{}, // Node exists but has no labels
+			expectedErr: fmt.Errorf("node(test-node) label is empty"),
+		},
+		{
+			name:        "Success with kata labels",
+			nodeName:    "test-node",
+			setupClient: true,
+			labels: map[string]string{
+				"kubernetes.azure.com/kata-mshv-vm-isolation": "true",
+				"katacontainers.io/kata-runtime":              "false",
+			},
+			expectedVals: [2]string{"true", "false"},
+			expectedErr:  nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.TODO()
+			var clientset kubernetes.Interface
+
+			if tc.setupClient {
+				clientset = fake.NewSimpleClientset()
+			}
+
+			if tc.labels != nil && tc.setupClient {
+				node := &v1api.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   tc.nodeName,
+						Labels: tc.labels,
+					},
+				}
+				_, err := clientset.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+
+			kataVMIsolation, kataRuntime, err := getNodeInfoFromLabels(ctx, tc.nodeName, clientset)
+
+			if tc.expectedErr != nil {
+				assert.EqualError(t, err, tc.expectedErr.Error())
+			} else {
+				assert.NoError(t, err)
+				assert.Equal(t, tc.expectedVals[0], kataVMIsolation)
+				assert.Equal(t, tc.expectedVals[1], kataRuntime)
+			}
+		})
+	}
+}
+
+func TestIsKataNode(t *testing.T) {
+	testCases := []struct {
+		name        string
+		nodeName    string
+		labels      map[string]string
+		setupClient bool
+		expected    bool
+	}{
+		{
+			name:        "Node does not exist",
+			nodeName:    "",
+			setupClient: true,
+			expected:    false,
+		},
+		{
+			name:        "Node exists but has no kata labels",
+			nodeName:    "test-node",
+			setupClient: true,
+			labels: map[string]string{
+				"some-other-label": "value",
+			},
+			expected: false,
+		},
+		{
+			name:        "Node has kata labels",
+			nodeName:    "test-node",
+			setupClient: true,
+			labels: map[string]string{
+				"kubernetes.azure.com/kata-mshv-vm-isolation": "true",
+			},
+			expected: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.TODO()
+			var clientset kubernetes.Interface
+
+			if tc.setupClient {
+				clientset = fake.NewSimpleClientset()
+			}
+
+			if tc.labels != nil && tc.setupClient {
+				node := &v1api.Node{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:   tc.nodeName,
+						Labels: tc.labels,
+					},
+				}
+				_, err := clientset.CoreV1().Nodes().Create(ctx, node, metav1.CreateOptions{})
+				assert.NoError(t, err)
+			}
+			result := isKataNode(ctx, tc.nodeName, clientset)
+			assert.Equal(t, tc.expected, result)
+		})
+	}
+}

pkg/azurefile/controllerserver.go

@@ -264,10 +264,6 @@ func (d *Driver) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest)
 			accountQuota = int32(value)
 		case tagValueDelimiterField:
 			tagValueDelimiter = v
-		case enableKataCCMountField:
-			if _, err := strconv.ParseBool(v); err != nil {
-				return nil, status.Errorf(codes.InvalidArgument, "invalid %s: %s in storage class", enableKataCCMountField, v)
-			}
 		default:
 			return nil, status.Errorf(codes.InvalidArgument, "invalid parameter %q in storage class", k)
 		}

pkg/azurefile/nodeserver.go

@@ -100,42 +100,40 @@ func (d *Driver) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolu
 			}
 		}
 
-		if d.enableKataCCMount {
-			enableKataCCMount := getValueInMap(context, enableKataCCMountField)
-			if strings.EqualFold(enableKataCCMount, trueValue) && context[podNameField] != "" && context[podNamespaceField] != "" {
-				runtimeClass, err := getRuntimeClassForPodFunc(ctx, d.kubeClient, context[podNameField], context[podNamespaceField])
+		enableKataCCMount := d.isKataNode && d.enableKataCCMount
+		if enableKataCCMount && context[podNameField] != "" && context[podNamespaceField] != "" {
+			runtimeClass, err := getRuntimeClassForPodFunc(ctx, d.kubeClient, context[podNameField], context[podNamespaceField])
+			if err != nil {
+				return nil, status.Errorf(codes.Internal, "failed to get runtime class for pod %s/%s: %v", context[podNamespaceField], context[podNameField], err)
+			}
+			klog.V(2).Infof("NodePublishVolume: volume(%s) mount on %s with runtimeClass %s", volumeID, target, runtimeClass)
+			isConfidentialRuntimeClass, err := isConfidentialRuntimeClassFunc(ctx, d.kubeClient, runtimeClass)
+			if err != nil {
+				return nil, status.Errorf(codes.Internal, "failed to check if runtime class %s is confidential: %v", runtimeClass, err)
+			}
+			if isConfidentialRuntimeClass {
+				klog.V(2).Infof("NodePublishVolume for volume(%s) where runtimeClass is %s", volumeID, runtimeClass)
+				source := req.GetStagingTargetPath()
+				if len(source) == 0 {
+					return nil, status.Error(codes.InvalidArgument, "Staging target not provided")
+				}
+				// Load the mount info from staging area
+				mountInfo, err := d.directVolume.VolumeMountInfo(source)
 				if err != nil {
-					return nil, status.Errorf(codes.Internal, "failed to get runtime class for pod %s/%s: %v", context[podNamespaceField], context[podNameField], err)
+					return nil, status.Errorf(codes.Internal, "failed to load mount info from %s: %v", source, err)
 				}
-				klog.V(2).Infof("NodePublishVolume: volume(%s) mount on %s with runtimeClass %s", volumeID, target, runtimeClass)
-				isConfidentialRuntimeClass, err := isConfidentialRuntimeClassFunc(ctx, d.kubeClient, runtimeClass)
+				if mountInfo == nil {
+					return nil, status.Errorf(codes.Internal, "mount info is nil for volume %s", volumeID)
+				}
+				data, err := json.Marshal(mountInfo)
 				if err != nil {
-					return nil, status.Errorf(codes.Internal, "failed to check if runtime class %s is confidential: %v", runtimeClass, err)
+					return nil, status.Errorf(codes.Internal, "failed to marshal mount info %s: %v", source, err)
 				}
-				if isConfidentialRuntimeClass {
-					klog.V(2).Infof("NodePublishVolume for volume(%s) where runtimeClass %s is kata-cc", volumeID, runtimeClass)
-					source := req.GetStagingTargetPath()
-					if len(source) == 0 {
-						return nil, status.Error(codes.InvalidArgument, "Staging target not provided")
-					}
-					// Load the mount info from staging area
-					mountInfo, err := d.directVolume.VolumeMountInfo(source)
-					if err != nil {
-						return nil, status.Errorf(codes.Internal, "failed to load mount info from %s: %v", source, err)
-					}
-					if mountInfo == nil {
-						return nil, status.Errorf(codes.Internal, "mount info is nil for volume %s", volumeID)
-					}
-					data, err := json.Marshal(mountInfo)
-					if err != nil {
-						return nil, status.Errorf(codes.Internal, "failed to marshal mount info %s: %v", source, err)
-					}
-					if err = d.directVolume.Add(target, string(data)); err != nil {
-						return nil, status.Errorf(codes.Internal, "failed to save mount info %s: %v", target, err)
-					}
-					klog.V(2).Infof("NodePublishVolume: direct volume mount %s at %s successfully", source, target)
-					return &csi.NodePublishVolumeResponse{}, nil
+				if err = d.directVolume.Add(target, string(data)); err != nil {
+					return nil, status.Errorf(codes.Internal, "failed to save mount info %s: %v", target, err)
 				}
+				klog.V(2).Infof("NodePublishVolume: direct volume mount %s at %s successfully", source, target)
+				return &csi.NodePublishVolumeResponse{}, nil
 			}
 		}
 	}
@@ -252,7 +250,7 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 	// don't respect fsType from req.GetVolumeCapability().GetMount().GetFsType()
 	// since it's ext4 by default on Linux
 	var fsType, server, protocol, ephemeralVolMountOptions, storageEndpointSuffix, folderName string
-	var ephemeralVol, enableKataCCMount bool
+	var ephemeralVol bool
 	fileShareNameReplaceMap := map[string]string{}
 
 	mountPermissions := d.mountPermissions
@@ -284,11 +282,6 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 			fileShareNameReplaceMap[pvcNameMetadata] = v
 		case pvNameKey:
 			fileShareNameReplaceMap[pvNameMetadata] = v
-		case enableKataCCMountField:
-			enableKataCCMount, err = strconv.ParseBool(v)
-			if err != nil {
-				return nil, status.Errorf(codes.InvalidArgument, "invalid %s: %s in storage class", enableKataCCMountField, v)
-			}
 		case mountPermissionsField:
 			if v != "" {
 				var err error
@@ -423,9 +416,9 @@ func (d *Driver) NodeStageVolume(ctx context.Context, req *csi.NodeStageVolumeRe
 		}
 		klog.V(2).Infof("volume(%s) mount %s on %s succeeded", volumeID, source, cifsMountPath)
 	}
-
+	enableKataCCMount := d.isKataNode && d.enableKataCCMount
 	// If runtime OS is not windows and protocol is not nfs, save mountInfo.json
-	if d.enableKataCCMount && enableKataCCMount {
+	if enableKataCCMount {
 		if runtime.GOOS != "windows" && protocol != nfs {
 			// Check if mountInfo.json is already present at the targetPath
 			isMountInfoPresent, err := d.directVolume.VolumeMountInfo(cifsMountPath)