operator: add ceph image version label to PVC

This PR makes Rook operator to add a label on PVC that contains an image version of Ceph when creating an OSD.

Signed-off-by: YZ775 <yuzuki-mimura@cybozu.co.jp>

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

@@ -19,6 +19,7 @@ package osd
 import (
 	"context"
 	"fmt"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -30,6 +31,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 // deviceSet is the processed version of the StorageClassDeviceSet
@@ -210,7 +212,8 @@ func (c *Cluster) createDeviceSetPVC(existingPVCs map[string]*v1.PersistentVolum
 		pvcID = deviceSetPVCID(deviceSetName, pvcTemplate.GetName(), setIndex)
 		existingPVC = existingPVCs[pvcID]
 	}
-	pvc := makeDeviceSetPVC(deviceSetName, pvcID, setIndex, pvcTemplate, c.clusterInfo.Namespace)
+
+	pvc := makeDeviceSetPVC(deviceSetName, pvcID, setIndex, pvcTemplate, c.clusterInfo.Namespace, createValidImageVersionLabel(c.spec.CephVersion.Image), createValidImageVersionLabel(c.rookVersion))
 	err := c.clusterInfo.OwnerInfo.SetControllerReference(pvc)
 	if err != nil {
 		return nil, errors.Wrapf(err, "failed to set owner reference to osd pvc %q", pvc.Name)
@@ -234,8 +237,8 @@ func (c *Cluster) createDeviceSetPVC(existingPVCs map[string]*v1.PersistentVolum
 	return deployedPVC, nil
 }
 
-func makeDeviceSetPVC(deviceSetName, pvcID string, setIndex int, pvcTemplate v1.PersistentVolumeClaim, namespace string) *v1.PersistentVolumeClaim {
-	pvcLabels := makeStorageClassDeviceSetPVCLabel(deviceSetName, pvcID, setIndex)
+func makeDeviceSetPVC(deviceSetName, pvcID string, setIndex int, pvcTemplate v1.PersistentVolumeClaim, namespace string, cephImage string, rookImage string) *v1.PersistentVolumeClaim {
+	pvcLabels := makeStorageClassDeviceSetPVCLabel(deviceSetName, pvcID, setIndex, cephImage, rookImage)
 
 	// Add user provided labels to pvcTemplates
 	for k, v := range pvcTemplate.GetLabels() {
@@ -293,3 +296,15 @@ func deviceSetPVCID(deviceSetName, pvcTemplateName string, setIndex int) string
 	deviceSetName = strings.Replace(deviceSetName, ".", "-", -1)
 	return fmt.Sprintf("%s-%s-%d", deviceSetName, cleanName, setIndex)
 }
+
+func createValidImageVersionLabel(image string) string {
+	// regex to replace characters used by image name format that are not allowed in label values
+	re := regexp.MustCompile("[/:]")
+	cephImageVersion := re.ReplaceAllString(image, "_")
+
+	if validation.IsValidLabelValue(cephImageVersion) != nil {
+		logger.Infof("image %q contains invalid character, skipping adding label", image)
+		cephImageVersion = ""
+	}
+	return cephImageVersion
+}

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

@@ -83,6 +83,14 @@ func testPrepareDeviceSets(t *testing.T, setTemplateName bool) {
 	}
 	assert.Equal(t, fmt.Sprintf("mydata-%s-0", expectedName), pvcs.Items[0].GenerateName)
 	assert.Equal(t, cluster.clusterInfo.Namespace, pvcs.Items[0].Namespace)
+
+	//Verify that the PVC has correct Image Version Label
+	cephImageVersion := createValidImageVersionLabel(cluster.spec.CephVersion.Image)
+	for _, item := range pvcs.Items {
+		val, exist := item.Labels[CephImageLabelKey]
+		assert.Equal(t, true, exist)
+		assert.Equal(t, cephImageVersion, val)
+	}
 }
 
 func TestPrepareDeviceSetWithHolesInPVCs(t *testing.T) {
@@ -284,3 +292,12 @@ func TestPVCName(t *testing.T) {
 	id = deviceSetPVCID("device.set.with.dots", "b", 10)
 	assert.Equal(t, "device-set-with-dots-b-10", id)
 }
+
+func TestCreateValidImageVersionLabel(t *testing.T) {
+	image := "ceph/ceph:v17.2.6"
+	assert.Equal(t, "ceph_ceph_v17.2.6", createValidImageVersionLabel(image))
+	image = "rook/ceph:master"
+	assert.Equal(t, "rook_ceph_master", createValidImageVersionLabel(image))
+	image = ".invalid_label"
+	assert.Equal(t, "", createValidImageVersionLabel(image))
+}

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

@@ -36,13 +36,19 @@ const (
 	OSDOverPVCLabelKey = "ceph.rook.io/pvc"
 	// TopologyLocationLabel is the crush location label added to OSD deployments
 	TopologyLocationLabel = "topology-location-%s"
+	// CephImageLabelKey is the ceph image version label added to PVC
+	CephImageLabelKey = "ceph.rook.io/cephImageAtCreation"
+	// RookImageLabelKey is the rook image version label added to PVC
+	RookImageLabelKey = "ceph.rook.io/rookImageAtCreation"
 )
 
-func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int) map[string]string {
+func makeStorageClassDeviceSetPVCLabel(storageClassDeviceSetName, pvcStorageClassDeviceSetPVCId string, setIndex int, cephImage string, rookImage string) map[string]string {
 	return map[string]string{
 		CephDeviceSetLabelKey:      storageClassDeviceSetName,
 		CephSetIndexLabelKey:       fmt.Sprintf("%d", setIndex),
 		CephDeviceSetPVCIDLabelKey: pvcStorageClassDeviceSetPVCId,
+		CephImageLabelKey:          cephImage,
+		RookImageLabelKey:          rookImage,
 	}
 }