diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go
index 689f1ab8d..1be9e5533 100644
--- a/pkg/operator/ceph/cluster/controller.go
+++ b/pkg/operator/ceph/cluster/controller.go
@@ -682,6 +682,7 @@ func (c *ClusterController) detectAndValidateCephVersion(cluster *cluster, image
 	if err := cluster.validateCephVersion(version); err != nil {
 		return nil, false, err
 	}
+	cephver.RegisterImageVersion(image, *version)
 	return version, false, nil
 }
 
diff --git a/pkg/operator/ceph/cluster/crash/crash.go b/pkg/operator/ceph/cluster/crash/crash.go
index 94a59afb0..7d545d86f 100644
--- a/pkg/operator/ceph/cluster/crash/crash.go
+++ b/pkg/operator/ceph/cluster/crash/crash.go
@@ -28,6 +28,7 @@ import (
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
+	"github.com/rook/rook/pkg/operator/ceph/version"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
@@ -50,7 +51,7 @@ var clusterResource = opkit.CustomResource{
 }
 
 // createOrUpdateCephCrash is a wrapper around controllerutil.CreateOrUpdate
-func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []corev1.Toleration, cephCluster cephv1.CephCluster) (controllerutil.OperationResult, error) {
+func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []corev1.Toleration, cephCluster cephv1.CephCluster, cephVersion *version.CephVersion) (controllerutil.OperationResult, error) {
 	// Create or Update the deployment default/foo
 	nodeHostnameLabel, ok := node.ObjectMeta.Labels[corev1.LabelHostname]
 	if !ok {
@@ -86,6 +87,10 @@ func (r *ReconcileNode) createOrUpdateCephCrash(node corev1.Node, tolerations []
 		}
 
 		deploy.ObjectMeta.Labels = deploymentLabels
+		k8sutil.AddRookVersionLabelToDeployment(deploy)
+		if cephVersion != nil {
+			opspec.AddCephVersionLabelToDeployment(*cephVersion, deploy)
+		}
 		deploy.Spec.Template = corev1.PodTemplateSpec{
 			ObjectMeta: metav1.ObjectMeta{
 				Labels: deploymentLabels,
diff --git a/pkg/operator/ceph/cluster/crash/reconcile.go b/pkg/operator/ceph/cluster/crash/reconcile.go
index f8e613a9a..8e1f20bf1 100644
--- a/pkg/operator/ceph/cluster/crash/reconcile.go
+++ b/pkg/operator/ceph/cluster/crash/reconcile.go
@@ -18,6 +18,7 @@ package crash
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mgr"
@@ -25,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/version"
 
 	"github.com/coreos/pkg/capnslog"
 
@@ -41,6 +43,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
+const (
+	getVersionRetryInterval = 5
+	getVersionMaxRetries    = 60
+)
+
 var (
 	logger = capnslog.NewPackageLogger("github.com/rook/rook", controllerName)
 	// Implement reconcile.Reconciler so the controller can reconcile objects
@@ -117,6 +124,12 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
 			logger.Errorf("more than one CephCluster found in the namespace %q, choosing the first one %q", namespace, cephCluster.GetName())
 		}
 
+		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)
+		}
+
 		uniqueTolerations := controllerconfig.TolerationSet{}
 		hasCephPods := false
 		for _, cephPod := range cephPods {
@@ -132,7 +145,7 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
 
 		if hasCephPods {
 			tolerations := uniqueTolerations.ToList()
-			op, err := r.createOrUpdateCephCrash(*node, tolerations, cephCluster)
+			op, err := r.createOrUpdateCephCrash(*node, tolerations, cephCluster, cephVersion)
 			if err != nil {
 				return reconcile.Result{}, errors.Wrapf(err, "node reconcile failed on op %q", op)
 			}
@@ -159,3 +172,15 @@ func (r *ReconcileNode) cephPodList() ([]corev1.Pod, error) {
 
 	return cephPods, nil
 }
+
+// getImageVersion returns the CephVersion registered for a specified image (if any) and whether any image was found.
+func getImageVersion(image string) (*version.CephVersion, bool) {
+	for i := 0; i < getVersionMaxRetries; i++ {
+		cephVersion, ok := version.GetImageVersion(image)
+		if ok {
+			return cephVersion, true
+		}
+		<-time.After(getVersionRetryInterval)
+	}
+	return nil, false
+}
diff --git a/pkg/operator/ceph/version/version.go b/pkg/operator/ceph/version/version.go
index 752d03819..45248eeed 100644
--- a/pkg/operator/ceph/version/version.go
+++ b/pkg/operator/ceph/version/version.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"regexp"
 	"strconv"
+	"sync"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -57,6 +58,9 @@ 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
@@ -294,3 +298,18 @@ 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
+}