core: rotate fs mirror daemon cephx key

Rotate fs mirror daemon cephx keys based on the cephXconfig in the
cephcluster spec and also update the fs mirror status with the latest
cephxStatus

Signed-off-by: Santosh Pillai <sapillai@redhat.com>
(cherry picked from commit 8df6b8d0d0)

Documentation/CRDs/specification.md

@@ -1426,8 +1426,8 @@ string
 <td>
 <code>status</code><br/>
 <em>
-<a href="#ceph.rook.io/v1.Status">
-Status
+<a href="#ceph.rook.io/v1.FileMirrorStatus">
+FileMirrorStatus
 </a>
 </em>
 </td>
@@ -6857,6 +6857,51 @@ A policy can however contain multiple count-time period pairs in order to specif
 </tr>
 </tbody>
 </table>
+<h3 id="ceph.rook.io/v1.FileMirrorStatus">FileMirrorStatus
+</h3>
+<p>
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephFilesystemMirror">CephFilesystemMirror</a>)
+</p>
+<div>
+<p>FileMirrorStatus represents the status of the FileSystem mirror resource</p>
+</div>
+<table>
+<thead>
+<tr>
+<th>Field</th>
+<th>Description</th>
+</tr>
+</thead>
+<tbody>
+<tr>
+<td>
+<code>Status</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.Status">
+Status
+</a>
+</em>
+</td>
+<td>
+<p>
+(Members of <code>Status</code> are embedded into this type.)
+</p>
+</td>
+</tr>
+<tr>
+<td>
+<code>cephx</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.LocalCephxStatus">
+LocalCephxStatus
+</a>
+</em>
+</td>
+<td>
+</td>
+</tr>
+</tbody>
+</table>
 <h3 id="ceph.rook.io/v1.FilesystemMirrorInfoPeerSpec">FilesystemMirrorInfoPeerSpec
 </h3>
 <p>
@@ -8756,7 +8801,7 @@ int
 <h3 id="ceph.rook.io/v1.LocalCephxStatus">LocalCephxStatus
 </h3>
 <p>
