WIP cephfs CSI plugin

Makefile

@@ -14,10 +14,13 @@
 
 .PHONY: all rbdplugin
 
-IMAGE_NAME=quay.io/cephcsi/rbdplugin
-IMAGE_VERSION=v0.2.0
+RBD_IMAGE_NAME=quay.io/cephcsi/rbdplugin
+RBD_IMAGE_VERSION=v0.2.0
 
-all: rbdplugin
+CEPHFS_IMAGE_NAME=cephfsplugin
+CEPHFS_IMAGE_VERSION=latest
+
+all: rbdplugin cephfsplugin
 
 test:
 	go test github.com/ceph/ceph-csi/pkg/... -cover
@@ -27,13 +30,23 @@ rbdplugin:
 	if [ ! -d ./vendor ]; then dep ensure; fi
 	CGO_ENABLED=0 GOOS=linux go build -a -ldflags '-extldflags "-static"' -o  _output/rbdplugin ./rbd
 
-container: rbdplugin 
-	cp _output/rbdplugin  deploy/docker
-	docker build -t $(IMAGE_NAME):$(IMAGE_VERSION) deploy/docker
+rbdplugin-container: rbdplugin 
+	cp _output/rbdplugin  deploy/rbd/docker
+	docker build -t $(IMAGE_NAME):$(IMAGE_VERSION) deploy/rbd/docker
+
+cephfsplugin:
+	if [ ! -d ./vendor ]; then dep ensure; fi
+	CGO_ENABLED=0 GOOS=linux go build -a -ldflags '-extldflags "-static"' -o  _output/cephfsplugin ./cephfs
+
+cephfsplugin-container: cephfsplugin 
+	cp _output/cephfsplugin deploy/cephfs/docker
+	docker build -t $(CEPHFS_IMAGE_NAME):$(CEPHFS_IMAGE_VERSION) deploy/cephfs/docker
+
+push-container: rbdplugin-container
+	docker push $(RBD_IMAGE_NAME):$(RBD_IMAGE_VERSION)
 
-push-container: container
-	docker push $(IMAGE_NAME):$(IMAGE_VERSION)
 clean:
 	go clean -r -x
-	rm -f deploy/docker/rbdplugin
+	rm -f deploy/rbd/docker/rbdplugin
+	rm -f deploy/cephfs/docker/rbdplugin
 	-rm -rf _output

cephfs/main.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2018 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 main
+
+import (
+	"flag"
+	"os"
+	"path"
+
+	"github.com/ceph/ceph-csi/pkg/cephfs"
+	"github.com/golang/glog"
+)
+
+func init() {
+	flag.Set("logtostderr", "true")
+}
+
+var (
+	endpoint   = flag.String("endpoint", "unix://tmp/csi.sock", "CSI endpoint")
+	driverName = flag.String("drivername", "cephfsplugin", "name of the driver")
+	nodeID     = flag.String("nodeid", "", "node id")
+)
+
+func main() {
+	flag.Parse()
+
+	if err := createPersistentStorage(path.Join(cephfs.PluginFolder, "controller")); err != nil {
+		glog.Errorf("failed to create persisten storage for controller %v", err)
+		os.Exit(1)
+	}
+
+	if err := createPersistentStorage(path.Join(cephfs.PluginFolder, "node")); err != nil {
+		glog.Errorf("failed to create persisten storage for node %v", err)
+		os.Exit(1)
+	}
+
+	driver := cephfs.NewCephFSDriver()
+	driver.Run(*driverName, *nodeID, *endpoint)
+
+	os.Exit(0)
+}
+
+func createPersistentStorage(persistentStoragePath string) error {
+	return os.MkdirAll(persistentStoragePath, os.FileMode(0755))
+}

deploy/cephfs/docker/Dockerfile

@@ -0,0 +1,13 @@
+FROM ubuntu:16.04
+LABEL maintainers="Kubernetes Authors"
+LABEL description="CephFS CSI Plugin"
+
+ENV CEPH_VERSION "luminous"
+
+RUN apt-get update && \
+    apt-get install -y ceph-fuse attr && \
+     apt-get autoremove
+
+COPY cephfsplugin /cephfsplugin
+RUN chmod +x /cephfsplugin
+ENTRYPOINT ["/cephfsplugin"]

