Fix CSI and add multi mount

csi/alluxio/controllerserver.go

@@ -36,11 +36,6 @@ type controllerServer struct {
 func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
 	volumeID := sanitizeVolumeID(req.GetName())
 
-	if err := cs.Driver.ValidateControllerServiceRequest(csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
-		glog.V(3).Infof("invalid create volume req: %v", req)
-		return nil, err
-	}
-
 	// Check arguments
 	if len(volumeID) == 0 {
 		return nil, status.Error(codes.InvalidArgument, "Name missing in request")
@@ -77,38 +72,6 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 	return &csi.DeleteVolumeResponse{}, nil
 }
 
-func (cs *controllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
-
-	// Check arguments
-	if len(req.GetVolumeId()) == 0 {
-		return nil, status.Error(codes.InvalidArgument, "Volume ID missing in request")
-	}
-	if req.GetVolumeCapabilities() == nil {
-		return nil, status.Error(codes.InvalidArgument, "Volume capabilities missing in request")
-	}
-
-	// We currently only support RWO
-	supportedAccessMode := &csi.VolumeCapability_AccessMode{
-		Mode: csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
-	}
-
-	for _, cap := range req.VolumeCapabilities {
-		if cap.GetAccessMode().GetMode() != supportedAccessMode.GetMode() {
-			return &csi.ValidateVolumeCapabilitiesResponse{Message: "Only single node writer is supported"}, nil
-		}
-	}
-
-	return &csi.ValidateVolumeCapabilitiesResponse{
-		Confirmed: &csi.ValidateVolumeCapabilitiesResponse_Confirmed{
-			VolumeCapabilities: []*csi.VolumeCapability{
-				{
-					AccessMode: supportedAccessMode,
-				},
-			},
-		},
-	}, nil
-}
-
 func (cs *controllerServer) ControllerExpandVolume(ctx context.Context, req *csi.ControllerExpandVolumeRequest) (*csi.ControllerExpandVolumeResponse, error) {
 	return &csi.ControllerExpandVolumeResponse{}, status.Error(codes.Unimplemented, "ControllerExpandVolume is not implemented")
 }

csi/alluxio/nodeserver.go

@@ -40,6 +40,8 @@ type nodeServer struct {
 	mutex  sync.Mutex
 }
 
+const alluxioFuseHostPath = "/mnt/alluxio/fuse"
+
 /*
  * When there is no app pod using the pv, the first app pod using the pv would trigger NodeStageVolume().
  * Only after a successful return, NodePublishVolume() is called.
@@ -55,8 +57,7 @@ type nodeServer struct {
 
 func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
 	targetPath := req.GetTargetPath()
-	stagingPath := req.GetStagingTargetPath()
-
+	stagingPath := fmt.Sprintf("%s-%s", alluxioFuseHostPath, req.VolumeId)
 	notMnt, err := ensureMountPoint(targetPath)
 	if err != nil {
 		glog.V(3).Infof("Error checking mount point: %+v.", err)
@@ -109,7 +110,8 @@ func (ns *nodeServer) NodeStageVolume(ctx context.Context, req *csi.NodeStageVol
 		glog.V(3).Infof("Error creating CSI Fuse pod. %+v", err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
-	if err := checkIfMountPointReady(req.GetStagingTargetPath()); err != nil {
+	stagingPath := fmt.Sprintf("%s-%s", alluxioFuseHostPath, req.VolumeId)
+	if err := checkIfMountPointReady(stagingPath); err != nil {
 		glog.V(3).Infof("Mount point is not ready, or error occurs. %+v", err)
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -126,26 +128,20 @@ func getAndCompleteFusePodObj(ns *nodeServer, req *csi.NodeStageVolumeRequest) (
 		return nil, errors.Wrap(err, "Error getting Fuse pod object from template.")
 	}
 
-	// Append extra information to pod name for uniqueness but not exceed maximum
-	csiFusePodObj.Name = getFusePodName(alluxioNamespacedName.Name, ns.nodeId, req.GetVolumeId())[:64]
+	// Append extra information to pod name for uniqueness
+	csiFusePodObj.Name = getFusePodName(alluxioNamespacedName.Name, ns.nodeId, req.GetVolumeId())
 
 	csiFusePodObj.Namespace = alluxioNamespacedName.Namespace
 
 	// Set node name for scheduling
 	csiFusePodObj.Spec.NodeName = ns.nodeId
 
-	// Set fuse mount point
-	csiFusePodObj.Spec.Containers[0].Args[2] = req.GetStagingTargetPath()
+	// Use unique mount path
+	stagingPath := fmt.Sprintf("%s-%s", alluxioFuseHostPath, req.VolumeId)
+	csiFusePodObj.Spec.InitContainers[0].Command[2] = stagingPath
+	csiFusePodObj.Spec.Containers[0].Args[0] = strings.ReplaceAll(csiFusePodObj.Spec.Containers[0].Args[0], alluxioFuseHostPath, stagingPath)
+	csiFusePodObj.Spec.Containers[0].Lifecycle.PreStop.Exec.Command[2] = stagingPath
 
-	// Set pre-stop command (umount) in pod lifecycle
-	lifecycle := &v1.Lifecycle{
-		PreStop: &v1.Handler{
-			Exec: &v1.ExecAction{
-				Command: []string{"/opt/alluxio/integration/fuse/bin/alluxio-fuse", "unmount", req.GetStagingTargetPath()},
-			},
-		},
-	}
-	csiFusePodObj.Spec.Containers[0].Lifecycle = lifecycle
 	return csiFusePodObj, nil
 }
 
@@ -187,12 +183,6 @@ func (ns *nodeServer) NodeUnstageVolume(ctx context.Context, req *csi.NodeUnstag
 	podName := getFusePodName(alluxioNamespacedName.Name, ns.nodeId, req.GetVolumeId())
 	if err := ns.client.CoreV1().Pods(alluxioNamespacedName.Namespace).Delete(podName, &metav1.DeleteOptions{}); err != nil {
 		if strings.Contains(err.Error(), "not found") {
-			// Pod not found. Try to clean up the mount point.
-			command := exec.Command("umount", req.GetStagingTargetPath())
-			_, err := command.CombinedOutput()
-			if err != nil {
-				glog.V(3).Infof("Error running command %v: %+v", command, err)
-			}
 			return &csi.NodeUnstageVolumeResponse{}, nil
 		}
 		glog.V(3).Infof("Error deleting pod with name %v. %+v.", podName, err)
@@ -268,5 +258,9 @@ func getFusePodObj(ns *nodeServer, alluxioNamespacedName types.NamespacedName) (
 
 func getFusePodName(clusterName, nodeId, volumeId string) string {
 	volumeIdParts := strings.Split(volumeId, "-")
-	return strings.Join([]string{clusterName, nodeId, volumeIdParts[len(volumeIdParts)-1]}, "-")
+	fusePodName := strings.Join([]string{clusterName, "fuse", nodeId, volumeIdParts[len(volumeIdParts)-1]}, "-")
+	if len(fusePodName) > 64 {
+		fusePodName = fusePodName[:64]
+	}
+	return fusePodName
 }

deploy/charts/alluxio-csi/templates/nodeplugin.yaml

@@ -12,6 +12,7 @@
 {{- $name := include "alluxio-csi.name" . }}
 {{- $fullName := include "alluxio-csi.fullname" . }}
 {{- $chart := include "alluxio-csi.chart" . }}
+{{- $hostMountPath := "/mnt/alluxio" }}
 
 kind: DaemonSet
 apiVersion: apps/v1
@@ -120,6 +121,9 @@ spec:
           volumeMounts:
             - name: plugin-dir
               mountPath: /plugin
+            - name: alluxio-fuse-mount
+              mountPath: {{ $hostMountPath }}
+              mountPropagation: "HostToContainer"
             - name: pods-mount-dir
               mountPath: /var/lib/kubelet
               mountPropagation: "Bidirectional"
@@ -128,6 +132,10 @@ spec:
           hostPath:
             path: /var/lib/kubelet/plugins/csi-alluxio-plugin
             type: DirectoryOrCreate
+        - name: alluxio-fuse-mount
+          hostPath:
+            path: {{ $hostMountPath }}
+            type: DirectoryOrCreate
         - name: pods-mount-dir
           hostPath:
             path: /var/lib/kubelet

deploy/charts/alluxio/templates/csi/csi-fuse.yaml

@@ -12,7 +12,10 @@
 {{- $name := include "alluxio.name" . }}
 {{- $fullName := include "alluxio.fullname" . }}
 {{- $chart := include "alluxio.chart" . }}
+{{- $hostMountPath := include "alluxio.mount.basePath" "" }}
 {{- $alluxioFuseMountPoint := include "alluxio.mount.basePath" "/fuse" }}
+{{- $alluxioFuseLogDir := include "alluxio.basePath" "/logs"}}
+{{- $alluxioFuseLogVolumeName := include "alluxio.getVolumeName" (dict "prefix" $fullName "component" "fuse-log") }}
 
 ---
 kind: ConfigMap
@@ -37,63 +40,71 @@ data:
         role: alluxio-fuse
     spec:
       securityContext:
-        runAsUser: 0 # required for mounting to csi designated path
-        runAsGroup: 0
-        fsGroup: 0
+        runAsUser: {{ .Values.fuse.user }}
+        runAsGroup: {{ .Values.fuse.group }}
+        fsGroup: {{ .Values.fsGroup }}
+      hostNetwork: {{ .Values.hostNetwork }}
+      dnsPolicy: {{ .Values.dnsPolicy | default (.Values.hostNetwork | ternary "ClusterFirstWithHostNet" "ClusterFirst") }}
       {{- if .Values.serviceAccountName }}
       serviceAccountName: {{ .Values.serviceAccountName }}
       {{- end }}
       {{- if .Values.imagePullSecrets }}
     {{ include "alluxio.imagePullSecrets" . | indent 2 }}
       {{- end}}
       initContainers:
+      - name: create-alluxio-fuse-dir
+        image: {{ .Values.image }}:{{ .Values.imageTag }}
+        imagePullPolicy: {{ .Values.imagePullPolicy }}
+        command: [ "mkdir", "-p", {{ $alluxioFuseMountPoint }}]
+        volumeMounts:
+          - name: alluxio-fuse-mount
+            mountPath: {{ $hostMountPath }}
+      {{- if .Values.master.enabled }}
       - name: wait-master
         image: {{ .Values.image }}:{{ .Values.imageTag }}
         command: ["/bin/sh", "-c"]
         args:
           - until nslookup {{ $fullName }}-master-0;
             do sleep 2;
             done
-        volumeMounts:
-        - name: {{ $fullName }}-alluxio-conf
-          mountPath: /opt/alluxio/conf
+      {{- end }}
       containers:
         - name: alluxio-fuse
           image: {{ .Values.image }}:{{ .Values.imageTag }}
           imagePullPolicy: {{ .Values.imagePullPolicy }}
           {{- if .Values.fuse.resources }}
-          resources:
-            {{- if .Values.fuse.resources.limits }}
-            limits:
-              cpu: {{ .Values.fuse.resources.limits.cpu }}
-              memory: {{ .Values.fuse.resources.limits.memory }}
-            {{- end }}
-            {{- if .Values.fuse.resources.requests }}
-              cpu: {{ .Values.fuse.resources.requests.cpu }}
-              memory: {{ .Values.fuse.resources.requests.memory }}
-            {{- end }}
+{{ include "alluxio.resources" .Values.fuse.resources | indent 10 }}
           {{- end }}
-          command: [ "/entrypoint.sh" ]
+          command: ["/bin/sh", "-c"]
           args:
-            - fuse
-            - {{ required "The path of the dataset must be set." .Values.dataset.path }}
-            - {{ $alluxioFuseMountPoint }}
-            {{- range .Values.fuse.mountOptions }}
-            - -o {{ . }}
-            {{- end }}
+            - umount -l {{ $alluxioFuseMountPoint }};
+              {{- $mountTableSource := get .Values.properties "alluxio.mount.table.source" }}
+              {{- if or (eq $mountTableSource "ETCD") (eq $mountTableSource "STATIC_FILE") }}
+              /entrypoint.sh fuse {{ $alluxioFuseMountPoint }} {{- range .Values.fuse.mountOptions }} -o {{ . }} {{- end }}
+              {{- else }}
+              /entrypoint.sh fuse {{ required "The path of the dataset must be set." .Values.dataset.path }} {{ $alluxioFuseMountPoint }} {{- range .Values.fuse.mountOptions }} -o {{ . }} {{- end }}
+              {{- end }}
           env:
           {{- range $key, $value := .Values.fuse.env }}
           - name: "{{ $key }}"
             value: "{{ $value }}"
           {{- end }}
           securityContext:
             privileged: true # required by bidirectional mount
+          lifecycle:
+            preStop:
+              exec:
+                command: ["fusermount", "-u", {{ $alluxioFuseMountPoint }}]
           volumeMounts:
             - name: {{ $fullName }}-alluxio-conf
               mountPath: /opt/alluxio/conf
-            - name: pods-mount-dir
-              mountPath: /var/lib/kubelet
-              mountPropagation: "Bidirectional"
+            - name: alluxio-fuse-mount
+              mountPath: {{ $hostMountPath }}
+              mountPropagation: Bidirectional
+            {{- if .Values.hostPathForLogging }}
+            - name: {{ $alluxioFuseLogVolumeName }}
+              mountPath: {{ $alluxioFuseLogDir }}
+            {{- end }}
             {{- if .Values.secrets }}
     {{- include "alluxio.volumeMounts" (dict "volumeMounts" .Values.secrets.fuse "readOnly" true) | indent 8 }}
             {{- end }}
@@ -105,13 +116,19 @@ data:
             {{- end }}
       restartPolicy: Always
       volumes:
-        - name: pods-mount-dir
+        - name: alluxio-fuse-mount
           hostPath:
-            path: /var/lib/kubelet
-            type: Directory
+            path: {{ $hostMountPath }}
+            type: DirectoryOrCreate
         - name: {{ $fullName }}-alluxio-conf
           configMap:
             name: {{ $fullName }}-alluxio-conf
+        {{- if .Values.hostPathForLogging }}
+        - name: {{ $alluxioFuseLogVolumeName }}
+          hostPath:
+            path: {{ .Values.fuse.hostPathForLogs }}
+            type: DirectoryOrCreate
+        {{- end }}
         {{- if .Values.secrets }}
     {{- include "alluxio.secretVolumes" .Values.secrets.fuse | indent 4 }}
         {{- end }}

deploy/charts/alluxio/templates/csi/pvc.yaml

@@ -16,8 +16,6 @@ kind: PersistentVolumeClaim
 metadata:
   name: {{ $fullName }}-alluxio-csi-fuse-pvc
 spec:
-  accessModes:
-    - ReadWriteOnce
   storageClassName: {{ $fullName }}-csi-storage-class
   resources:
     requests:

deploy/charts/alluxio/templates/csi/storageClass.yaml

@@ -16,7 +16,7 @@ kind: StorageClass
 metadata:
   name: {{ $fullName }}-csi-storage-class
 provisioner: alluxio
-volumeBindingMode: Immediate
+volumeBindingMode: WaitForFirstConsumer
 parameters:
   alluxioClusterName: {{ $fullName }}
   alluxioClusterNamespace: {{ .Release.Namespace }}

deploy/charts/alluxio/templates/fuse/daemonset.yaml

@@ -119,9 +119,6 @@ spec:
             - until nslookup {{ $fullName }}-master-0;
               do sleep 2;
               done
-          volumeMounts:
-          - name: {{ $fullName }}-alluxio-conf
-            mountPath: /opt/alluxio/conf
         {{- end }}
       containers:
         - name: alluxio-fuse