ceph: add ceph version to cluster status

In order for deployments managed outside the immediate scope of
the ClusterController to accurately detect the Ceph version of
their images, the ClusterController must store the detected
version. This change records the Ceph version as part of the
cluster status and modifies the crashcollector deployment to
consume this field to record its own Ceph version. This replaces
an earlier solution to this problem in which the Ceph version
for a given image to version mapping was stored in a global
in-memory map by the operator.

Resolves: rook#4357
Signed-off-by: Elise Gafford <egafford@redhat.com>

pkg/apis/ceph.rook.io/v1/types.go

@@ -139,9 +139,10 @@ type MonitoringSpec struct {
 }
 
 type ClusterStatus struct {
-	State      ClusterState `json:"state,omitempty"`
-	Message    string       `json:"message,omitempty"`
-	CephStatus *CephStatus  `json:"ceph,omitempty"`
+	State       ClusterState    `json:"state,omitempty"`
+	Message     string          `json:"message,omitempty"`
+	CephStatus  *CephStatus     `json:"ceph,omitempty"`
+	CephVersion *ClusterVersion `json:"version,omitempty"`
 }
 
 type CephStatus struct {
@@ -152,6 +153,11 @@ type CephStatus struct {
 	PreviousHealth string                       `json:"previousHealth,omitempty"`
 }
 
+type ClusterVersion struct {
+	Image   string `json:"image,omitempty"`
+	Version string `json:"version,omitempty"`
+}
+
 type CephHealthMessage struct {
 	Severity string `json:"severity"`
 	Message  string `json:"message"`

pkg/operator/ceph/cluster/controller.go

@@ -44,6 +44,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/object/bucket"
 	objectuser "github.com/rook/rook/pkg/operator/ceph/object/user"
 	"github.com/rook/rook/pkg/operator/ceph/pool"
+	"github.com/rook/rook/pkg/operator/ceph/spec"
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 
@@ -680,7 +681,7 @@ func (c *ClusterController) detectAndValidateCephVersion(cluster *cluster, image
 	if err := cluster.validateCephVersion(version); err != nil {
 		return nil, false, err
 	}
-	cephver.RegisterImageVersion(image, *version)
+	c.updateClusterCephVersion(cluster.Namespace, cluster.crdName, image, *version)
 	return version, false, nil
 }
 
@@ -1049,6 +1050,26 @@ func (c *ClusterController) updateClusterStatus(namespace, name string, state ce
 	}
 }
 
+func (c *ClusterController) updateClusterCephVersion(namespace string, name string, image string, cephVersion cephver.CephVersion) {
+	logger.Infof("cluster %q: version %q detected for image %q", namespace, cephVersion.String(), image)
+
+	// get the most recent cluster CRD object
+	cluster, err := c.context.RookClientset.CephV1().CephClusters(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		logger.Errorf("failed to get cluster from namespace %q prior to updating its Ceph version to %q. %v", namespace, cephVersion.String(), err)
+	}
+	clusterVersion := &cephv1.ClusterVersion{
+		Image:   image,
+		Version: spec.GetCephVersionLabel(cephVersion),
+	}
+	// update the Ceph version on the retrieved cluster object
+	// do not overwrite the ceph status that is updated in a separate goroutine
+	cluster.Status.CephVersion = clusterVersion
+	if _, err := c.context.RookClientset.CephV1().CephClusters(namespace).Update(cluster); err != nil {
+		logger.Errorf("failed to update version for cluster %q. %v", namespace, err)
+	}
+}
+
 func ClusterOwnerRef(clusterName, clusterID string) metav1.OwnerReference {
 	blockOwner := true
 	return metav1.OwnerReference{

pkg/operator/ceph/cluster/crash/reconcile.go

@@ -26,6 +26,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/cluster/rbd"
 	"github.com/rook/rook/pkg/operator/ceph/file/mds"
 	"github.com/rook/rook/pkg/operator/ceph/object"
+	"github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/pkg/operator/ceph/version"
 
 	"github.com/coreos/pkg/capnslog"
@@ -125,9 +126,10 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
 		}
 
 		clusterImage := cephCluster.Spec.CephVersion.Image
-		cephVersion, ok := getImageVersion(clusterImage)
-		if !ok {
-			logger.Warningf("ceph version not found for image %q used by cluster %q", clusterImage, cephCluster.Name)
+		cephVersion, err := getImageVersion(cephCluster)
+		if err != nil {
+			logger.Errorf("ceph version not found for image %q used by cluster %q. %v", clusterImage, cephCluster.Name, err)
+			return reconcile.Result{}, nil
 		}
 
 		uniqueTolerations := controllerconfig.TolerationSet{}
@@ -174,14 +176,15 @@ func (r *ReconcileNode) cephPodList() ([]corev1.Pod, error) {
 }
 
 // getImageVersion returns the CephVersion registered for a specified image (if any) and whether any image was found.
-func getImageVersion(image string) (*version.CephVersion, bool) {
+func getImageVersion(cephCluster cephv1.CephCluster) (*version.CephVersion, error) {
 	for i := 0; i < getVersionMaxRetries; i++ {
-		cephVersion, ok := version.GetImageVersion(image)
-		if ok {
-			logger.Debugf("ceph version found %+v", cephVersion)
-			return cephVersion, true
+		// If the Ceph cluster has not yet recorded the image and version for the current image in its spec, then the Crash
+		// controller should wait for the version to be detected.
+		if cephCluster.Spec.CephVersion.Image == cephCluster.Status.CephVersion.Image {
+			logger.Debugf("ceph version found %q", cephCluster.Status.CephVersion.Version)
+			return spec.ExtractCephVersionFromLabel(cephCluster.Status.CephVersion.Version)
 		}
 		<-time.After(time.Second * getVersionRetryInterval)
 	}
-	return nil, false
+	return nil, errors.New("attempt to determine ceph version for the current cluster image timed out")
 }

pkg/operator/ceph/spec/label.go

@@ -38,10 +38,20 @@ const (
 func addCephVersionLabel(cephVersion version.CephVersion, labels map[string]string) {
 	// cephVersion.String() returns a string with a space in it, and labels in k8s are limited to
 	// alphanum characters plus '-', '_', '.'
-	labels[CephVersionLabelKey] = fmt.Sprintf("%d.%d.%d-%d",
+	labels[CephVersionLabelKey] = GetCephVersionLabel(cephVersion)
+}
+
+// GetCephVersionLabel returns a formatted serialization of a provided CephVersion for use in resource labels.
+func GetCephVersionLabel(cephVersion version.CephVersion) string {
+	return fmt.Sprintf("%d.%d.%d-%d",
 		cephVersion.Major, cephVersion.Minor, cephVersion.Extra, cephVersion.Build)
 }
 
+// ExtractCephVersionFromLabel returns a CephVersion struct deserialized from a provided version label.
+func ExtractCephVersionFromLabel(labelVersion string) (*version.CephVersion, error) {
+	return version.ExtractCephVersion(fmt.Sprintf("ceph version %s", labelVersion))
+}
+
 // AddCephVersionLabelToDeployment adds a label reporting the Ceph version which Rook has detected is
 // running in the Deployment's pods.
 func AddCephVersionLabelToDeployment(cephVersion version.CephVersion, d *apps.Deployment) {

pkg/operator/ceph/version/version.go

@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"regexp"
 	"strconv"
-	"sync"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -58,9 +57,6 @@ var (
 
 	// for parsing the output of `ceph --version`
 	versionPattern = regexp.MustCompile(`ceph version (\d+)\.(\d+)\.(\d+)`)
-	// for storage of the versions of images for access in managed reconciliations
-	imageToVersionMap     = map[string]CephVersion{}
-	imageToVersionMapLock = &sync.Mutex{}
 
 	// For a build release the output is "ceph version 14.2.4-64.el8cp"
 	// So we need to detect the build version change
@@ -298,18 +294,3 @@ func ValidateCephVersionsBetweenLocalAndExternalClusters(localVersion, externalV
 
 	return nil
 }
-
-// RegisterImageVersion stores the CephVersion detected for a specified image for global access.
-func RegisterImageVersion(image string, version CephVersion) {
-	imageToVersionMapLock.Lock()
-	imageToVersionMap[image] = version
-	imageToVersionMapLock.Unlock()
-}
-
-// GetImageVersion returns the CephVersion registered for a specified image (if any) and whether any image was found.
-func GetImageVersion(image string) (*CephVersion, bool) {
-	imageToVersionMapLock.Lock()
-	version, ok := imageToVersionMap[image]
-	imageToVersionMapLock.Unlock()
-	return &version, ok
-}