deploy/cephfs/kubernetes/cephfs-storage-class.yaml

@@ -0,0 +1,8 @@
+apiVersion: storage.k8s.io/v1
+kind: StorageClass
+metadata:
+  name: cephfs
+provisioner: cephfsplugin
+parameters:
+  provisionRoot: /cephfs
+reclaimPolicy: Delete 

deploy/cephfs/kubernetes/cephfsplugin.yaml

@@ -0,0 +1,137 @@
+# This YAML defines all API objects to create RBAC roles for csi node plugin.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: csi-nodeplugin
+
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-nodeplugin
+rules:
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get", "list", "update"]
+  - apiGroups: [""]
+    resources: ["secrets"]
+    verbs: ["get", "list"]
+  - apiGroups: [""]
+    resources: ["namespaces"]
+    verbs: ["get", "list"]
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch", "update"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["volumeattachments"]
+    verbs: ["get", "list", "watch", "update"]
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-nodeplugin
+subjects:
+  - kind: ServiceAccount
+    name: csi-nodeplugin
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: csi-nodeplugin
+  apiGroup: rbac.authorization.k8s.io          
+
+---
+# This YAML file contains driver-registrar & csi driver nodeplugin API objects,
+# which are necessary to run csi nodeplugin for cephfs.
+
+kind: DaemonSet
+apiVersion: apps/v1beta2
+metadata:
+  name: csi-nodeplugin-cephfsplugin
+spec:
+  selector:
+    matchLabels:
+      app: csi-nodeplugin-cephfsplugin
+  template:
+    metadata:
+      labels:
+        app: csi-nodeplugin-cephfsplugin
+    spec:
+      serviceAccount: csi-nodeplugin
+      hostNetwork: true
+      containers:
+        - name: driver-registrar
+          image: quay.io/k8scsi/driver-registrar:latest
+          args:
+            - "--v=5"
+            - "--csi-address=$(ADDRESS)"
+          env:
+            - name: ADDRESS
+              value: /var/lib/kubelet/plugins/cephfsplugin/csi.sock
+            - name: KUBE_NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /var/lib/kubelet/plugins/cephfsplugin
+        - name: cephfsplugin 
+          securityContext:
+            privileged: true
+            capabilities:
+              add: ["SYS_ADMIN"]
+            allowPrivilegeEscalation: true
+          image: csi_images/cephfsplugin:latest
+          args :
+            - "--nodeid=$(NODE_ID)"
+            - "--endpoint=$(CSI_ENDPOINT)"
+            - "--v=5"
+            - "--drivername=cephfsplugin"
+          env:
+            - name: NODE_ID
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
+            - name: CSI_ENDPOINT
+              value: unix://var/lib/kubelet/plugins/cephfsplugin/csi.sock
+          imagePullPolicy: "IfNotPresent"
+          volumeMounts:
+            - name: plugin-dir
+              mountPath: /var/lib/kubelet/plugins/cephfsplugin
+            - name: pods-mount-dir
+              mountPath: /var/lib/kubelet/pods
+              mountPropagation: "Bidirectional"
+            - mountPath: /dev
+              name: host-dev
+            - mountPath: /sys
+              name: host-sys
+            - mountPath: /lib/modules
+              name: lib-modules
+              readOnly: true
+            - name: cephfs-config
+              mountPath: /etc/ceph
+      volumes:
+        - name: plugin-dir
+          hostPath:
+            path: /var/lib/kubelet/plugins/cephfsplugin
+            type: DirectoryOrCreate
+        - name: pods-mount-dir
+          hostPath:
+            path: /var/lib/kubelet/pods
+            type: Directory
+        - name: socket-dir
+          hostPath:
+            path: /var/lib/kubelet/plugins/cephfsplugin
+            type: DirectoryOrCreate
+        - name: host-dev
+          hostPath:
+            path: /dev
+        - name: host-sys
+          hostPath:
+            path: /sys
+        - name: lib-modules
+          hostPath:
+            path: /lib/modules
+        - name: cephfs-config
+          hostPath:
+            path: /etc/ceph

deploy/cephfs/kubernetes/csi-attacher.yaml

