Merge pull request #16398 from rook/mergify/bp/release-1.18/pr-16271

core: rotate client.admin cephx key (backport #16271)

Documentation/CRDs/specification.md

@@ -5101,6 +5101,19 @@ CSI key rotation can affect existing PV connections, so take care when exercisin
 <tbody>
 <tr>
 <td>
+<code>admin</code><br/>
+<em>
+<a href="#ceph.rook.io/v1.CephxStatus">
+CephxStatus
+</a>
+</em>
+</td>
+<td>
+<p>Admin shows the CephX key status for the client.admin key</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>mon</code><br/>
 <em>
 <a href="#ceph.rook.io/v1.CephxStatus">

Documentation/Storage-Configuration/Advanced/cephx-key-rotation.md

@@ -93,6 +93,9 @@ type managed for the cluster is listed.
 status:
   # ...
   cephx:
+    admin:
+      keyCephVersion: 19.2.3-0
+      keyGeneration: 2
     cephExporter:
       keyCephVersion: 19.2.3-0
       keyGeneration: 2

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

@@ -5990,6 +5990,28 @@ spec:
                 cephx:
                   description: ClusterCephxStatus defines the cephx key rotation status of various daemons on the cephCluster resource
                   properties:
+                    admin:
+                      description: Admin shows the CephX key status for the client.admin key
+                      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
                     cephExporter:
                       description: Ceph Exporter represents the cephx key rotation status of the ceph exporter daemon
                       properties:

deploy/examples/crds.yaml

@@ -5988,6 +5988,28 @@ spec:
                 cephx:
                   description: ClusterCephxStatus defines the cephx key rotation status of various daemons on the cephCluster resource
                   properties:
+                    admin:
+                      description: Admin shows the CephX key status for the client.admin key
+                      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
                     cephExporter:
                       description: Ceph Exporter represents the cephx key rotation status of the ceph exporter daemon
                       properties:

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

@@ -748,6 +748,8 @@ type LocalCephxStatus struct {
 
 // ClusterCephxStatus defines the cephx key rotation status of various daemons on the cephCluster resource
 type ClusterCephxStatus struct {
+	// Admin shows the CephX key status for the client.admin key
+	Admin CephxStatus `json:"admin,omitempty"`
 	// Mon represents the CephX key status of the Monitor daemons
 	Mon CephxStatus `json:"mon,omitempty"`
 	// Mgr represents the cephx key rotation status of the ceph manager daemon

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

@@ -2077,6 +2077,7 @@ func (in *ClusterCephxConfig) DeepCopy() *ClusterCephxConfig {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterCephxStatus) DeepCopyInto(out *ClusterCephxStatus) {
 	*out = *in
+	out.Admin = in.Admin
 	out.Mon = in.Mon
 	out.Mgr = in.Mgr
 	out.OSD = in.OSD

pkg/daemon/ceph/client/command.go

@@ -95,11 +95,15 @@ func FinalizeCephCommandArgs(command string, clusterInfo *ClusterInfo, args []st
 	default:
 		// Append the standard flags for config and keyring
 		keyringFile := fmt.Sprintf("%s.keyring", clusterInfo.CephCred.Username)
+		keyringPath := path.Join(configDir, clusterInfo.Namespace, keyringFile)
+		if clusterInfo.KeyringFileOverride != "" {
+			keyringPath = clusterInfo.KeyringFileOverride
+		}
 		configArgs = []string{
 			fmt.Sprintf("--cluster=%s", clusterInfo.Namespace),
 			fmt.Sprintf("--conf=%s", cephConfPath),
 			fmt.Sprintf("--name=%s", clusterInfo.CephCred.Username),
-			fmt.Sprintf("--keyring=%s", path.Join(configDir, clusterInfo.Namespace, keyringFile)),
+			fmt.Sprintf("--keyring=%s", keyringPath),
 		}
 	}
 

pkg/daemon/ceph/client/command_test.go

@@ -47,9 +47,19 @@ func TestFinalizeCephCommandArgs(t *testing.T) {
 	}
 
 	clusterInfo := AdminTestClusterInfo("rook")
-	cmd, args := FinalizeCephCommandArgs(expectedCommand, clusterInfo, args, configDir)
+	cmd, gotArgs := FinalizeCephCommandArgs(expectedCommand, clusterInfo, args, configDir)
 	assert.Exactly(t, expectedCommand, cmd)
-	assert.Exactly(t, expectedArgs, args)
+	assert.Exactly(t, expectedArgs, gotArgs)
+
+	t.Run("keyring override", func(t *testing.T) {
+		// run the same test as above, but set the keyring override
+		clusterInfo.KeyringFileOverride = "/some/random/file/path.extension"
+		// keyring arg should change but nothing else
+		expectedArgs[len(expectedArgs)-1] = "--keyring=/some/random/file/path.extension"
+		cmd, args := FinalizeCephCommandArgs(expectedCommand, clusterInfo, args, configDir)
+		assert.Exactly(t, expectedCommand, cmd)
+		assert.Exactly(t, expectedArgs, args)
+	})
 }
 
 func TestFinalizeRadosGWAdminCommandArgs(t *testing.T) {

pkg/daemon/ceph/client/config.go

@@ -93,7 +93,7 @@ func GenerateConnectionConfig(context *clusterd.Context, cluster *ClusterInfo) (
 func GenerateConnectionConfigWithSettings(context *clusterd.Context, clusterInfo *ClusterInfo, settings *CephConfig) (string, error) {
 	root := path.Join(context.ConfigDir, clusterInfo.Namespace)
 	keyringPath := path.Join(root, fmt.Sprintf("%s.keyring", clusterInfo.CephCred.Username))
-	err := writeKeyring(CephKeyring(clusterInfo.CephCred), keyringPath)
+	err := WriteKeyring(keyringPath, CephKeyring(clusterInfo.CephCred))
 	if err != nil {
 		return "", errors.Wrapf(err, "failed to write keyring %q to %s", clusterInfo.CephCred.Username, root)
 	}

pkg/daemon/ceph/client/info.go

@@ -60,6 +60,10 @@ type ClusterInfo struct {
 	// Whereas if passed through clusterInfo, we don't have that problem since clusterInfo is
 	// re-hydrated when a context is cancelled.
 	Context context.Context
+
+	// KeyringFileOverride, if set, configures a ceph client executor to use a specific keyring file
+	// instead of the default.
+	KeyringFileOverride string
 }
 
 func (c *ClusterInfo) AllMonitors() map[string]*MonInfo {

pkg/daemon/ceph/client/keyring.go

@@ -53,15 +53,6 @@ func CephKeyring(cred CephCred) string {
 	return fmt.Sprintf(UserKeyringTemplate, cred.Username, cred.Secret)
 }
 
-// WriteKeyring calls the generate contents function with auth key as an argument then saves the
-// output of the generateContents function to disk at the keyring path
-// TODO: Kludgey; can keyring files be generated w/ go-ini package or using the '-o' option to
-// 'ceph auth get-or-create ...'?
-func WriteKeyring(keyringPath, authKey string, generateContents func(string) string) error {
-	contents := generateContents(authKey)
-	return writeKeyring(contents, keyringPath)
-}
-
 // CreateKeyring creates a keyring for access to the cluster with the desired set of privileges
 // and writes it to disk at the keyring path
 func CreateKeyring(context *clusterd.Context, clusterInfo *ClusterInfo, username, keyringPath string, access []string, generateContents func(string) string) error {
@@ -81,12 +72,12 @@ func CreateKeyring(context *clusterd.Context, clusterInfo *ClusterInfo, username
 		return errors.Wrapf(err, "failed to get or create auth key for %s", username)
 	}
 
-	return WriteKeyring(keyringPath, key, generateContents)
+	contents := generateContents(key)
+	return WriteKeyring(keyringPath, contents)
 }
 
-// writes the keyring to disk
-// TODO: Write keyring only to the default ceph config location since we are in a container
-func writeKeyring(keyring, keyringPath string) error {
+// Writes the keyring to disk, creating parent dirs as necessary
+func WriteKeyring(keyringPath, keyring string) error {
 	// save the keyring to the given path
 	if err := os.MkdirAll(filepath.Dir(keyringPath), 0o700); err != nil {
 		return errors.Wrapf(err, "failed to create keyring directory for %s", keyringPath)

pkg/operator/ceph/cluster/cephx.go

@@ -0,0 +1,422 @@
+/*
+Copyright 2025 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cluster
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sync"
+
+	"github.com/pkg/errors"
+	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/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/ceph/reporting"
+	"github.com/rook/rook/pkg/operator/k8sutil"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/client-go/util/retry"
+)
+
+const (
+	//#nosec G101 -- This is only a secret name prefix
+	adminRotatorSecretPrefix = "rook-ceph-admin-rotator"
+	//#nosec G101 -- This is only the admin client name
+	adminRotatorUsername = "client.admin-rotator"
+)
+
+// note: must match adminKeyringTemplate
+var (
+	adminKeyAccessCaps = []string{"mds", "allow *", "mon", "allow *", "osd", "allow *", "mgr", "allow *"}
+
+	// ReloadManager sends SIGHUP to the main process. Allow this to be stubbed for unit tests.
+	reloadManagerFunc = controller.ReloadManager
+
+	// admin key rotation ends with a rook restart, so return a reconcile error after to help
+	// prevent rook from continuing reconcile when restart is imminent
+	errSuccessfulAdminKeyRotation = fmt.Errorf("successful admin cephx key rotation requires the current cluster reconcile to restart")
+
+	// Prevent multiple admin rotation operations from running simultaneously. This is important
+	// because the admin rotation routine does not stop as soon as the parent context is canceled.
+	// in the worst case, it could still be running while a CephCluster re-reconcile gets back
+	// around to admin key rotation again.
+	adminRotationMutex      = sync.Mutex{}
+	adminRotationInProgress = map[string]struct{}{}
+)
+
+// claimAdminRotationLock tries to claim a lock on admin key rotation for a cluster namespace
+// returns error if the lock could not be granted.
+// Because (*Mutex).Lock() blocks while trying to get a lock, we could end up stalling Rook
+// reconciles if we rely solely on the lock itself. To prevent stalls, use a map 'set' to indicate
+// whether rotation is in progress, and only lock access to the set. When a rotation is locked,
+// return an error so that the reconcile can be requeued to try again later.
+func claimAdminRotationLock(namespace string) error {
+	adminRotationMutex.Lock()
+	defer adminRotationMutex.Unlock()
+	if _, ok := adminRotationInProgress[namespace]; ok {
+		return fmt.Errorf("failed to claim admin rotation lock for cluster in namespace %q", namespace)
+	}
+	logger.Debugf("claimed admin rotation lock for cluster in namespace %q", namespace)
+	adminRotationInProgress[namespace] = struct{}{}
+	return nil
+}
+
+// releaseAdminRotationLock releases a claimed lock for a cluster namespace
+func releaseAdminRotationLock(namespace string) {
+	adminRotationMutex.Lock()
+	defer adminRotationMutex.Unlock()
+	delete(adminRotationInProgress, namespace)
+}
+
+// turn a client name and its caps into keyring file contents
+// example: client.my-user, []string{"mon", "allow *"} becomes:
+//
+//	[client.my-user]
+//		caps mon = "allow *"
+func genKeyring(clientName, authKey string, clientCaps []string) (string, error) {
+	// DO NOT LOG authKey
+	if authKey == "" {
+		return "", fmt.Errorf("cannot generate %q keyring with empty cephx key", clientName)
+	}
+	o := fmt.Sprintf("[%s]\n", clientName)
+	o += fmt.Sprintf("	key = %s\n", authKey)
+	if len(clientCaps)%2 != 0 {
+		return "", fmt.Errorf("cannot generate %q keyring for caps list %v with uneven number of items", clientName, clientCaps)
+	}
+	for i := 0; i < len(clientCaps); i += 2 {
+		o += fmt.Sprintf("	caps %s = %q\n", clientCaps[i], clientCaps[i+1])
+	}
+	return o, nil
+}
+
+// main routine for rotating the client.admin key
+// should be called after mon daemons are upgraded to current ceph image in the cephcluster spec
+func rotateAdminCephxKey(
+	clusterdCtx *clusterd.Context,
+	clusterInfo *cephclient.ClusterInfo,
+	ownerInfo *k8sutil.OwnerInfo,
+	cephCluster *cephv1.CephCluster,
+) error {
+	if clusterInfo.CephCred.Username != cephclient.AdminUsername {
+		// this shouldn't happen during normal runtime - this could indicate external mode cluster
+		logger.Infof("cannot rotate admin cephx key with non-rotatable username %q for cluster in namespace %q", clusterInfo.CephCred.Username, clusterInfo.Namespace)
+		return nil
+	}
+
+	desiredCephVersion := clusterInfo.CephVersion // TODO: update this when/if WithCephVersionUpdate is implemented
+	shouldRotate, err := keyring.ShouldRotateCephxKeys(
+		cephCluster.Spec.Security.CephX.Daemon, clusterInfo.CephVersion, desiredCephVersion, cephCluster.Status.Cephx.Admin)
+	if err != nil {
+		return errors.Wrap(err, "failed to determine if admin cephx key should be rotated")
+	}
+	if !shouldRotate {
+		logger.Debugf("not rotating admin cephx key for cluster in namespace %q", clusterInfo.Namespace)
+		// no rotation, but still update cephx status - critical for greenfield Uninitialized clusters
+		err := updateCephClusterAdminCephxStatus(clusterdCtx, clusterInfo, false)
+		if err != nil {
+			return errors.Wrap(err, "failed to update admin cephx key status after not rotating")
+		}
+		return nil
+	}
+
+	if err := claimAdminRotationLock(clusterInfo.Namespace); err != nil {
+		return err
+	}
+	defer releaseAdminRotationLock(clusterInfo.Namespace)
+
+	logger.Infof("beginning admin cephx key rotation for cluster in namespace %q", clusterInfo.Namespace)
+
+	// As an optimization, use a (shallow) copy of clusterInfo with a context that won't be canceled
+	// during normal runtime. This disallows the rotation process from being interrupted partway
+	// through in the event of a CephCluster spec change, which can happen for any number of reasons
+	// during normal runtime. While this rotation is designed to be able to be recovered if
+	// interrupted, it is still safest to allow the rotation process to finish in full.
+	clusterInfoShallowCopy := *clusterInfo
+	clusterInfoCopy := &clusterInfoShallowCopy
+	clusterInfoCopy.Context = context.Background()
+
+	s := keyring.GetSecretStore(clusterdCtx, clusterInfoCopy, ownerInfo)
+
+	// generate client.admin-rotator admin user
+	// if client.admin rotation fails or is blocked, the client.admin-rotator user can be used to
+	// recover from bugs/blockages in rotation of primary client.admin key
+	rotatorKey, err := s.GenerateKey(adminRotatorUsername, adminKeyAccessCaps)
+	if err != nil {
+		return errors.Wrapf(err, "failed to generate cephx key for admin rotator %q", adminRotatorUsername)
+	}
+
+	// store client.admin-rotator in secret
+	rotatorKeyring, err := genKeyring(adminRotatorUsername, rotatorKey, adminKeyAccessCaps)
+	if err != nil {
+		return errors.Wrapf(err, "failed to generate admin rotator cephx keyring file")
+	}
+	_, err = s.CreateOrUpdate(adminRotatorSecretPrefix, rotatorKeyring)
+	if err != nil {
+		return errors.Wrapf(err, "failed to store admin rotator cephx keyring to secret")
+	}
+
+	// keep keyring files used for admin key rotation in a subdir intended to be temporary
+	// stored in rook-ceph-operator emptyDir
+	// do not clean up in defer() so that any issues can be debugged with shell into operator pod
+	tmpDir := adminRotationTmpDir(clusterdCtx, clusterInfoCopy)
+
+	return rotateAdminCephxKeyUsingRotator(clusterdCtx, clusterInfoCopy, ownerInfo, tmpDir, rotatorKeyring)
+}
+
+// routine for recovering from client.admin key rotation that was interrupted in a prior reconcile
+// because the currently-stored client.admin key could be incorrect, this should be called very
+// early in the cephcluster reconcile, before any ceph commands are run: failures running commands
+// could block the reconcile from recovering admin key rotation
+func recoverPriorAdminCephxKeyRotation(
+	clusterdCtx *clusterd.Context,
+	clusterInfo *cephclient.ClusterInfo,
+	ownerInfo *k8sutil.OwnerInfo,
+	clusterNamespace string,
+) error {
+	if clusterInfo == nil {
+		// new cluster hasn't been deployed yet. there should be no way for recovery to be needed
+		logger.Debugf("no admin cephx key recovery possible for cluster with empty clusterInfo in namespace %q", clusterNamespace)
+		return nil
+	}
+
+	s := keyring.GetSecretStore(clusterdCtx, clusterInfo, ownerInfo)
+
+	rotatorKeyring, err := s.GetKeyringFromSecret(adminRotatorSecretPrefix)
+	if err != nil {
+		if kerrors.IsNotFound(err) {
+			logger.Debugf("no admin cephx key recovery needed for cluster in namespace %q", clusterInfo.Namespace)
+			return nil
+		}
+		return errors.Wrap(err, "failed to get admin rotator cephx key secret to check if admin cephx key rotation recovery is needed")
+	}
+
+	if err := claimAdminRotationLock(clusterNamespace); err != nil {
+		return err
+	}
+	defer releaseAdminRotationLock(clusterNamespace)
+
+	logger.Infof("recovering from interrupted admin cephx key rotation for cluster in namespace %q", clusterInfo.Namespace)
+
+	// As an optimization, use a (shallow) copy of clusterInfo with a context that won't be canceled
+	// during normal runtime. This disallows the recovery process from being interrupted partway
+	// through in the event of a CephCluster spec change, which can happen for any number of reasons
+	// during normal runtime. While even the recovery is designed to be able to be recovered if
+	// interrupted, it is still safest to allow the rotation process to recover in full.
+	clusterInfoShallowCopy := *clusterInfo
+	clusterInfoCopy := &clusterInfoShallowCopy
+	clusterInfoCopy.Context = context.Background()
+
+	// if the operator pod has restarted and recovery is needed, the ceph.conf file will not be
+	// present in the rook temp dir. Write/update the cluster config to handle this case
+	err = mon.WriteConnectionConfig(clusterdCtx, clusterInfoCopy)
+	if err != nil {
+		return errors.Wrap(err, "failed to write/update cluster config in ceph.conf before recovering admin cephx key rotation")
+	}
+
+	// keep keyring files used for admin key rotation in a subdir intended to be temporary
+	// stored in rook-ceph-operator emptyDir
+	// do not clean up in defer() so that any issues can be debugged with shell into operator pod
+	tmpDir := adminRotationTmpDir(clusterdCtx, clusterInfoCopy)
+
+	authList, err := cephclient.AuthList(clusterdCtx, clusterInfoCopy)
+	if err != nil {
+		logger.Debugf("primary admin user failed to list ceph auth - this is an expected condition when recovering from admin cephx key rotation")
+	} else if !entityExistsInAuthList(adminRotatorUsername, authList) { // err == nil
+		// corner case: if the primary client.admin works AND client.admin-rotator is not present in
+		// the `auth ls` output, it means rotation succeeded and was interrupted during cleanup.
+		// client.auth-rotator is no longer present/valid, so it can't be used to take any actions
+		return finalizeAdminKeyRotation(clusterdCtx, clusterInfoCopy, ownerInfo, tmpDir)
+	}
+
+	// a prior rotation failed, but we don't know exactly where in the process it failed. we have to
+	// re-run the main rotation routine from the beginning to ensure we don't skip rotation
+	return rotateAdminCephxKeyUsingRotator(clusterdCtx, clusterInfoCopy, ownerInfo, tmpDir, rotatorKeyring)
+}
+
+func entityExistsInAuthList(entity string, authList cephclient.AuthListOutput) bool {
+	for _, a := range authList.AuthDump {
+		if a.Entity == entity {
+			return true
+		}
+	}
+	return false
+}
+
+// helper routine that rotate client.admin cephx key including cleanup and cephx status update
+// uses the provided client.admin-rotator keyring to rotate client.admin
+func rotateAdminCephxKeyUsingRotator(
+	clusterdCtx *clusterd.Context,
+	clusterInfo *cephclient.ClusterInfo,
+	ownerInfo *k8sutil.OwnerInfo,
+	tmpDir, rotatorKeyring string,
+) error {
+	// write admin rotator keyring file to tempdir so it can be used for rotation
+	rotatorKeyringTmpfile := filepath.Join(tmpDir, adminRotatorUsername+".keyring")
+	logger.Infof("temporarily storing admin rotator keyring at %q for cluster in namespace %q", rotatorKeyringTmpfile, clusterInfo.Namespace)
+	err := cephclient.WriteKeyring(rotatorKeyringTmpfile, rotatorKeyring)
+	if err != nil {
+		return errors.Wrapf(err, "failed to write admin rotator keyring to %q", rotatorKeyringTmpfile)
+	}
+
+	// make copy of ClusterInfo struct that uses admin-rotator user/key for ceph commands
+	rotatorInfo := minimalCopyClusterInfo(clusterInfo, adminRotatorUsername, rotatorKeyringTmpfile)
+
+	// as `client.admin-rotator`: run `ceph auth ls` to ensure it has permissions
+	_, err = cephclient.AuthList(clusterdCtx, rotatorInfo)
+	if err != nil {
+		return errors.Wrapf(err, "failed to list ceph auth using admin rotator client")
+	}
+
+	// as `client.admin-rotator`: run `ceph auth rotate client.admin`
+	logger.Infof("admin cephx key will be rotated for cluster in namespace %q. rook will restart afterwards. some reconciles and health checks may fail in between - this is normal", clusterInfo.Namespace)
+	newAdminKey, err := cephclient.AuthRotate(clusterdCtx, rotatorInfo, cephclient.AdminUsername)
+	if err != nil {
+		return errors.Wrapf(err, "failed to rotate admin key using admin rotator client")
+	}
+	newAdminKeyring := cephclient.CephKeyring(cephclient.CephCred{Username: cephclient.AdminUsername, Secret: newAdminKey})
+
+	// write new admin keyring to tempdir so it can be verified before it's stored permanently
+	newAdminKeyringTmpfile := filepath.Join(tmpDir, cephclient.AdminUsername+".keyring")
+	logger.Infof("temporarily storing rotated admin cephx keyring at %q for cluster in namespace %q", newAdminKeyringTmpfile, clusterInfo.Namespace)
+	err = cephclient.WriteKeyring(newAdminKeyringTmpfile, newAdminKeyring)
+	if err != nil {
+		return errors.Wrapf(err, "failed to write admin rotator keyring to %q", newAdminKeyringTmpfile)
+	}
+
+	// `client.admin` run `ceph auth ls` to ensure it has permissions
+	// make copy of ClusterInfo struct that uses rotated admin's temporary keyring for ceph commands
+	newAdminInfo := minimalCopyClusterInfo(clusterInfo, cephclient.AdminUsername, newAdminKeyringTmpfile)
+	_, err = cephclient.AuthList(clusterdCtx, newAdminInfo)
+	if err != nil {
+		return errors.Wrapf(err, "admin rotator failed to list ceph auth")
+	}
+
+	// now that the new key is verified working, update clusterInfo with new key
+	clusterInfo.CephCred.Secret = newAdminKey
+
+	// update the on-disk ceph config file with new `client.admin` key (now in clusterInfo)
+	err = mon.WriteConnectionConfig(clusterdCtx, clusterInfo)
+	if err != nil {
+		return errors.Wrapf(err, "failed to write latest cluster config to disk")
+	}
+
+	// update the `rook-ceph-mon` secret with new `client.admin` key (now in clusterInfo)
+	err = controller.UpdateClusterAccessSecret(clusterdCtx.Clientset, clusterInfo)
+	if err != nil {
+		return errors.Wrapf(err, "failed to update cluster access secret with latest %q admin cephx key", cephclient.AdminUsername)
+	}
+
+	// as `client.admin`: run `ceph auth rm client.admin-rotator`
+	err = cephclient.AuthDelete(clusterdCtx, clusterInfo, adminRotatorUsername)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete cephx auth for admin rotator %q", adminRotatorUsername)
+	}
+
+	return finalizeAdminKeyRotation(clusterdCtx, clusterInfo, ownerInfo, tmpDir)
+}
+
+// helper routine that performs final cleanup steps for admin key rotation. should be used
+// immediately after `ceph auth rm client.admin-rotator`. updates cephx status when finished
+func finalizeAdminKeyRotation(
+	clusterdCtx *clusterd.Context,
+	clusterInfo *cephclient.ClusterInfo,
+	ownerInfo *k8sutil.OwnerInfo,
+	tmpDir string,
+) error {
+	// clean up tmpDir now that it's no longer useful for recovery
+	if err := os.RemoveAll(tmpDir); err != nil {
+		// most likely, operator pod got rescheduled to a different node where dir isn't present
+		logger.Infof("non-critical failure removing admin cephx key rotation temp dir %q. %v", tmpDir, err)
+	}
+
+	// delete the `rook-ceph-admin-rotator-keyring` secret (store.Delete)
+	s := keyring.GetSecretStore(clusterdCtx, clusterInfo, ownerInfo)
+	err := s.Delete(adminRotatorSecretPrefix)
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete admin rotator cephx keyring secret")
+	}
+
+	// during recovery, clusterInfo doesn't know the ceph version, so set it if needed
+	if clusterInfo.CephVersion.Major == 0 { // Major==0 should be good enough for detecting unknown ver
+		ver, err := cephclient.LeastUptodateDaemonVersion(clusterdCtx, clusterInfo, config.MonType)
+		if err != nil {
+			// this should be rare, and failure here doesn't need to block finalization of admin key
+			// rotation. status will be missing ceph version, but cluster can continue to reconcile
+			logger.Errorf("non-critical failure to determine missing ceph version of cluster in namespace %q for admin cephx key status", clusterInfo.Namespace)
+		} else {
+			clusterInfo.CephVersion = ver
+		}
+	}
+
+	// update `CephCluster.status.cephx.admin` with updated CephX status
+	// rotation is a given if this code line was reached
+	if err := updateCephClusterAdminCephxStatus(clusterdCtx, clusterInfo, true); err != nil {
+		return err
+	}
+
+	// restart Rook to force health checkers to restart with latest clusterInfo
+	// do this even during rotation recovery since the health checkers could still be running
+	logger.Infof("restarting rook operator after successful admin cephx key rotation for cluster in namespace %q", clusterInfo.Namespace)
+	reloadManagerFunc()
+
+	return errSuccessfulAdminKeyRotation
+}
+
+// helper routine that updates cephcluster 'admin' cephx status
+func updateCephClusterAdminCephxStatus(clusterdCtx *clusterd.Context, clusterInfo *cephclient.ClusterInfo, didRotate bool) error {
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		cluster := &cephv1.CephCluster{}
+		if err := clusterdCtx.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cluster); err != nil {
+			return errors.Wrap(err, "failed to get CephCluster to update the admin key cephx status")
+		}
+		cephx := keyring.UpdatedCephxStatus(didRotate, cluster.Spec.Security.CephX.Daemon, clusterInfo.CephVersion, cluster.Status.Cephx.Admin)
+		cluster.Status.Cephx.Admin = cephx
+		logger.Debugf("updating admin key cephx status to %+v", cephx)
+		if err := reporting.UpdateStatus(clusterdCtx.Client, cluster); err != nil {
+			return errors.Wrap(err, "failed to update admin key cephx status")
+		}
+
+		return nil
+	})
+	return err
+}
+
+func adminRotationTmpDir(clusterdCtx *clusterd.Context, clusterInfo *cephclient.ClusterInfo) string {
+	return filepath.Join(clusterdCtx.ConfigDir, clusterInfo.Namespace, "admin-rotate")
+}
+
+// make a minimal copy of the clusterInfo that tells called cephclient functions to use the given
+// username and keyring path instead the defaults
+func minimalCopyClusterInfo(clusterInfo *cephclient.ClusterInfo, username, keyringFilePath string) *cephclient.ClusterInfo {
+	c := &cephclient.ClusterInfo{
+		Context:   clusterInfo.Context,
+		Namespace: clusterInfo.Namespace,
+		CephCred: cephclient.CephCred{
+			Username: username,
+			// Secret not used by cephclient executor
+		},
+		CephVersion:         clusterInfo.CephVersion,
+		KeyringFileOverride: keyringFilePath,
+	}
+	c.SetName(clusterInfo.NamespacedName().Name)
+	return c
+}

pkg/operator/ceph/cluster/cluster.go

@@ -204,6 +204,9 @@ func (c *ClusterController) initializeCluster(cluster *cluster) error {
 		if err != nil {
 			if errors.Is(err, controller.ClusterInfoNoClusterNoSecret) {
 				logger.Info("clusterInfo not yet found, must be a new cluster.")
+				// ClusterInfoNoClusterNoSecret should return nil clusterInfo, but this is a
+				// mandatory condition for recoverPriorAdminCephxKeyRotation() below, so ensure it
+				clusterInfo = nil
 			} else {
 				return errors.Wrap(err, "failed to load cluster info")
 			}
@@ -211,7 +214,21 @@ func (c *ClusterController) initializeCluster(cluster *cluster) error {
 			clusterInfo.OwnerInfo = cluster.ownerInfo
 			clusterInfo.SetName(c.namespacedName.Name)
 			cluster.ClusterInfo = clusterInfo
+			if cluster.mons.ClusterInfo == nil {
+				// ClusterInfo stored in cluster.mons can be lost from the object stored in the
+				// clusterMap during admin key rotation corner cases. rehydrate it if needed
+				logger.Debugf("applying missing ClusterInfo to tracked mons info for cluster in namespace %q", cluster.Namespace)
+				cluster.mons.ClusterInfo = clusterInfo
+			}
 		}
+
+		// if necessary, recover from failed/interrupted admin key rotation
+		// this must be done before any ceph cli commands are run
+		err = recoverPriorAdminCephxKeyRotation(c.context, clusterInfo, cluster.ownerInfo, cluster.Namespace)
+		if err != nil {
+			return errors.Wrap(err, "failed to recover from prior admin cephx key rotation")
+		}
+
 		// If the local cluster has already been configured, immediately start monitoring the cluster.
 		// Test if the cluster has already been configured if the mgr deployment has been created.
 		// If the mgr does not exist, the mons have never been verified to be in quorum.
@@ -466,11 +483,18 @@ func (c *cluster) preMonStartupActions(cephVersion cephver.CephVersion) error {
 // It gets executed right after the main mon Start() method
 // Basically, it is executed between the monitors and the manager sequence
 func (c *cluster) postMonStartupActions() error {
+	// full cluster spec/status will be used to inform various cephx rotations
 	clusterObj := &cephv1.CephCluster{}
 	if err := c.context.Client.Get(c.ClusterInfo.Context, c.ClusterInfo.NamespacedName(), clusterObj); err != nil {
 		return errors.Wrapf(err, "failed to get cluster %v.", c.ClusterInfo.NamespacedName())
 	}
 
+	// rotate admin key first thing after mons are updated
+	err := rotateAdminCephxKey(c.context, c.ClusterInfo, c.ownerInfo, clusterObj) // TODO: rename?
+	if err != nil {
+		return errors.Wrapf(err, "failed to rotate admin cephx key")
+	}
+
 	// rotate mon cephx keys if required
 	didRotateMonCephxKeys, err := c.mons.RotateMonCephxKeys(clusterObj)
 	if err != nil {

pkg/operator/ceph/cluster/mon/health.go

@@ -128,13 +128,13 @@ func (hc *HealthChecker) Check(monitoringRoutines map[string]*controller.Cluster
 
 		select {
 		case <-monitoringRoutines[daemon].InternalCtx.Done():
-			logger.Infof("stopping monitoring of mons in namespace %q", hc.monCluster.Namespace)
+			logger.Infof("stopping monitoring of mons in namespace %q (internal context canceled)", hc.monCluster.Namespace)
 			delete(monitoringRoutines, daemon)
 			return
 
 		// Since c.ClusterInfo.IsInitialized() below uses a different context, we need to check if the context is done
 		case <-hc.monCluster.ClusterInfo.Context.Done():
-			logger.Infof("stopping monitoring of mons in namespace %q", hc.monCluster.Namespace)
+			logger.Infof("stopping monitoring of mons in namespace %q (mon cluster info context canceled)", hc.monCluster.Namespace)
 			delete(monitoringRoutines, daemon)
 			return
 

pkg/operator/ceph/config/keyring/store.go

@@ -20,6 +20,8 @@ limitations under the License.
 package keyring
 
 import (
+	"fmt"
+
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
@@ -149,3 +151,20 @@ func (k *SecretStore) CreateSecret(secret *v1.Secret) (string, error) {
 	}
 	return s.ResourceVersion, nil
 }
+
+// GetKeyringFromSecret returns the keyring present in a keyring secret
+func (k *SecretStore) GetKeyringFromSecret(resourceName string) (string, error) {
+	secretName := keyringSecretName(resourceName)
+	s, err := k.context.Clientset.CoreV1().Secrets(k.clusterInfo.Namespace).Get(k.clusterInfo.Context, secretName, metav1.GetOptions{})
+	if err != nil {
+		return "", err // do not wrap so that err can be compared to does not exist if needed
+	}
+	if keyring, ok := s.StringData[keyringFileName]; ok { // only available in unit tests
+		return keyring, nil
+	}
+	keyring, ok := s.Data[keyringFileName]
+	if !ok {
+		return "", fmt.Errorf("keyring secret %q does not contain keyring", secretName)
+	}
+	return string(keyring), nil
+}

pkg/operator/ceph/config/keyring/store_test.go

@@ -111,6 +111,11 @@ func TestKeyringStore(t *testing.T) {
 	assertKeyringData("test-resource-keyring", "qwertyuiop")
 	assertKeyringData("second-resource-keyring", "lkjhgfdsa")
 
+	// get key from secret
+	keyring, err := k.GetKeyringFromSecret("test-resource")
+	assert.NoError(t, err)
+	assert.Equal(t, "qwertyuiop", keyring)
+
 	// delete a key
 	err = k.Delete("test-resource")
 	assert.NoError(t, err)

pkg/operator/ceph/controller/cluster_info.go

@@ -422,6 +422,9 @@ func UpdateClusterAccessSecret(clientset kubernetes.Interface, clusterInfo *ceph
 		return errors.Wrapf(err, "failed to get secret %q in the namespace %q", AppName, clusterInfo.Namespace)
 	}
 
+	// update admin key secret
+	secret.Data[CephUserSecretKey] = []byte(clusterInfo.CephCred.Secret)
+
 	secret.Data[MonSecretNameKey] = []byte(clusterInfo.MonitorSecret)
 
 	logger.Infof("updating secret %q in the namespace %q", secret.Name, clusterInfo.Namespace)