ceph: add support for encrypted metadata pvc

Now the metadata pvc will also be encrypted if the storageClassDeviceSet
has the "encrypted" flag turned on.

Signed-off-by: Sébastien Han <seb@redhat.com>

pkg/daemon/ceph/osd/volume.go

@@ -24,6 +24,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -337,10 +338,16 @@ func (a *OsdAgent) initializeBlockPVC(context *clusterd.Context, devices *Device
 			if lvBackedPV || cephVolumeMode == "raw" && !isEncrypted {
 				blockPath = deviceArg
 			} else if cephVolumeMode == "raw" && isEncrypted {
-				blockPath = getEncryptedBlockPath(op)
+				blockPath = getEncryptedBlockPath(op, oposd.DmcryptBlockType)
 				if blockPath == "" {
 					return "", "", "", errors.New("failed to get encrypted block path from ceph-volume lvm prepare output")
 				}
+				if metadataBlockPath != "" {
+					metadataBlockPath = getEncryptedBlockPath(op, oposd.DmcryptMetadataType)
+					if metadataBlockPath == "" {
+						return "", "", "", errors.New("failed to get encrypted block.db path from ceph-volume lvm prepare output")
+					}
+				}
 			} else {
 				blockPath = getLVPath(op)
 				if blockPath == "" {
@@ -370,14 +377,16 @@ func getLVPath(op string) string {
 	return ""
 }
 
-func getEncryptedBlockPath(op string) string {
-	tmp := sys.Grep(op, "luksOpen")
-	tmpList := strings.Split(tmp, " ")
+func getEncryptedBlockPath(op, blockType string) string {
+	re := regexp.MustCompile("(?m)^.*luksOpen.*$")
+	matches := re.FindAllString(op, -1)
 
-	// The command looks like: "Running command: /usr/sbin/cryptsetup --key-file - --allow-discards luksOpen /dev/xvdbr ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt"
-	for _, item := range tmpList {
-		if strings.Contains(item, "block-dmcrypt") {
-			return fmt.Sprintf("/dev/mapper/%s", item)
+	for _, line := range matches {
+		lineSlice := strings.Fields(line)
+		for _, word := range lineSlice {
+			if strings.Contains(word, blockType) {
+				return fmt.Sprintf("/dev/mapper/%s", word)
+			}
 		}
 	}
 
@@ -770,6 +779,15 @@ func GetCephVolumeRawOSDs(context *clusterd.Context, clusterInfo *client.Cluster
 			if err != nil {
 				return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID)
 			}
+
+			if metadataBlock != "" {
+				// Close encrypted device
+				err = closeEncryptedDevice(context, metadataBlock)
+				if err != nil {
+					return nil, errors.Wrapf(err, "failed to close encrypted device %q for osd %d", block, osdID)
+				}
+
+			}
 		}
 
 		osds = append(osds, osd)

pkg/daemon/ceph/osd/volume_test.go

@@ -607,20 +607,30 @@ func TestPrintCVLogContent(t *testing.T) {
 }
 
 func TestGetEncryptedBlockPath(t *testing.T) {
+	cvOp := `
+Running command: /usr/bin/ceph-authtool --gen-print-key
+Running command: /usr/sbin/cryptsetup --batch-mode --key-file - luksFormat /dev/xvdbq
+Running command: /usr/sbin/cryptsetup --key-file - --allow-discards luksOpen /dev/xvdbq ceph-22971ee4-017c-48ec-b52b-68cdd2cc0ed2-xvdbq-block-dmcrypt
+Running command: /usr/sbin/cryptsetup --batch-mode --key-file - luksFormat /dev/xvdbm
+Running command: /usr/sbin/cryptsetup --key-file - --allow-discards luksOpen /dev/xvdbm ceph-22971ee4-017c-48ec-b52b-68cdd2cc0ed2-xvdbm-db-dmcrypt
+Running command: /usr/bin/mount -t tmpfs tmpfs /var/lib/ceph/osd/ceph-1`
+
 	type args struct {
-		op string
+		op        string
+		blockType string
 	}
 	tests := []struct {
 		name string
 		args args
 		want string
 	}{
-		{"not-found", args{"Running command: /usr/bin/mount -t tmpfs tmpfs /var/lib/ceph/osd/ceph-1"}, ""},
-		{"found", args{"Running command: /usr/sbin/cryptsetup --key-file - --allow-discards luksOpen /dev/xvdbr ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt"}, "/dev/mapper/ceph-43e9efed-0676-4731-b75a-a4c42ece1bb1-xvdbr-block-dmcrypt"},
+		{"not-found", args{"Running command: /usr/bin/mount -t tmpfs tmpfs /var/lib/ceph/osd/ceph-1", "block-dmcrypt"}, ""},
+		{"found-block", args{cvOp, "block-dmcrypt"}, "/dev/mapper/ceph-22971ee4-017c-48ec-b52b-68cdd2cc0ed2-xvdbq-block-dmcrypt"},
+		{"found-db", args{cvOp, "db-dmcrypt"}, "/dev/mapper/ceph-22971ee4-017c-48ec-b52b-68cdd2cc0ed2-xvdbm-db-dmcrypt"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			if got := getEncryptedBlockPath(tt.args.op); got != tt.want {
+			if got := getEncryptedBlockPath(tt.args.op, tt.args.blockType); got != tt.want {
 				t.Errorf("getEncryptedBlockPath() = %v, want %v", got, tt.want)
 			}
 		})

pkg/operator/ceph/cluster/osd/config.go

@@ -121,12 +121,12 @@ func encryptionKeyPath() string {
 	return fmt.Sprintf("%s/%s", opconfig.EtcCephDir, encryptionKeyFileName)
 }
 
-func encryptionDMName(pvcName string) string {
-	return fmt.Sprintf("%s-block-dmcrypt", pvcName)
+func encryptionDMName(pvcName, blockType string) string {
+	return fmt.Sprintf("%s-%s", pvcName, blockType)
 }
 
-func encryptionDMPath(pvcName string) string {
-	return fmt.Sprintf("/dev/mapper/%s", encryptionDMName(pvcName))
+func encryptionDMPath(pvcName, blockType string) string {
+	return fmt.Sprintf("/dev/mapper/%s", encryptionDMName(pvcName, blockType))
 }
 
 func generateDmCryptKey() (string, error) {
@@ -158,7 +158,7 @@ func generateOSDEncryptionSecretName(pvcName string) string {
 	return fmt.Sprintf("%s-%s", osdEncryptionSecretNamePrefix, pvcName)
 }
 
-func (c *Cluster) isCephVolumeRAwModeSupported() bool {
+func (c *Cluster) isCephVolumeRawModeSupported() bool {
 	if c.clusterInfo.CephVersion.IsAtLeast(cephVolumeRawEncryptionModeMinNautilusCephVersion) && !c.clusterInfo.CephVersion.IsOctopus() {
 		return true
 	}

pkg/operator/ceph/cluster/osd/config_test.go

@@ -76,7 +76,7 @@ func TestClusterIsCephVolumeRAwModeSupported(t *testing.T) {
 				ValidStorage: tt.fields.ValidStorage,
 				kv:           tt.fields.kv,
 			}
-			if got := c.isCephVolumeRAwModeSupported(); got != tt.want {
+			if got := c.isCephVolumeRawModeSupported(); got != tt.want {
 				t.Errorf("Cluster.isCephVolumeRAwModeSupported() = %v, want %v", got, tt.want)
 			}
 		})

pkg/operator/ceph/cluster/osd/osd.go

@@ -254,7 +254,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
 
 		// If the deviceSet template has "encrypted" but the Ceph version is not compatible
 		if osdProps.encrypted {
-			if !c.isCephVolumeRAwModeSupported() {
+			if !c.isCephVolumeRawModeSupported() {
 				errMsg := fmt.Sprintf("failed to validate storageClassDeviceSet %q. min required ceph version to support encryption is %q or %q", volume.Name, cephVolumeRawEncryptionModeMinNautilusCephVersion.String(), cephVolumeRawEncryptionModeMinOctopusCephVersion.String())
 				config.addError(errMsg)
 				continue

pkg/operator/ceph/cluster/osd/spec.go

@@ -35,18 +35,26 @@ import (
 )
 
 const (
-	rookBinariesMountPath                 = "/rook"
-	rookBinariesVolumeName                = "rook-binaries"
-	activateOSDVolumeName                 = "activate-osd"
-	activateOSDMountPath                  = "/var/lib/ceph/osd/ceph-"
-	blockPVCMapperInitContainer           = "blkdevmapper"
-	blockEncryptionOpenInitContainer      = "encryption-open"
-	blockPVCMapperEncryptionInitContainer = "blkdevmapper-encryption"
-	blockPVCMetadataMapperInitContainer   = "blkdevmapper-metadata"
-	blockPVCWalMapperInitContainer        = "blkdevmapper-wal"
-	activatePVCOSDInitContainer           = "activate"
-	expandPVCOSDInitContainer             = "expand-bluefs"
-	encryptionKeyFileName                 = "luks_key"
+	rookBinariesMountPath                         = "/rook"
+	rookBinariesVolumeName                        = "rook-binaries"
+	activateOSDVolumeName                         = "activate-osd"
+	activateOSDMountPath                          = "/var/lib/ceph/osd/ceph-"
+	blockPVCMapperInitContainer                   = "blkdevmapper"
+	blockEncryptionOpenInitContainer              = "encryption-open"
+	blockEncryptionOpenMetadataInitContainer      = "encryption-open-metadata"
+	blockPVCMapperEncryptionInitContainer         = "blkdevmapper-encryption"
+	blockPVCMapperEncryptionMetadataInitContainer = "blkdevmapper-metadata-encryption"
+	blockPVCMetadataMapperInitContainer           = "blkdevmapper-metadata"
+	blockPVCWalMapperInitContainer                = "blkdevmapper-wal"
+	activatePVCOSDInitContainer                   = "activate"
+	expandPVCOSDInitContainer                     = "expand-bluefs"
+	encryptionKeyFileName                         = "luks_key"
+	// DmcryptBlockType is a portion of the device mapper name for the encrypted OSD on PVC block.db (rocksdb db)
+	DmcryptBlockType = "block-dmcrypt"
+	// DmcryptMetadataType is a portion of the device mapper name for the encrypted OSD on PVC block
+	DmcryptMetadataType = "db-dmcrypt"
+	dmcryptBlockName    = "block"
+	dmcryptMetadataName = "block.db"
 )
 
 const (
@@ -320,8 +328,8 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd OSDInfo, provisionC
 	if osdProps.onPVC() && osd.CVMode == "raw" {
 		if osdProps.encrypted {
 			// Add a new init container to open the encrypted disk!
-			initContainers = append(initContainers, c.getPVCEncryptionOpenInitContainerActivate(osdProps))
-			initContainers = append(initContainers, c.getPVCEncryptionInitContainerActivate(osdDataDirPath, osdProps))
+			initContainers = append(initContainers, c.getPVCEncryptionOpenInitContainerActivate(osdProps)...)
+			initContainers = append(initContainers, c.getPVCEncryptionInitContainerActivate(osdDataDirPath, osdProps)...)
 			// TODO: ADD METADATA DEV
 		} else {
 			initContainers = append(initContainers, c.getPVCInitContainerActivate(osdDataDirPath, osdProps))
@@ -547,7 +555,7 @@ func (c *Cluster) getPVCInitContainerActivate(mountPath string, osdProps osdProp
 		Command: []string{
 			"cp",
 		},
-		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.pvc.ClaimName), path.Join(mountPath, "block")},
+		Args: []string{"-a", fmt.Sprintf("/%s", osdProps.pvc.ClaimName), path.Join(mountPath, dmcryptBlockName)},
 		VolumeDevices: []v1.VolumeDevice{
 			{
 				Name:       osdProps.pvc.ClaimName,
@@ -560,45 +568,84 @@ func (c *Cluster) getPVCInitContainerActivate(mountPath string, osdProps osdProp
 	}
 }
 
-func (c *Cluster) getPVCEncryptionOpenInitContainerActivate(osdProps osdProperties) v1.Container {
-	container := v1.Container{
-		Name:  blockEncryptionOpenInitContainer,
+func (c *Cluster) generateEncryptionOpenBlockContainer(resources v1.ResourceRequirements, containerName, pvcName, blockType string) v1.Container {
+	return v1.Container{
+		Name:  containerName,
 		Image: c.spec.CephVersion.Image,
 		Command: []string{
 			"cryptsetup",
 		},
 		// Runs "cryptsetup --verbose --key-file /etc/ceph/luks_key --allow-discards luksOpen /set1-data-0-7dwll set1-data-0-7dwll-block-dmcrypt"
-		Args: []string{"--verbose", "--key-file", encryptionKeyPath(), "--allow-discards", "luksOpen", fmt.Sprintf("/%s", osdProps.pvc.ClaimName), encryptionDMName(osdProps.pvc.ClaimName)},
+		Args: []string{
+			"--verbose",
+			"--key-file",
+			encryptionKeyPath(),
+			"--allow-discards",
+			"luksOpen",
+			fmt.Sprintf("/%s", pvcName),
+			encryptionDMName(pvcName, blockType),
+		},
 		VolumeDevices: []v1.VolumeDevice{
 			{
-				Name:       osdProps.pvc.ClaimName,
-				DevicePath: fmt.Sprintf("/%s", osdProps.pvc.ClaimName),
+				Name:       pvcName,
+				DevicePath: fmt.Sprintf("/%s", pvcName),
 			},
 		},
 		VolumeMounts:    []v1.VolumeMount{getDeviceMapperMount()},
 		SecurityContext: opmon.PodSecurityContext(),
-		Resources:       osdProps.resources,
+		Resources:       resources,
 	}
-	_, volMount := getEncryptionVolume(osdProps.pvc.ClaimName)
-	container.VolumeMounts = append(container.VolumeMounts, volMount)
-
-	return container
 }
 
-func (c *Cluster) getPVCEncryptionInitContainerActivate(mountPath string, osdProps osdProperties) v1.Container {
+func (c *Cluster) getPVCEncryptionOpenInitContainerActivate(osdProps osdProperties) []v1.Container {
+	containers := []v1.Container{}
+
+	// Main block container
+	blockContainer := c.generateEncryptionOpenBlockContainer(osdProps.resources, blockEncryptionOpenInitContainer, osdProps.pvc.ClaimName, DmcryptBlockType)
+	_, volMount := getEncryptionVolume(osdProps.pvc.ClaimName)
+	blockContainer.VolumeMounts = append(blockContainer.VolumeMounts, volMount)
+	containers = append(containers, blockContainer)
+
+	// If there is a metadata PVC
+	if osdProps.metadataPVC.ClaimName != "" {
+		metadataContainer := c.generateEncryptionOpenBlockContainer(osdProps.resources, blockEncryptionOpenMetadataInitContainer, osdProps.metadataPVC.ClaimName, DmcryptMetadataType)
+		// We use the same key for both block and block.db so we must use osdProps.pvc.ClaimName for the getEncryptionVolume()
+		_, volMount := getEncryptionVolume(osdProps.pvc.ClaimName)
+		metadataContainer.VolumeMounts = append(metadataContainer.VolumeMounts, volMount)
+		containers = append(containers, metadataContainer)
+	}
+
+	return containers
+}
+
+func (c *Cluster) generateEncryptionCopyBlockContainer(resources v1.ResourceRequirements, containerName, pvcName, mountPath, volumeMountPVCName, blockName, blockType string) v1.Container {
 	return v1.Container{
-		Name:  blockPVCMapperEncryptionInitContainer,
+		Name:  containerName,
 		Image: c.spec.CephVersion.Image,
 		Command: []string{
 			"cp",
 		},
-		Args:            []string{"-a", encryptionDMPath(osdProps.pvc.ClaimName), path.Join(mountPath, "block")},
-		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, osdProps.pvc.ClaimName)},
+		Args: []string{"-a", encryptionDMPath(pvcName, blockType), path.Join(mountPath, blockName)},
+		// volumeMountPVCName is crucial, especially when the block we copy is the metadata block
+		// its value must be the name of the block PV so that all init containers use the same bridge (the emptyDir shared by all the init containers)
+		VolumeMounts:    []v1.VolumeMount{getPvcOSDBridgeMountActivate(mountPath, volumeMountPVCName)},
 		SecurityContext: opmon.PodSecurityContext(),
-		Resources:       osdProps.resources,
+		Resources:       resources,
 	}
 }
 
+func (c *Cluster) getPVCEncryptionInitContainerActivate(mountPath string, osdProps osdProperties) []v1.Container {
+	containers := []v1.Container{}
+	containers = append(containers, c.generateEncryptionCopyBlockContainer(osdProps.resources, blockPVCMapperEncryptionInitContainer, osdProps.pvc.ClaimName, mountPath, osdProps.pvc.ClaimName, dmcryptBlockName, DmcryptBlockType))
+
+	// If there is a metadata PVC
+	if osdProps.metadataPVC.ClaimName != "" {
+		containers = append(containers, c.generateEncryptionCopyBlockContainer(osdProps.resources, blockPVCMapperEncryptionMetadataInitContainer, osdProps.metadataPVC.ClaimName, mountPath, osdProps.pvc.ClaimName, dmcryptMetadataName, DmcryptMetadataType))
+	}
+
+	return containers
+}
+
 // The reason why this is not part of getPVCInitContainer is that this will change the deployment spec object
 // and thus restart the osd deployment, so it is better to have it separated and only enable it
 // It will change the deployment spec because we must add a new argument to the method like 'mountPath' and use it in the container name

pkg/operator/ceph/cluster/osd/spec_test.go

@@ -26,14 +26,13 @@ import (
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
+	opconfig "github.com/rook/rook/pkg/operator/ceph/config"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
+	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/client-go/kubernetes/fake"
-
-	opconfig "github.com/rook/rook/pkg/operator/ceph/config"
-	"github.com/stretchr/testify/assert"
 )
 
 func TestPodContainer(t *testing.T) {
@@ -385,3 +384,41 @@ func TestOsdPrepareResources(t *testing.T) {
 	assert.Equal(t, "0", r.Limits.Memory().String())
 	assert.Equal(t, "250", r.Requests.Memory().String())
 }
+
+func TestClusterGetPVCEncryptionOpenInitContainerActivate(t *testing.T) {
+	c := New(&clusterd.Context{}, &cephclient.ClusterInfo{}, cephv1.ClusterSpec{}, "rook/rook:myversion")
+	osdProperties := osdProperties{
+		pvc: v1.PersistentVolumeClaimVolumeSource{
+			ClaimName: "pvc1",
+		},
+	}
+
+	// No metadata PVC
+	containers := c.getPVCEncryptionOpenInitContainerActivate(osdProperties)
+	assert.Equal(t, 1, len(containers))
+
+	// With metadata PVC
+	osdProperties.metadataPVC.ClaimName = "pvc2"
+	containers = c.getPVCEncryptionOpenInitContainerActivate(osdProperties)
+	assert.Equal(t, 2, len(containers))
+}
+
+func TestClusterGetPVCEncryptionInitContainerActivate(t *testing.T) {
+	c := New(&clusterd.Context{}, &cephclient.ClusterInfo{}, cephv1.ClusterSpec{}, "rook/rook:myversion")
+	osdProperties := osdProperties{
+		pvc: v1.PersistentVolumeClaimVolumeSource{
+			ClaimName: "pvc1",
+		},
+		resources: v1.ResourceRequirements{},
+	}
+	mountPath := "/var/lib/ceph/osd/ceph-0"
+
+	// No metadata PVC
+	containers := c.getPVCEncryptionInitContainerActivate(mountPath, osdProperties)
+	assert.Equal(t, 1, len(containers))
+
+	// With metadata PVC
+	osdProperties.metadataPVC.ClaimName = "pvc2"
+	containers = c.getPVCEncryptionInitContainerActivate(mountPath, osdProperties)
+	assert.Equal(t, 2, len(containers))
+}