@@ -0,0 +1,84 @@
+# This YAML file contains RBAC API objects,
+# which are necessary to run external csi attacher for cinder.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: csi-attacher
+
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: external-attacher-runner
+rules:
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch", "update"]
+  - apiGroups: [""]
+    resources: ["nodes"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["volumeattachments"]
+    verbs: ["get", "list", "watch", "update"]
+
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-attacher-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-attacher
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: external-attacher-runner
+  apiGroup: rbac.authorization.k8s.io
+---
+
+kind: Service
+apiVersion: v1
+metadata:
+  name: csi-attacher
+  labels:
+    app: csi-attacher
+spec:
+  selector:
+    app: csi-attacher
+  ports:
+    - name: dummy
+      port: 12345
+
+---
+kind: StatefulSet
+apiVersion: apps/v1beta1
+metadata:
+  name: csi-attacher
+spec:
+  serviceName: "csi-attacher"
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: csi-attacher
+    spec:
+      serviceAccount: csi-attacher
+      containers:
+        - name: csi-attacher
+          image: quay.io/k8scsi/csi-attacher:latest
+          args:
+            - "--v=5"
+            - "--csi-address=$(ADDRESS)"
+          env:
+            - name: ADDRESS
+              value: /var/lib/kubelet/plugins/cephfsplugin/csi.sock
+          imagePullPolicy: "IfNotPresent"
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /var/lib/kubelet/plugins/cephfsplugin
+      volumes:
+        - name: socket-dir
+          hostPath:
+            path: /var/lib/kubelet/plugins/cephfsplugin
+            type: DirectoryOrCreate

deploy/cephfs/kubernetes/csi-provisioner.yaml

@@ -0,0 +1,95 @@
+# This YAML file contains all API objects that are necessary to run external
+# CSI provisioner.
+#
+# In production, this needs to be in separate files, e.g. service account and
+# role and role binding needs to be created once, while stateful set may
+# require some tuning.
+#
+# In addition, mock CSI driver is hardcoded as the CSI driver.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: csi-provisioner
+
+---
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: external-provisioner-runner
+rules:
+  - apiGroups: [""]
+    resources: ["persistentvolumes"]
+    verbs: ["get", "list", "watch", "create", "delete"]
+  - apiGroups: [""]
+    resources: ["persistentvolumeclaims"]
+    verbs: ["get", "list", "watch", "update"]
+  - apiGroups: ["storage.k8s.io"]
+    resources: ["storageclasses"]
+    verbs: ["get", "list", "watch"]
+  - apiGroups: [""]
+    resources: ["events"]
+    verbs: ["list", "watch", "create", "update", "patch"]
+    
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: csi-provisioner-role
+subjects:
+  - kind: ServiceAccount
+    name: csi-provisioner
+    namespace: default
+roleRef:
+  kind: ClusterRole
+  name: external-provisioner-runner
+  apiGroup: rbac.authorization.k8s.io
+
+---
+kind: Service
+apiVersion: v1
+metadata:
+  name: csi-provisioner
+  labels:
+    app: csi-provisioner
+spec:
+  selector:
+    app: csi-provisioner
+  ports:
+    - name: dummy
+      port: 12345
+
+---
+kind: StatefulSet
+apiVersion: apps/v1beta1
+metadata:
+  name: csi-provisioner
+spec:
+  serviceName: "csi-provisioner"
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        app: csi-provisioner
+    spec:
+      serviceAccount: csi-provisioner
+      containers:
+        - name: csi-provisioner
+          image: quay.io/k8scsi/csi-provisioner:latest
+          # image: quay.io/k8scsi/csi-provisioner:latest
+          args:
+            - "--provisioner=cephfsplugin"
+            - "--csi-address=$(ADDRESS)"
+            - "--v=5"
+          env:
+            - name: ADDRESS
+              value: /var/lib/kubelet/plugins/cephfsplugin/csi.sock
+          imagePullPolicy: "IfNotPresent"
+          volumeMounts:
+            - name: socket-dir
+              mountPath: /var/lib/kubelet/plugins/cephfsplugin
+      volumes:
+        - name: socket-dir
+          hostPath:
+            path: /var/lib/kubelet/plugins/cephfsplugin
+            type: DirectoryOrCreate

deploy/cephfs/kubernetes/deploy-csi.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+objects=(cephfs-storage-class cephfsplugin csi-attacher csi-provisioner)
+
+for obj in ${objects[@]}; do
+	kubectl create -f "./$obj.yaml"
+done

deploy/cephfs/kubernetes/deploy-pod.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+
+kubectl create -f ./pvc.yaml
+kubectl create -f ./pod.yaml

deploy/cephfs/kubernetes/pod.yaml

@@ -0,0 +1,17 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: web-server
+spec:
+  containers:
+   - name: web-server
+     image: nginx
+     volumeMounts:
+       - mountPath: /var/lib/www/html
+         name: mypvc
+  volumes:
+   - name: mypvc
+     persistentVolumeClaim:
+       claimName: cephfs-pvc
+       readOnly: false
+

deploy/cephfs/kubernetes/pvc.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: cephfs-pvc
+spec:
+  accessModes:
+  - ReadWriteOnce
+  resources:
+    requests:
+      storage: 5Gi
+  storageClassName: cephfs

deploy/cephfs/kubernetes/teardown-csi.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+objects=(cephfsplugin csi-provisioner csi-attacher cephfs-storage-class)
+
+for obj in ${objects[@]}; do
+	kubectl delete -f "./$obj.yaml"
+done

deploy/cephfs/kubernetes/teardown-pod.sh

@@ -0,0 +1,4 @@
+#!/bin/sh
+
+kubectl delete -f ./pod.yaml
+kubectl delete -f ./pvc.yaml

pkg/cephfs/cephfs.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"github.com/golang/glog"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/drivers/pkg/csi-common"
+)
+
+const (
+	PluginFolder = "/var/lib/kubelet/plugins/cephfsplugin"
+)
+
+type cephfsDriver struct {
+	driver *csicommon.CSIDriver
+
+	ids *identityServer
+	ns  *nodeServer
+	cs  *controllerServer
+
+	caps   []*csi.VolumeCapability_AccessMode
+	cscaps []*csi.ControllerServiceCapability
+}
+
+var (
+	provisionRoot = "/cephfs"
+
+	driver  *cephfsDriver
+	version = csi.Version{
+		Minor: 1,
+	}
+)
+
+func GetSupportedVersions() []*csi.Version {
+	return []*csi.Version{&version}
+}
+
+func NewCephFSDriver() *cephfsDriver {
+	return &cephfsDriver{}
+}
+
+func NewIdentityServer(d *csicommon.CSIDriver) *identityServer {
+	return &identityServer{
+		DefaultIdentityServer: csicommon.NewDefaultIdentityServer(d),
+	}
+}
+
+func NewControllerServer(d *csicommon.CSIDriver) *controllerServer {
+	return &controllerServer{
+		DefaultControllerServer: csicommon.NewDefaultControllerServer(d),
+	}
+}
+
+func NewNodeServer(d *csicommon.CSIDriver) *nodeServer {
+	return &nodeServer{
+		DefaultNodeServer: csicommon.NewDefaultNodeServer(d),
+	}
+}
+
+func (fs *cephfsDriver) Run(driverName, nodeId, endpoint string) {
+	glog.Infof("Driver: %v version: %v", driverName, GetVersionString(&version))
+
+	// Initialize default library driver
+
+	fs.driver = csicommon.NewCSIDriver(driverName, &version, GetSupportedVersions(), nodeId)
+	if fs.driver == nil {
+		glog.Fatalln("Failed to initialize CSI driver")
+	}
+
+	fs.driver.AddControllerServiceCapabilities([]csi.ControllerServiceCapability_RPC_Type{
+		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
+		csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
+	})
+
+	fs.driver.AddVolumeCapabilityAccessModes([]csi.VolumeCapability_AccessMode_Mode{
+		csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
+	})
+
+	// Create gRPC servers
+
+	fs.ids = NewIdentityServer(fs.driver)
+	fs.ns = NewNodeServer(fs.driver)
+	fs.cs = NewControllerServer(fs.driver)
+
+	server := csicommon.NewNonBlockingGRPCServer()
+	server.Start(endpoint, fs.ids, fs.cs, fs.ns)
+	server.Wait()
+}