-(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephFilesystemStatus">CephFilesystemStatus</a>, <a href="#ceph.rook.io/v1.ObjectStoreStatus">ObjectStoreStatus</a>, <a href="#ceph.rook.io/v1.RBDMirrorStatus">RBDMirrorStatus</a>)
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephFilesystemStatus">CephFilesystemStatus</a>, <a href="#ceph.rook.io/v1.FileMirrorStatus">FileMirrorStatus</a>, <a href="#ceph.rook.io/v1.ObjectStoreStatus">ObjectStoreStatus</a>, <a href="#ceph.rook.io/v1.RBDMirrorStatus">RBDMirrorStatus</a>)
 </p>
 <div>
 </div>
@@ -14050,7 +14095,7 @@ int
 <h3 id="ceph.rook.io/v1.Status">Status
 </h3>
 <p>
-(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephBucketNotification">CephBucketNotification</a>, <a href="#ceph.rook.io/v1.CephFilesystemMirror">CephFilesystemMirror</a>, <a href="#ceph.rook.io/v1.CephNFS">CephNFS</a>, <a href="#ceph.rook.io/v1.CephObjectRealm">CephObjectRealm</a>, <a href="#ceph.rook.io/v1.CephObjectZone">CephObjectZone</a>, <a href="#ceph.rook.io/v1.CephObjectZoneGroup">CephObjectZoneGroup</a>, <a href="#ceph.rook.io/v1.RBDMirrorStatus">RBDMirrorStatus</a>)
+(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephBucketNotification">CephBucketNotification</a>, <a href="#ceph.rook.io/v1.CephNFS">CephNFS</a>, <a href="#ceph.rook.io/v1.CephObjectRealm">CephObjectRealm</a>, <a href="#ceph.rook.io/v1.CephObjectZone">CephObjectZone</a>, <a href="#ceph.rook.io/v1.CephObjectZoneGroup">CephObjectZoneGroup</a>, <a href="#ceph.rook.io/v1.FileMirrorStatus">FileMirrorStatus</a>, <a href="#ceph.rook.io/v1.RBDMirrorStatus">RBDMirrorStatus</a>)
 </p>
 <div>
 <p>Status represents the status of an object</p>

deploy/charts/rook-ceph/templates/resources.yaml

@@ -7503,8 +7503,33 @@ spec:
                   type: object
               type: object
             status:
-              description: Status represents the status of an object
+              description: FileMirrorStatus represents the status of the FileSystem mirror resource
               properties:
+                cephx:
+                  properties:
+                    daemon:
+                      description: Daemon shows the CephX key status for local Ceph daemons associated with this resources.
+                      properties:
+                        keyCephVersion:
+                          description: |-
+                            KeyCephVersion reports the Ceph version that created the current generation's keys. This is
+                            same string format as reported by `CephCluster.status.version.version` to allow them to be
+                            compared. E.g., `20.2.0-0`.
+                            For all newly-created resources, this field set to the version of Ceph that created the key.
+                            The special value "Uninitialized" indicates that keys are being created for the first time.
+                            An empty string indicates that the version is unknown, as expected in brownfield deployments.
+                          type: string
+                        keyGeneration:
+                          description: |-
+                            KeyGeneration represents the CephX key generation for the last successful reconcile.
+                            For all newly-created resources, this field is set to `1`.
+                            When keys are rotated due to any rotation policy, the generation is incremented or updated to
+                            the configured policy generation.
+                            Generation `0` indicates that keys existed prior to the implementation of key tracking.
+                          format: int32
+                          type: integer
+                      type: object
+                  type: object
                 conditions:
                   items:
                     description: Condition represents a status condition on any Rook-Ceph Custom Resource.

deploy/examples/crds.yaml

@@ -7499,8 +7499,33 @@ spec:
                   type: object
               type: object
             status:
-              description: Status represents the status of an object
+              description: FileMirrorStatus represents the status of the FileSystem mirror resource
               properties:
+                cephx:
+                  properties:
+                    daemon:
+                      description: Daemon shows the CephX key status for local Ceph daemons associated with this resources.
+                      properties:
+                        keyCephVersion:
+                          description: |-
+                            KeyCephVersion reports the Ceph version that created the current generation's keys. This is
+                            same string format as reported by `CephCluster.status.version.version` to allow them to be
+                            compared. E.g., `20.2.0-0`.
+                            For all newly-created resources, this field set to the version of Ceph that created the key.
+                            The special value "Uninitialized" indicates that keys are being created for the first time.
+                            An empty string indicates that the version is unknown, as expected in brownfield deployments.
+                          type: string
+                        keyGeneration:
+                          description: |-
+                            KeyGeneration represents the CephX key generation for the last successful reconcile.
+                            For all newly-created resources, this field is set to `1`.
+                            When keys are rotated due to any rotation policy, the generation is incremented or updated to
+                            the configured policy generation.
+                            Generation `0` indicates that keys existed prior to the implementation of key tracking.
+                          format: int32
+                          type: integer
+                      type: object
+                  type: object
                 conditions:
                   items:
                     description: Condition represents a status condition on any Rook-Ceph Custom Resource.

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

@@ -3339,7 +3339,13 @@ type CephFilesystemMirror struct {
 	metav1.ObjectMeta `json:"metadata"`
 	Spec              FilesystemMirroringSpec `json:"spec"`
 	// +optional
-	Status *Status `json:"status,omitempty"`
+	Status *FileMirrorStatus `json:"status,omitempty"`
+}
+
+// FileMirrorStatus represents the status of the FileSystem mirror resource
+type FileMirrorStatus struct {
+	Status `json:",inline"`
+	Cephx  LocalCephxStatus `json:"cephx,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -1139,7 +1139,7 @@ func (in *CephFilesystemMirror) DeepCopyInto(out *CephFilesystemMirror) {
 	in.Spec.DeepCopyInto(&out.Spec)
 	if in.Status != nil {
 		in, out := &in.Status, &out.Status
-		*out = new(Status)
+		*out = new(FileMirrorStatus)
 		(*in).DeepCopyInto(*out)
 	}
 	return
@@ -2630,6 +2630,24 @@ func (in *FSMirroringSpec) DeepCopy() *FSMirroringSpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *FileMirrorStatus) DeepCopyInto(out *FileMirrorStatus) {
+	*out = *in
+	in.Status.DeepCopyInto(&out.Status)
+	out.Cephx = in.Cephx
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FileMirrorStatus.
+func (in *FileMirrorStatus) DeepCopy() *FileMirrorStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(FileMirrorStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *FilesystemMirrorInfoPeerSpec) DeepCopyInto(out *FilesystemMirrorInfoPeerSpec) {
 	*out = *in

pkg/operator/ceph/file/mirror/config.go

@@ -19,6 +19,7 @@ package mirror
 import (
 	"fmt"
 
+	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/operator/ceph/config"
 	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -58,6 +59,15 @@ func (r *ReconcileFilesystemMirror) generateKeyring(daemonConfig *daemonConfig)
 		return "", err
 	}
 
+	if r.shouldRotateCephxKeys {
+		logger.Infof("rotating CephX key for CephFileSystemMirror %q in the namespace %q", daemonConfig.ResourceName, r.clusterInfo.Namespace)
+		newKey, err := s.RotateKey(user)
+		if err != nil {
+			return "", errors.Wrapf(err, "failed to rotate CephX key for CephFileSystemMirror %q in the namespace %q", daemonConfig.ResourceName, r.clusterInfo.Namespace)
+		}
+		key = newKey
+	}
+
 	keyring := fmt.Sprintf(keyringTemplate, key)
 	return s.CreateOrUpdate(daemonConfig.ResourceName, keyring)
 }

pkg/operator/ceph/file/mirror/controller.go

@@ -27,6 +27,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/clusterd"
 	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/ceph/reporting"
 	"github.com/rook/rook/pkg/operator/k8sutil"
@@ -37,6 +38,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -70,14 +72,15 @@ var currentAndDesiredCephVersion = opcontroller.CurrentAndDesiredCephVersion
 
 // ReconcileFilesystemMirror reconciles a CephFilesystemMirror object
 type ReconcileFilesystemMirror struct {
-	context          *clusterd.Context
-	clusterInfo      *cephclient.ClusterInfo
-	client           client.Client
-	scheme           *runtime.Scheme
-	cephClusterSpec  *cephv1.ClusterSpec
-	opManagerContext context.Context
-	opConfig         opcontroller.OperatorConfig
-	recorder         record.EventRecorder
+	context               *clusterd.Context
+	clusterInfo           *cephclient.ClusterInfo
+	client                client.Client
+	scheme                *runtime.Scheme
+	cephClusterSpec       *cephv1.ClusterSpec
+	opManagerContext      context.Context
+	opConfig              opcontroller.OperatorConfig
+	recorder              record.EventRecorder
+	shouldRotateCephxKeys bool
 }
 
 // Add creates a new CephFilesystemMirror Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -154,7 +157,10 @@ func (r *ReconcileFilesystemMirror) Reconcile(context context.Context, request r
 	// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
 	reconcileResponse, cephFilesystemMirror, err := r.reconcile(request)
 	if err != nil {
-		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.FailedStatus)
+		err := r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.FailedStatus, nil)
+		if err != nil {
+			return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed set ready failure status to the cephFileSystemMirror %q", request.NamespacedName)
+		}
 		logger.Errorf("failed to reconcile %v", err)
 	}
 
@@ -181,7 +187,16 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
 
 	// The CR was just created, initializing status fields
 	if filesystemMirror.Status == nil {
-		r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.EmptyStatus)
+		cephxUninitialized := keyring.UninitializedCephxStatus()
+		err := r.updateStatus(k8sutil.ObservedGenerationNotAvailable, request.NamespacedName, k8sutil.EmptyStatus, &cephxUninitialized)
+		if err != nil {
+			return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrapf(err, "failed set empty status to the cephFileSystemMirror %q", request.NamespacedName)
+		}
+		filesystemMirror.Status = &cephv1.FileMirrorStatus{
+			Cephx: cephv1.LocalCephxStatus{
+				Daemon: cephxUninitialized,
+			},
+		}
 	}
 
 	// Make sure a CephCluster is present otherwise do nothing
@@ -232,6 +247,15 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
 	}
 	r.clusterInfo.CephVersion = *runningCephVersion
 
+	// check if cephRBDMirror daemon keys should be rotated or not
+	r.shouldRotateCephxKeys, err = keyring.ShouldRotateCephxKeys(cephCluster.Spec.Security.CephX.Daemon, *runningCephVersion, *runningCephVersion, filesystemMirror.Status.Cephx.Daemon)
+	if err != nil {
+		return reconcile.Result{}, *filesystemMirror, errors.Wrapf(err, "failed to determine if cephx keys should be rotated for the cephFileSystemMirror %q", request.NamespacedName)
+	}
+	if r.shouldRotateCephxKeys {
+		logger.Infof("cephx keys for CephFilesystemMirror %q will be rotated", request.NamespacedName)
+	}
+
 	// CREATE/UPDATE
 	logger.Debug("reconciling ceph filesystem mirror deployments")
 	reconcileResponse, err = r.reconcileFilesystemMirror(filesystemMirror)
@@ -239,9 +263,14 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
 		return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrap(err, "failed to create ceph filesystem mirror deployments")
 	}
 
+	cephxStatus := keyring.UpdatedCephxStatus(r.shouldRotateCephxKeys, r.cephClusterSpec.Security.CephX.Daemon, r.clusterInfo.CephVersion, filesystemMirror.Status.Cephx.Daemon)
+
 	// update ObservedGeneration in status at the end of reconcile
 	// Set Ready status, we are done reconciling
-	r.updateStatus(observedGeneration, request.NamespacedName, k8sutil.ReadyStatus)
+	err = r.updateStatus(observedGeneration, request.NamespacedName, k8sutil.ReadyStatus, &cephxStatus)
+	if err != nil {
+		return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrapf(err, "failed set ready status to the cephFileSystemMirror %q", request.NamespacedName)
+	}
 
 	// Return and do not requeue
 	logger.Debug("done reconciling ceph filesystem mirror")
@@ -267,29 +296,41 @@ func (r *ReconcileFilesystemMirror) reconcileFilesystemMirror(filesystemMirror *
 }
 
 // updateStatus updates an object with a given status
-func (r *ReconcileFilesystemMirror) updateStatus(observedGeneration int64, name types.NamespacedName, status string) {
+func (r *ReconcileFilesystemMirror) updateStatus(observedGeneration int64, namespacedName types.NamespacedName, status string, cephx *cephv1.CephxStatus) error {
 	fsMirror := &cephv1.CephFilesystemMirror{}
-	err := r.client.Get(r.opManagerContext, name, fsMirror)
-	if err != nil {
-		if kerrors.IsNotFound(err) {
-			logger.Debug("CephFilesystemMirror resource not found. Ignoring since object must be deleted.")
-			return
+
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		err := r.client.Get(r.opManagerContext, namespacedName, fsMirror)
+		if err != nil {
+			if kerrors.IsNotFound(err) {
+				logger.Debugf("CephFilesystemMirror resource %q not found. Ignoring since object must be deleted.", namespacedName)
+				return nil
+			}
+			return errors.Wrapf(err, "failed to retrieve filesystem mirror %q to update status to %v", namespacedName, status)
 		}
-		logger.Warningf("failed to retrieve filesystem mirror %q to update status to %q. %v", name, status, err)
-		return
+
+		if fsMirror.Status == nil {
+			fsMirror.Status = &cephv1.FileMirrorStatus{}
+		}
+
+		fsMirror.Status.Phase = status
+		if observedGeneration != k8sutil.ObservedGenerationNotAvailable {
+			fsMirror.Status.ObservedGeneration = observedGeneration
+		}
+
+		if cephx != nil {
+			fsMirror.Status.Cephx.Daemon = *cephx
+		}
+
+		if err := reporting.UpdateStatus(r.client, fsMirror); err != nil {
+			return errors.Wrapf(err, "failed to set filesystem mirror %q status to %+v", namespacedName, status)
+		}
+		return nil
+	})
+	if err != nil {
+		return err
 	}
 
-	if fsMirror.Status == nil {
-		fsMirror.Status = &cephv1.Status{}
-	}
-
-	fsMirror.Status.Phase = status
-	if observedGeneration != k8sutil.ObservedGenerationNotAvailable {
-		fsMirror.Status.ObservedGeneration = observedGeneration
-	}
-	if err := reporting.UpdateStatus(r.client, fsMirror); err != nil {
-		logger.Errorf("failed to set filesystem mirror %q status to %q. %v", fsMirror.Name, status, err)
-		return
-	}
-	logger.Debugf("filesystem mirror %q status updated to %q", name, status)
+	logger.Debugf("filesystem mirror %q status updated to %q", namespacedName, status)
+	return nil
 }

pkg/operator/ceph/file/mirror/controller_test.go

@@ -261,3 +261,223 @@ func TestCephFilesystemMirrorController(t *testing.T) {
 		assert.Equal(t, "Ready", fsMirror.Status.Phase, fsMirror)
 	})
 }
+
+func TestFSMirrorKeyRotation(t *testing.T) {
+	ctx := context.TODO()
+	var (
+		name      = "my-fs-mirror"
+		namespace = "rook-ceph"
+	)
+	// Set DEBUG logging
+	t.Setenv("ROOK_LOG_LEVEL", "DEBUG")
+
+	currentAndDesiredCephVersion = func(ctx context.Context, rookImage, namespace, jobName string, ownerInfo *k8sutil.OwnerInfo, context *clusterd.Context, cephClusterSpec *cephv1.ClusterSpec, clusterInfo *client.ClusterInfo) (*version.CephVersion, *version.CephVersion, error) {
+		rotationSupportedVer := version.CephVersion{Major: 20, Minor: 2, Extra: 0}
+		return &rotationSupportedVer, &rotationSupportedVer, nil
+	}
+
+	fsMirror := &cephv1.CephFilesystemMirror{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec:     cephv1.FilesystemMirroringSpec{},
+		TypeMeta: controllerTypeMeta,
+	}
+
+	// Mock request to simulate Reconcile() being called on an event for a
+	// watched resource .
+	req := reconcile.Request{
+		NamespacedName: types.NamespacedName{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+
+	cephCluster := &cephv1.CephCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      namespace,
+			Namespace: namespace,
+		},
+		Spec: cephv1.ClusterSpec{
+			Security: cephv1.ClusterSecuritySpec{
+				CephX: cephv1.ClusterCephxConfig{
+					Daemon: cephv1.CephxConfig{},
+				},
+			},
+		},
+		Status: cephv1.ClusterStatus{
+			Phase: k8sutil.ReadyStatus,
+			CephStatus: &cephv1.CephStatus{
+				Health: "HEALTH_OK",
+			},
+		},
+	}
+
+	// Objects to track in the fake client.
+	object := []runtime.Object{
+		fsMirror,
+		cephCluster,
+	}
+
+	s := scheme.Scheme
+
+	mirrorDaemonRotatedKey := `{"key":"AQCvzWBeIV9lFRAAninzm+8XFxbSfTiPwoX50g=="}`
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+			if args[0] == "status" {
+				return `{"fsid":"c47cac40-9bee-4d52-823b-ccd803ba5bfe","health":{"checks":{},"status":"HEALTH_ERR"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
+			}
+			if args[0] == "auth" && args[1] == "get-or-create-key" {
+				return cephAuthGetOrCreateKey, nil
+			}
+			if args[0] == "auth" && args[1] == "rotate" {
+				t.Logf("rotating key and returning: %s", mirrorDaemonRotatedKey)
+				return mirrorDaemonRotatedKey, nil
+			}
+			return "", nil
+		},
+	}
+	clientset := test.New(t, 3)
+	c := &clusterd.Context{
+		Executor:      executor,
+		RookClientset: rookclient.NewSimpleClientset(),
+		Clientset:     clientset,
+	}
+
+	// Register operator types with the runtime scheme.
+	s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephRBDMirror{})
+	s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephCluster{})
+
+	// Create a fake client to mock API calls.
+	cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
+	r := &ReconcileFilesystemMirror{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
+
+	t.Run("first reconcile", func(t *testing.T) {
+		// Mock clusterInfo
+		secrets := map[string][]byte{
+			"fsid":         []byte(name),
+			"mon-secret":   []byte("monsecret"),
+			"admin-secret": []byte("adminsecret"),
+		}
+		secret := &v1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "rook-ceph-mon",
+				Namespace: namespace,
+			},
+			Data: secrets,
+			Type: k8sutil.RookType,
+		}
+		_, err := c.Clientset.CoreV1().Secrets(namespace).Create(ctx, secret, metav1.CreateOptions{})
+		assert.NoError(t, err)
+
+		_, err = r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, uint32(1), fsMirror.Status.Cephx.Daemon.KeyGeneration)
+		assert.Equal(t, "20.2.0-0", fsMirror.Status.Cephx.Daemon.KeyCephVersion)
+	})
+
+	t.Run("subsequent reconcile - retain cephx status", func(t *testing.T) {
+		r := &ReconcileFilesystemMirror{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
+		_, err := r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, uint32(1), fsMirror.Status.Cephx.Daemon.KeyGeneration)
+		assert.Equal(t, "20.2.0-0", fsMirror.Status.Cephx.Daemon.KeyCephVersion)
+	})
+
+	t.Run("brownfield reconcile - retain unknown cephx status", func(t *testing.T) {
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err := cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		fsMirror.Status.Cephx.Daemon = cephv1.CephxStatus{}
+		err = cl.Update(ctx, &fsMirror)
+		assert.NoError(t, err)
+
+		_, err = r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, cephv1.CephxStatus{}, fsMirror.Status.Cephx.Daemon)
+	})
+
+	t.Run("rotate key - brownfield unknown status becomes known", func(t *testing.T) {
+		cluster := cephv1.CephCluster{}
+		err := cl.Get(ctx, types.NamespacedName{Namespace: namespace, Name: namespace}, &cluster)
+		assert.NoError(t, err)
+		cluster.Spec.Security.CephX.Daemon = cephv1.CephxConfig{
+			KeyRotationPolicy: "KeyGeneration",
+			KeyGeneration:     2,
+		}
+		err = cl.Update(ctx, &cluster)
+		assert.NoError(t, err)
+
+		mirrorDaemonRotatedKey = `[{"key":"BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=="}]`
+
+		_, err = r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, uint32(2), fsMirror.Status.Cephx.Daemon.KeyGeneration)
+		assert.Equal(t, "20.2.0-0", fsMirror.Status.Cephx.Daemon.KeyCephVersion)
+
+		secret, err := clientset.CoreV1().Secrets(namespace).Get(ctx, "rook-ceph-fs-mirror-keyring", metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.Contains(t, secret.StringData["keyring"], "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB==")
+	})
+
+	t.Run("brownfield reconcile - no further rotation happens", func(t *testing.T) {
+		// not expecting any rotation. So `ceph auth rotate` should not run and secret should not be updated
+		mirrorDaemonRotatedKey = `[{"key":"CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC=="}]`
+
+		res, err := r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+		assert.False(t, res.Requeue)
+
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, uint32(2), fsMirror.Status.Cephx.Daemon.KeyGeneration)
+		assert.Equal(t, "20.2.0-0", fsMirror.Status.Cephx.Daemon.KeyCephVersion)
+
+		secret, err := clientset.CoreV1().Secrets(namespace).Get(ctx, "rook-ceph-fs-mirror-keyring", metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.NotContains(t, secret.StringData["keyring"], "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC==")
+	})
+
+	t.Run("rotate key - cephx status updated", func(t *testing.T) {
+		cluster := cephv1.CephCluster{}
+		err := cl.Get(ctx, types.NamespacedName{Namespace: namespace, Name: namespace}, &cluster)
+		assert.NoError(t, err)
+		cluster.Spec.Security.CephX.Daemon = cephv1.CephxConfig{
+			KeyRotationPolicy: "KeyGeneration",
+			KeyGeneration:     3,
+		}
+		err = cl.Update(ctx, &cluster)
+		assert.NoError(t, err)
+
+		mirrorDaemonRotatedKey = `[{"key":"CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC=="}]`
+
+		_, err = r.Reconcile(ctx, req)
+		assert.NoError(t, err)
+
+		fsMirror := cephv1.CephFilesystemMirror{}
+		err = cl.Get(ctx, req.NamespacedName, &fsMirror)
+		assert.NoError(t, err)
+		assert.Equal(t, uint32(3), fsMirror.Status.Cephx.Daemon.KeyGeneration)
+		assert.Equal(t, "20.2.0-0", fsMirror.Status.Cephx.Daemon.KeyCephVersion)
+
+		secret, err := clientset.CoreV1().Secrets(namespace).Get(ctx, "rook-ceph-fs-mirror-keyring", metav1.GetOptions{})
+		assert.NoError(t, err)
+		assert.Contains(t, secret.StringData["keyring"], "CCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCCC==")
+	})
+}

