Merge pull request #39 from gman0/fix-cephfs-provisioner

Fix broken cephfs provisioner

deploy/cephfs/kubernetes/csi-provisioner.yaml

@@ -78,7 +78,7 @@ spec:
       serviceAccount: csi-provisioner
       containers:
         - name: csi-provisioner
-          image: quay.io/k8scsi/csi-provisioner:v0.2.0
+          image: quay.io/k8scsi/csi-provisioner:v0.2.1
           args:
             - "--provisioner=csi-cephfsplugin"
             - "--csi-address=$(ADDRESS)"

pkg/cephfs/cephconf.go

@@ -52,8 +52,8 @@ const cephSecret = `{{.Key}}`
 const (
 	cephConfigRoot         = "/etc/ceph"
 	cephConfigFileNameFmt  = "ceph.share.%s.conf"
-	cephKeyringFileNameFmt = "ceph.client.%s.keyring"
+	cephKeyringFileNameFmt = "ceph.share.%s.client.%s.keyring"
-	cephSecretFileNameFmt  = "ceph.client.%s.secret"
+	cephSecretFileNameFmt  = "ceph.share.%s.client.%s.secret"
 )
 
 var (
@@ -115,34 +115,37 @@ type cephKeyringData struct {
 	UserId, Key     string
 	RootPath        string
 	Pool, Namespace string
+	VolumeUuid      string
 }
 
 func (d *cephKeyringData) writeToFile() error {
-	return writeCephTemplate(fmt.Sprintf(cephKeyringFileNameFmt, d.UserId), 0600, cephKeyringTempl, d)
+	return writeCephTemplate(fmt.Sprintf(cephKeyringFileNameFmt, d.VolumeUuid, d.UserId), 0600, cephKeyringTempl, d)
 }
 
 type cephFullCapsKeyringData struct {
 	UserId, Key string
+	VolumeUuid  string
 }
 
 func (d *cephFullCapsKeyringData) writeToFile() error {
-	return writeCephTemplate(fmt.Sprintf(cephKeyringFileNameFmt, d.UserId), 0600, cephFullCapsKeyringTempl, d)
+	return writeCephTemplate(fmt.Sprintf(cephKeyringFileNameFmt, d.VolumeUuid, d.UserId), 0600, cephFullCapsKeyringTempl, d)
 }
 
 type cephSecretData struct {
 	UserId, Key string
+	VolumeUuid  string
 }
 
 func (d *cephSecretData) writeToFile() error {
-	return writeCephTemplate(fmt.Sprintf(cephSecretFileNameFmt, d.UserId), 0600, cephSecretTempl, d)
+	return writeCephTemplate(fmt.Sprintf(cephSecretFileNameFmt, d.VolumeUuid, d.UserId), 0600, cephSecretTempl, d)
 }
 
-func getCephSecretPath(userId string) string {
+func getCephSecretPath(volUuid, userId string) string {
-	return path.Join(cephConfigRoot, fmt.Sprintf(cephSecretFileNameFmt, userId))
+	return path.Join(cephConfigRoot, fmt.Sprintf(cephSecretFileNameFmt, volUuid, userId))
 }
 
-func getCephKeyringPath(userId string) string {
+func getCephKeyringPath(volUuid, userId string) string {
-	return path.Join(cephConfigRoot, fmt.Sprintf(cephKeyringFileNameFmt, userId))
+	return path.Join(cephConfigRoot, fmt.Sprintf(cephKeyringFileNameFmt, volUuid, userId))
 }
 
 func getCephConfPath(volUuid string) string {

pkg/cephfs/cephuser.go

@@ -57,26 +57,16 @@ func getCephUser(userId string) (*cephEntity, error) {
 	return &ents[0], nil
 }
 
-func (e *cephEntity) create() error {
+func createCephUser(volOptions *volumeOptions, cr *credentials, volUuid string) (*cephEntity, error) {
-	return execCommandJson(e, "ceph", "auth", "get-or-create", e.Entity, "mds", e.Caps.Mds, "osd", e.Caps.Osd, "mon", e.Caps.Mon)
-
-}
-
-func createCephUser(volOptions *volumeOptions, volUuid string, readOnly bool) (*cephEntity, error) {
-	access := "rw"
-	if readOnly {
-		access = "r"
-	}
-
 	caps := cephEntityCaps{
-		Mds: fmt.Sprintf("allow %s path=%s", access, getVolumeRootPath_ceph(volUuid)),
+		Mds: fmt.Sprintf("allow rw path=%s", getVolumeRootPath_ceph(volUuid)),
 		Mon: "allow r",
-		Osd: fmt.Sprintf("allow %s pool=%s namespace=%s", access, volOptions.Pool, getVolumeNamespace(volUuid)),
+		Osd: fmt.Sprintf("allow rw pool=%s namespace=%s", volOptions.Pool, getVolumeNamespace(volUuid)),
 	}
 
 	var ents []cephEntity
 	args := [...]string{
-		"auth", "-f", "json",
+		"auth", "-f", "json", "-c", getCephConfPath(volUuid), "-n", cephEntityClientPrefix + cr.id,
 		"get-or-create", cephEntityClientPrefix + getCephUserName(volUuid),
 		"mds", caps.Mds,
 		"mon", caps.Mon,
@@ -90,15 +80,20 @@ func createCephUser(volOptions *volumeOptions, volUuid string, readOnly bool) (*
 	return &ents[0], nil
 }
 
-func deleteCephUser(volUuid string) error {
+func deleteCephUser(cr *credentials, volUuid string) error {
 	userId := getCephUserName(volUuid)
 
-	if err := execCommandAndValidate("ceph", "auth", "rm", cephEntityClientPrefix+userId); err != nil {
+	args := [...]string{
+		"-c", getCephConfPath(volUuid), "-n", cephEntityClientPrefix + cr.id,
+		"auth", "rm", cephEntityClientPrefix + userId,
+	}
+
+	if err := execCommandAndValidate("ceph", args[:]...); err != nil {
 		return err
 	}
 
-	os.Remove(getCephKeyringPath(userId))
+	os.Remove(getCephKeyringPath(volUuid, userId))
-	os.Remove(getCephSecretPath(userId))
+	os.Remove(getCephSecretPath(volUuid, userId))
 
 	return nil
 }

pkg/cephfs/controllerserver.go

@@ -77,6 +77,12 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 
 	volId := newVolumeIdentifier(volOptions, req)
 
+	conf := cephConfigData{Monitors: volOptions.Monitors, VolumeUuid: volId.uuid}
+	if err = conf.writeToFile(); err != nil {
+		glog.Errorf("failed to write ceph config file to %s: %v", getCephConfPath(volId.uuid), err)
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	// Create a volume in case the user didn't provide one
 
 	if volOptions.ProvisionVolume {
@@ -87,7 +93,7 @@ func (cs *controllerServer) CreateVolume(ctx context.Context, req *csi.CreateVol
 			return nil, status.Error(codes.InvalidArgument, err.Error())
 		}
 
-		if err = storeCephAdminCredentials(cr); err != nil {
+		if err = storeCephAdminCredentials(volId.uuid, cr); err != nil {
 			glog.Errorf("failed to store admin credentials for '%s': %v", cr.id, err)
 			return nil, status.Error(codes.Internal, err.Error())
 		}
@@ -164,16 +170,16 @@ func (cs *controllerServer) DeleteVolume(ctx context.Context, req *csi.DeleteVol
 
 	if ent.VolOptions.ProvisionVolume {
 		// The user is no longer needed
-		if err := deleteCephUser(volUuid); err != nil {
+		if err := deleteCephUser(cr, volUuid); err != nil {
 			glog.Warningf("failed to delete ceph user '%s': %v", cr.id, err)
 		}
 
 		userId := getCephUserName(volUuid)
-		os.Remove(getCephKeyringPath(userId))
+		os.Remove(getCephKeyringPath(volUuid, userId))
-		os.Remove(getCephSecretPath(userId))
+		os.Remove(getCephSecretPath(volUuid, userId))
 	} else {
-		os.Remove(getCephKeyringPath(cr.id))
+		os.Remove(getCephKeyringPath(volUuid, cr.id))
-		os.Remove(getCephSecretPath(cr.id))
+		os.Remove(getCephSecretPath(volUuid, cr.id))
 	}
 
 	if err := volCache.erase(volUuid); err != nil {

pkg/cephfs/identityserver.go

@@ -17,9 +17,26 @@ limitations under the License.
 package cephfs
 
 import (
+	"context"
+
+	"github.com/container-storage-interface/spec/lib/go/csi/v0"
 	"github.com/kubernetes-csi/drivers/pkg/csi-common"
 )
 
 type identityServer struct {
 	*csicommon.DefaultIdentityServer
 }
+
+func (is *identityServer) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) {
+	return &csi.GetPluginCapabilitiesResponse{
+		Capabilities: []*csi.PluginCapability{
+			{
+				Type: &csi.PluginCapability_Service_{
+					Service: &csi.PluginCapability_Service{
+						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
+					},
+				},
+			},
+		},
+	}, nil
+}

pkg/cephfs/nodeserver.go

@@ -71,7 +71,20 @@ func handleUser(volOptions *volumeOptions, volUuid string, req *csi.NodePublishV
 	if volOptions.ProvisionVolume {
 		// The volume is provisioned dynamically, create a dedicated user
 
-		if ent, err := createCephUser(volOptions, volUuid, req.GetReadonly()); err != nil {
+		// First, store admin credentials - those are needed for creating a user
+
+		adminCr, err := getAdminCredentials(req.GetNodePublishSecrets())
+		if err != nil {
+			return nil, err
+		}
+
+		if err = storeCephAdminCredentials(volUuid, adminCr); err != nil {
+			return nil, err
+		}
+
+		// Then create the user
+
+		if ent, err := createCephUser(volOptions, adminCr, volUuid); err != nil {
 			return nil, err
 		} else {
 			cr.id = ent.Entity[len(cephEntityClientPrefix):]

pkg/cephfs/util.go

@@ -82,6 +82,7 @@ func storeCephUserCredentials(volUuid string, cr *credentials, volOptions *volum
 		UserId:     cr.id,
 		Key:        cr.key,
 		RootPath:   volOptions.RootPath,
+		VolumeUuid: volUuid,
 	}
 
 	if volOptions.ProvisionVolume {
@@ -89,14 +90,14 @@ func storeCephUserCredentials(volUuid string, cr *credentials, volOptions *volum
 		keyringData.Namespace = getVolumeNamespace(volUuid)
 	}
 
-	return storeCephCredentials(cr, &keyringData)
+	return storeCephCredentials(volUuid, cr, &keyringData)
 }
 
-func storeCephAdminCredentials(cr *credentials) error {
+func storeCephAdminCredentials(volUuid string, cr *credentials) error {
-	return storeCephCredentials(cr, &cephFullCapsKeyringData{UserId: cr.id, Key: cr.key})
+	return storeCephCredentials(volUuid, cr, &cephFullCapsKeyringData{UserId: cr.id, Key: cr.key, VolumeUuid: volUuid})
 }
 
-func storeCephCredentials(cr *credentials, keyringData cephConfigWriter) error {
+func storeCephCredentials(volUuid string, cr *credentials, keyringData cephConfigWriter) error {
 	if err := keyringData.writeToFile(); err != nil {
 		return err
 	}
@@ -104,6 +105,7 @@ func storeCephCredentials(cr *credentials, keyringData cephConfigWriter) error {
 	secret := cephSecretData{
 		UserId:     cr.id,
 		Key:        cr.key,
+		VolumeUuid: volUuid,
 	}
 
 	if err := secret.writeToFile(); err != nil {

pkg/cephfs/volume.go

@@ -88,7 +88,7 @@ func createVolume(volOptions *volumeOptions, adminCr *credentials, volUuid strin
 	// Access to cephfs's / is required
 	volOptions.RootPath = "/"
 
-	if err := mountKernel(cephRoot, adminCr, volOptions); err != nil {
+	if err := mountKernel(cephRoot, adminCr, volOptions, volUuid); err != nil {
 		return fmt.Errorf("error mounting ceph root: %v", err)
 	}
 
@@ -144,7 +144,7 @@ func purgeVolume(volId string, cr *credentials, volOptions *volumeOptions) error
 		return err
 	}
 
-	if err := mountKernel(volRoot, cr, volOptions); err != nil {
+	if err := mountKernel(volRoot, cr, volOptions, volUuid); err != nil {
 		return err
 	}
 

pkg/cephfs/volumemounter.go

@@ -38,7 +38,7 @@ func mountFuse(mountPoint string, cr *credentials, volOptions *volumeOptions, vo
 		mountPoint,
 		"-c", getCephConfPath(volUuid),
 		"-n", cephEntityClientPrefix + cr.id,
-		"--keyring", getCephKeyringPath(cr.id),
+		"--keyring", getCephKeyringPath(volUuid, cr.id),
 		"-r", volOptions.RootPath,
 	}
 
@@ -74,7 +74,7 @@ func (m *fuseMounter) mount(mountPoint string, cr *credentials, volOptions *volu
 
 type kernelMounter struct{}
 
-func mountKernel(mountPoint string, cr *credentials, volOptions *volumeOptions) error {
+func mountKernel(mountPoint string, cr *credentials, volOptions *volumeOptions, volUuid string) error {
 	if err := execCommandAndValidate("modprobe", "ceph"); err != nil {
 		return err
 	}
@@ -84,7 +84,7 @@ func mountKernel(mountPoint string, cr *credentials, volOptions *volumeOptions)
 		fmt.Sprintf("%s:%s", volOptions.Monitors, volOptions.RootPath),
 		mountPoint,
 		"-o",
-		fmt.Sprintf("name=%s,secretfile=%s", cr.id, getCephSecretPath(cr.id)),
+		fmt.Sprintf("name=%s,secretfile=%s", cr.id, getCephSecretPath(volUuid, cr.id)),
 	)
 }
 
@@ -99,7 +99,7 @@ func (m *kernelMounter) mount(mountPoint string, cr *credentials, volOptions *vo
 		return err
 	}
 
-	if err := mountKernel(localVolRoot, cr, volOptions); err != nil {
+	if err := mountKernel(localVolRoot, cr, volOptions, volUuid); err != nil {
 		return err
 	}