pkg/cephfs/controllerserver.go

@@ -0,0 +1,158 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"fmt"
+	"os"
+	"path"
+
+	"github.com/golang/glog"
+	"golang.org/x/net/context"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/drivers/pkg/csi-common"
+)
+
+type controllerServer struct {
+	*csicommon.DefaultControllerServer
+}
+
+const (
+	oneGB = 1073741824
+)
+
+func GetVersionString(v *csi.Version) string {
+	return fmt.Sprintf("%d.%d.%d", v.GetMajor(), v.GetMinor(), v.GetPatch())
+}
+
+func (cs *controllerServer) validateRequest(v *csi.Version) error {
+	if v == nil {
+		return status.Error(codes.InvalidArgument, "Version missing in request")
+	}
+
+	return cs.Driver.ValidateControllerServiceRequest(v, csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME)
+}
+
+func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVolumeRequest) (*csi.CreateVolumeResponse, error) {
+	if err := cs.validateRequest(req.Version); err != nil {
+		glog.Warningf("invalid create volume request: %v", req)
+		return nil, err
+	}
+
+	// Configuration
+
+	volOptions, err := newVolumeOptions(req.GetParameters())
+	if err != nil {
+		return nil, err
+	}
+
+	volId := newVolumeIdentifier(volOptions, req)
+	volSz := int64(oneGB)
+
+	if req.GetCapacityRange() != nil {
+		volSz = int64(req.GetCapacityRange().GetRequiredBytes())
+	}
+
+	if err := createMountPoint(provisionRoot); err != nil {
+		glog.Errorf("failed to create provision root at %s: %v", provisionRoot, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	// Exec ceph-fuse only if cephfs has not been not mounted yet
+
+	isMnt, err := isMountPoint(provisionRoot)
+
+	if err != nil {
+		glog.Errorf("stat failed: %v", err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if !isMnt {
+		if err = mountFuse(provisionRoot); err != nil {
+			glog.Error(err)
+			return nil, status.Error(codes.Internal, err.Error())
+		}
+	}
+
+	// Create a new directory inside the provision root for bind-mounting done by NodePublishVolume
+
+	volPath := path.Join(provisionRoot, volId.id)
+	if err := os.Mkdir(volPath, 0750); err != nil {
+		glog.Errorf("failed to create volume %s: %v", volPath, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	// Set attributes & quotas
+
+	if err = setVolAttributes(volPath, volSz); err != nil {
+		glog.Errorf("failed to set attributes for volume %s: %v", volPath, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	glog.V(4).Infof("cephfs: created volume %s", volPath)
+
+	return &csi.CreateVolumeResponse{
+		VolumeInfo: &csi.VolumeInfo{
+			Id:            volId.id,
+			CapacityBytes: uint64(volSz),
+			Attributes:    req.GetParameters(),
+		},
+	}, nil
+}
+
+func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
+	if err := cs.validateRequest(req.Version); err != nil {
+		glog.Warningf("invalid delete volume request: %v", req)
+		return nil, err
+	}
+
+	volId := req.GetVolumeId()
+	volPath := path.Join(provisionRoot, volId)
+
+	glog.V(4).Infof("deleting volume %s", volPath)
+
+	if err := deleteVolumePath(volPath); err != nil {
+		glog.Errorf("failed to delete volume %s: %v", volPath, err)
+		return nil, err
+	}
+
+	return &csi.DeleteVolumeResponse{}, nil
+}
+
+func (cs *controllerServer) ValidateVolumeCapabilities(ctx context.Context, req *csi.ValidateVolumeCapabilitiesRequest) (*csi.ValidateVolumeCapabilitiesResponse, error) {
+	res := &csi.ValidateVolumeCapabilitiesResponse{}
+
+	for _, capability := range req.VolumeCapabilities {
+		if capability.GetAccessMode().GetMode() != csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER {
+			return res, nil
+		}
+	}
+
+	res.Supported = true
+	return res, nil
+}
+
+func (cs *controllerServer) ControllerPublishVolume(ctx context.Context, req *csi.ControllerPublishVolumeRequest) (*csi.ControllerPublishVolumeResponse, error) {
+	return &csi.ControllerPublishVolumeResponse{}, nil
+}
+
+func (cs *controllerServer) ControllerUnpublishVolume(ctx context.Context, req *csi.ControllerUnpublishVolumeRequest) (*csi.ControllerUnpublishVolumeResponse, error) {
+	return &csi.ControllerUnpublishVolumeResponse{}, nil
+}

pkg/cephfs/identityserver.go

@@ -0,0 +1,25 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"github.com/kubernetes-csi/drivers/pkg/csi-common"
+)
+
+type identityServer struct {
+	*csicommon.DefaultIdentityServer
+}

pkg/cephfs/nodeserver.go

@@ -0,0 +1,144 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"context"
+	"path"
+
+	"github.com/golang/glog"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/kubernetes-csi/drivers/pkg/csi-common"
+	"k8s.io/kubernetes/pkg/util/keymutex"
+	"k8s.io/kubernetes/pkg/util/mount"
+)
+
+type nodeServer struct {
+	*csicommon.DefaultNodeServer
+}
+
+var nsMtx = keymutex.NewKeyMutex()
+
+func validateNodePublishVolumeRequest(req *csi.NodePublishVolumeRequest) error {
+	if req.GetVersion() == nil {
+		return status.Error(codes.InvalidArgument, "Version missing in request")
+	}
+
+	if req.GetVolumeCapability() == nil {
+		return status.Error(codes.InvalidArgument, "Volume capability missing in request")
+	}
+
+	if req.GetVolumeId() == "" {
+		return status.Error(codes.InvalidArgument, "Volume ID missing in request")
+	}
+
+	if req.GetTargetPath() == "" {
+		return status.Error(codes.InvalidArgument, "Target path missing in request")
+	}
+
+	return nil
+}
+
+func validateNodeUnpublishVolumeRequest(req *csi.NodeUnpublishVolumeRequest) error {
+	if req.GetVersion() == nil {
+		return status.Error(codes.InvalidArgument, "Version missing in request")
+	}
+
+	if req.GetVolumeId() == "" {
+		return status.Error(codes.InvalidArgument, "Volume ID missing in request")
+	}
+
+	if req.GetTargetPath() == "" {
+		return status.Error(codes.InvalidArgument, "Target path missing in request")
+	}
+
+	return nil
+}
+
+func (ns *nodeServer) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) {
+	if err := validateNodePublishVolumeRequest(req); err != nil {
+		return nil, err
+	}
+
+	// Configuration
+
+	volId := req.GetVolumeId()
+	targetPath := req.GetTargetPath()
+
+	if err := tryLock(volId, nsMtx, "NodeServer"); err != nil {
+		return nil, err
+	}
+	defer nsMtx.UnlockKey(volId)
+
+	if err := createMountPoint(targetPath); err != nil {
+		glog.Errorf("failed to create mount point at %s: %v", targetPath, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	// Check if the volume is already mounted
+
+	isMnt, err := isMountPoint(targetPath)
+
+	if err != nil {
+		glog.Errorf("stat failed: %v", err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	if isMnt {
+		return &csi.NodePublishVolumeResponse{}, nil
+	}
+
+	// It's not, do the bind-mount now
+
+	options := []string{"bind"}
+	if req.GetReadonly() {
+		options = append(options, "ro")
+	}
+
+	volPath := path.Join(provisionRoot, req.GetVolumeId())
+	if err := mount.New("").Mount(volPath, targetPath, "", options); err != nil {
+		glog.Errorf("bind-mounting %s to %s failed: %v", volPath, targetPath, err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	glog.V(4).Infof("cephfs: volume %s successfuly mounted to %s", volPath, targetPath)
+
+	return &csi.NodePublishVolumeResponse{}, nil
+}
+
+func (ns *nodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) {
+	if err := validateNodeUnpublishVolumeRequest(req); err != nil {
+		return nil, err
+	}
+
+	volId := req.GetVolumeId()
+	targetPath := req.GetTargetPath()
+
+	if err := tryLock(volId, nsMtx, "NodeServer"); err != nil {
+		return nil, err
+	}
+	defer nsMtx.UnlockKey(volId)
+
+	if err := mount.New("").Unmount(targetPath); err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
+	return &csi.NodeUnpublishVolumeResponse{}, nil
+}

pkg/cephfs/util.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	// "fmt"
+	"os/exec"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"k8s.io/kubernetes/pkg/util/keymutex"
+	"k8s.io/kubernetes/pkg/util/mount"
+)
+
+func execCommand(command string, args ...string) ([]byte, error) {
+	cmd := exec.Command(command, args...)
+	return cmd.CombinedOutput()
+}
+
+func isMountPoint(p string) (bool, error) {
+	notMnt, err := mount.New("").IsLikelyNotMountPoint(p)
+	if err != nil {
+		return false, status.Error(codes.Internal, err.Error())
+	}
+
+	return !notMnt, nil
+}
+
+func tryLock(id string, mtx keymutex.KeyMutex, name string) error {
+	// TODO uncomment this once TryLockKey gets into Kubernetes
+	/*
+		if !mtx.TryLockKey(id) {
+			msg := fmt.Sprintf("%s has a pending operation on %s", name, req.GetVolumeId())
+			glog.Infoln(msg)
+
+			return status.Error(codes.Aborted, msg)
+		}
+	*/
+
+	return nil
+}

pkg/cephfs/volume.go

@@ -0,0 +1,58 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"fmt"
+	"os"
+)
+
+func createMountPoint(root string) error {
+	return os.MkdirAll(root, 0750)
+}
+
+func deleteVolumePath(volPath string) error {
+	return os.RemoveAll(volPath)
+}
+
+func mountFuse(root string) error {
+	out, err := execCommand("ceph-fuse", root)
+	if err != nil {
+		return fmt.Errorf("cephfs: ceph-fuse failed with following error: %v\ncephfs: ceph-fuse output: %s", err, out)
+	}
+
+	return nil
+}
+
+func unmountFuse(root string) error {
+	out, err := execCommand("fusermount", "-u", root)
+	if err != nil {
+		return fmt.Errorf("cephfs: fusermount failed with following error: %v\ncephfs: fusermount output: %s", err, out)
+	}
+
+	return nil
+}
+
+func setVolAttributes(volPath string /*opts *fsVolumeOptions*/, maxBytes int64) error {
+	out, err := execCommand("setfattr", "-n", "ceph.quota.max_bytes",
+		"-v", fmt.Sprintf("%d", maxBytes), volPath)
+	if err != nil {
+		return fmt.Errorf("cephfs: setfattr failed with following error: %v\ncephfs: setfattr output: %s", err, out)
+	}
+
+	return nil
+}

pkg/cephfs/volumeidentifier.go

@@ -0,0 +1,41 @@
+/*
+Copyright 2018 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 cephfs
+
+import (
+	"github.com/container-storage-interface/spec/lib/go/csi"
+	"github.com/pborman/uuid"
+)
+
+type volumeIdentifier struct {
+	name, uuid, id string
+}
+
+func newVolumeIdentifier(volOptions *volumeOptions, req *csi.CreateVolumeRequest) *volumeIdentifier {
+	volId := volumeIdentifier{
+		name: req.GetName(),
+		uuid: uuid.NewUUID().String(),
+	}
+
+	volId.id = "csi-rbd-" + volId.uuid
+
+	if volId.name == "" {
+		volId.name = volOptions.Pool + "-dynamic-pvc-" + volId.uuid
+	}
+
+	return &volId
+}

pkg/cephfs/volumeoptions.go

@@ -0,0 +1,63 @@
+/*
+Copyright 2018 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 cephfs
+
+import "errors"
+
+type volumeOptions struct {
+	VolName     string `json:"volName"`
+	Monitor     string `json:"monitor"`
+	Pool        string `json:"pool"`
+	AdminId     string `json:"adminID"`
+	AdminSecret string `json:"adminSecret"`
+}
+
+func extractOption(dest *string, optionLabel string, options map[string]string) error {
+	if opt, ok := options[optionLabel]; !ok {
+		return errors.New("Missing required parameter " + optionLabel)
+	} else {
+		*dest = opt
+		return nil
+	}
+}
+
+func newVolumeOptions(volOptions map[string]string) (*volumeOptions, error) {
+	var opts volumeOptions
+	// XXX early return - we're not reading credentials from volOptions for now...
+	// i'll finish this once ceph-fuse accepts passing credentials through cmd args
+	return &opts, nil
+
+	/*
+		if err := extractOption(&opts.AdminId, "adminID", volOptions); err != nil {
+			return nil, err
+		}
+
+		if err := extractOption(&opts.AdminSecret, "adminSecret", volOptions); err != nil {
+			return nil, err
+		}
+
+		if err := extractOption(&opts.Monitors, "monitors", volOptions); err != nil {
+			return nil, err
+		}
+
+		if err := extractOption(&opts.Pool, "pool", volOptions); err != nil {
+			return nil, err
+		}
+
+		return &opts, nil
+	*/
+}