pkg/operator/ceph/file/mirror/mirror.go

@@ -23,6 +23,7 @@ import (
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/ceph/config"
+	"github.com/rook/rook/pkg/operator/ceph/config/keyring"
 	"github.com/rook/rook/pkg/operator/ceph/controller"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -54,7 +55,7 @@ func (r *ReconcileFilesystemMirror) start(filesystemMirror *cephv1.CephFilesyste
 		ownerInfo:    ownerInfo,
 	}
 
-	_, err = r.generateKeyring(daemonConf)
+	secretResourceVersion, err := r.generateKeyring(daemonConf)
 	if err != nil {
 		return errors.Wrapf(err, "failed to generate keyring for %q", AppName)
 	}
@@ -65,6 +66,9 @@ func (r *ReconcileFilesystemMirror) start(filesystemMirror *cephv1.CephFilesyste
 		return errors.Wrap(err, "failed to create filesystem-mirror deployment")
 	}
 
+	// apply cephx secret resource version to the deployment to ensure it restarts when keyring updates
+	d.Spec.Template.Annotations[keyring.CephxKeyIdentifierAnnotation] = secretResourceVersion
+
 	// Set owner ref to filesystemMirror object
 	err = controllerutil.SetControllerReference(filesystemMirror, d, r.scheme)
 	if err != nil {