Merge pull request #16281 from rook/mergify/bp/release-1.18/pr-16075

rbd mirror peer key rotation (backport #16075)
This commit is contained in:
mergify[bot] authored and GitHub committed 2025-08-11 17:39:04 +00:00
commit b5d64ea340
12 files changed
+735 -41

No files matched your search

+61 -1
View File
@@ -3555,6 +3555,19 @@ ConditionType
</tr>
<tr>
<td>
<code>cephx</code><br/>
<em>
<a href="#ceph.rook.io/v1.PeerTokenCephxStatus">
PeerTokenCephxStatus
</a>
</em>
</td>
<td>
<em>(Optional)</em>
</td>
</tr>
<tr>
<td>
<code>mirroringStatus</code><br/>
<em>
<a href="#ceph.rook.io/v1.MirroringStatusSpec">
@@ -4719,7 +4732,7 @@ are not rotated.</p>
<h3 id="ceph.rook.io/v1.CephxStatus">CephxStatus
</h3>
<p>
(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephClientStatus">CephClientStatus</a>, <a href="#ceph.rook.io/v1.CephxStatusWithKeyCount">CephxStatusWithKeyCount</a>, <a href="#ceph.rook.io/v1.ClusterCephxStatus">ClusterCephxStatus</a>, <a href="#ceph.rook.io/v1.LocalCephxStatus">LocalCephxStatus</a>)
(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephClientStatus">CephClientStatus</a>, <a href="#ceph.rook.io/v1.CephxStatusWithKeyCount">CephxStatusWithKeyCount</a>, <a href="#ceph.rook.io/v1.ClusterCephxStatus">ClusterCephxStatus</a>, <a href="#ceph.rook.io/v1.LocalCephxStatus">LocalCephxStatus</a>, <a href="#ceph.rook.io/v1.PeerTokenCephxStatus">PeerTokenCephxStatus</a>)
</p>
<div>
</div>
@@ -5040,6 +5053,22 @@ Ceph cluster. Daemon CephX keys can be rotated without affecting client connecti
</tr>
<tr>
<td>
<code>rbdMirrorPeer</code><br/>
<em>
<a href="#ceph.rook.io/v1.CephxConfig">
CephxConfig
</a>
</em>
</td>
<td>
<p>RBDMirrorPeer configures CephX key settings of the <code>rbd-mirror-peer</code> user that is used for creating
bootstrap peer token used connect peer clusters. Rotating the <code>rbd-mirror-peer</code> user key will update
the mirror peer token.
Rotation will affect any existing peers connected to this cluster, so take care when exercising this option.</p>
</td>
</tr>
<tr>
<td>
<code>csi</code><br/>
<em>
<a href="#ceph.rook.io/v1.CephXConfigWithPriorCount">
@@ -11979,6 +12008,37 @@ int
</tr>
</tbody>
</table>
<h3 id="ceph.rook.io/v1.PeerTokenCephxStatus">PeerTokenCephxStatus
</h3>
<p>
(<em>Appears on:</em><a href="#ceph.rook.io/v1.CephBlockPoolStatus">CephBlockPoolStatus</a>)
</p>
<div>
<p>PeerTokenCephxStatus represents the cephx key rotation status for peer tokens</p>
</div>
<table>
<thead>
<tr>
<th>Field</th>
<th>Description</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<code>peerToken</code><br/>
<em>
<a href="#ceph.rook.io/v1.CephxStatus">
CephxStatus
</a>
</em>
</td>
<td>
<p>PeerToken shows the rotation status of the peer token associated with the <code>rbd-mirror-peer</code> user.</p>
</td>
</tr>
</tbody>
</table>
<h3 id="ceph.rook.io/v1.PeersSpec">PeersSpec
</h3>
<p>
@@ -615,6 +615,32 @@ spec:
status:
description: CephBlockPoolStatus represents the mirroring status of Ceph Storage Pool
properties:
cephx:
description: PeerTokenCephxStatus represents the cephx key rotation status for peer tokens
properties:
peerToken:
description: PeerToken shows the rotation status of the peer token associated with the `rbd-mirror-peer` user.
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.
@@ -3783,6 +3809,38 @@ spec:
- KeyGeneration
type: string
type: object
rbdMirrorPeer:
description: |-
RBDMirrorPeer configures CephX key settings of the `rbd-mirror-peer` user that is used for creating
bootstrap peer token used connect peer clusters. Rotating the `rbd-mirror-peer` user key will update
the mirror peer token.
Rotation will affect any existing peers connected to this cluster, so take care when exercising this option.
properties:
keyGeneration:
description: |-
KeyGeneration specifies the desired CephX key generation. This is used when KeyRotationPolicy
is KeyGeneration and ignored for other policies. If this is set to greater than the current
key generation, relevant keys will be rotated, and the generation value will be updated to
this new value (generation values are not necessarily incremental, though that is the
intended use case). If this is set to less than or equal to the current key generation, keys
are not rotated.
format: int32
maximum: 4294967295
minimum: 0
type: integer
x-kubernetes-validations:
- message: keyGeneration cannot be decreased
rule: self >= oldSelf
keyRotationPolicy:
description: |-
KeyRotationPolicy controls if and when CephX keys are rotated after initial creation.
One of Disabled, or KeyGeneration. Default Disabled.
enum:
- ""
- Disabled
- KeyGeneration
type: string
type: object
type: object
keyRotation:
description: KeyRotation defines options for rotation of OSD disk encryption keys.
+58
View File
@@ -617,6 +617,32 @@ spec:
status:
description: CephBlockPoolStatus represents the mirroring status of Ceph Storage Pool
properties:
cephx:
description: PeerTokenCephxStatus represents the cephx key rotation status for peer tokens
properties:
peerToken:
description: PeerToken shows the rotation status of the peer token associated with the `rbd-mirror-peer` user.
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.
@@ -3781,6 +3807,38 @@ spec:
- KeyGeneration
type: string
type: object
rbdMirrorPeer:
description: |-
RBDMirrorPeer configures CephX key settings of the `rbd-mirror-peer` user that is used for creating
bootstrap peer token used connect peer clusters. Rotating the `rbd-mirror-peer` user key will update
the mirror peer token.
Rotation will affect any existing peers connected to this cluster, so take care when exercising this option.
properties:
keyGeneration:
description: |-
KeyGeneration specifies the desired CephX key generation. This is used when KeyRotationPolicy
is KeyGeneration and ignored for other policies. If this is set to greater than the current
key generation, relevant keys will be rotated, and the generation value will be updated to
this new value (generation values are not necessarily incremental, though that is the
intended use case). If this is set to less than or equal to the current key generation, keys
are not rotated.
format: int32
maximum: 4294967295
minimum: 0
type: integer
x-kubernetes-validations:
- message: keyGeneration cannot be decreased
rule: self >= oldSelf
keyRotationPolicy:
description: |-
KeyRotationPolicy controls if and when CephX keys are rotated after initial creation.
One of Disabled, or KeyGeneration. Default Disabled.
enum:
- ""
- Disabled
- KeyGeneration
type: string
type: object
type: object
keyRotation:
description: KeyRotation defines options for rotation of OSD disk encryption keys.
+14
View File
@@ -330,6 +330,12 @@ type ClusterCephxConfig struct {
// Ceph cluster. Daemon CephX keys can be rotated without affecting client connections.
Daemon CephxConfig `json:"daemon,omitempty"`
// RBDMirrorPeer configures CephX key settings of the `rbd-mirror-peer` user that is used for creating
// bootstrap peer token used connect peer clusters. Rotating the `rbd-mirror-peer` user key will update
// the mirror peer token.
// Rotation will affect any existing peers connected to this cluster, so take care when exercising this option.
RBDMirrorPeer CephxConfig `json:"rbdMirrorPeer,omitempty"`
// CSI configures CephX key rotation settings for the Ceph-CSI daemons in the current Kubernetes cluster.
// CSI key rotation can affect existing PV connections, so take care when exercising this option.
CSI CephXConfigWithPriorCount `json:"csi,omitempty"`
@@ -1015,6 +1021,8 @@ type CephBlockPoolStatus struct {
// +optional
Phase ConditionType `json:"phase,omitempty"`
// +optional
Cephx PeerTokenCephxStatus `json:"cephx,omitempty"`
// +optional
MirroringStatus *MirroringStatusSpec `json:"mirroringStatus,omitempty"`
// +optional
MirroringInfo *MirroringInfoSpec `json:"mirroringInfo,omitempty"`
@@ -1031,6 +1039,12 @@ type CephBlockPoolStatus struct {
Conditions []Condition `json:"conditions,omitempty"`
}
// PeerTokenCephxStatus represents the cephx key rotation status for peer tokens
type PeerTokenCephxStatus struct {
// PeerToken shows the rotation status of the peer token associated with the `rbd-mirror-peer` user.
PeerToken CephxStatus `json:"peerToken,omitempty"`
}
// MirroringStatusSpec is the status of the pool/radosNamespace mirroring
type MirroringStatusSpec struct {
// MirroringStatus is the mirroring status of a pool/radosNamespace
@@ -525,6 +525,7 @@ func (in *CephBlockPoolRadosNamespaceStatus) DeepCopy() *CephBlockPoolRadosNames
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *CephBlockPoolStatus) DeepCopyInto(out *CephBlockPoolStatus) {
*out = *in
out.Cephx = in.Cephx
if in.MirroringStatus != nil {
in, out := &in.MirroringStatus, &out.MirroringStatus
*out = new(MirroringStatusSpec)
@@ -2058,6 +2059,7 @@ func (in *ClientSpec) DeepCopy() *ClientSpec {
func (in *ClusterCephxConfig) DeepCopyInto(out *ClusterCephxConfig) {
*out = *in
out.Daemon = in.Daemon
out.RBDMirrorPeer = in.RBDMirrorPeer
out.CSI = in.CSI
return
}
@@ -4400,6 +4402,23 @@ func (in *PeerStatSpec) DeepCopy() *PeerStatSpec {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PeerTokenCephxStatus) DeepCopyInto(out *PeerTokenCephxStatus) {
*out = *in
out.PeerToken = in.PeerToken
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new PeerTokenCephxStatus.
func (in *PeerTokenCephxStatus) DeepCopy() *PeerTokenCephxStatus {
if in == nil {
return nil
}
out := new(PeerTokenCephxStatus)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *PeersSpec) DeepCopyInto(out *PeersSpec) {
*out = *in
+10 -1
View File
@@ -476,7 +476,7 @@ So the scenario looks like:
6. Repeat the steps flipping source and destination to enable
bi-directional mirroring
*/
func CreateRBDMirrorBootstrapPeerWithoutPool(context *clusterd.Context, clusterInfo *ClusterInfo) ([]byte, error) {
func CreateRBDMirrorBootstrapPeerWithoutPool(context *clusterd.Context, clusterInfo *ClusterInfo, rotateMirrorPeerKey bool) ([]byte, error) {
fullClientName := getQualifiedUser(rbdMirrorPeerKeyringID)
logger.Infof("create rbd-mirror bootstrap peer token %q", fullClientName)
key, err := AuthGetOrCreateKey(context, clusterInfo, fullClientName, rbdMirrorPeerCaps)
@@ -485,6 +485,15 @@ func CreateRBDMirrorBootstrapPeerWithoutPool(context *clusterd.Context, clusterI
}
logger.Infof("successfully created rbd-mirror bootstrap peer token for cluster %q", clusterInfo.NamespacedName().Name)
if rotateMirrorPeerKey {
logger.Infof("rotating cephx key for rbd-mirror peer key %q in namespace %q", fullClientName, clusterInfo.Namespace)
newKey, err := AuthRotate(context, clusterInfo, fullClientName)
if err != nil {
return nil, errors.Wrapf(err, "failed to rotate rbd-mirror peer key for %q", fullClientName)
}
key = newKey
}
mons := sets.New[string]()
for _, mon := range clusterInfo.AllMonitors() {
mons.Insert(mon.Endpoint)
+140 -1
View File
@@ -25,6 +25,7 @@ import (
"time"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
"github.com/rook/rook/pkg/clusterd"
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test"
@@ -38,7 +39,10 @@ import (
apps "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/kubernetes"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
)
func TestCheckHealth(t *testing.T) {
@@ -55,13 +59,19 @@ func TestCheckHealth(t *testing.T) {
return clienttest.MonInQuorumResponse(), nil
},
}
objects := []runtime.Object{
getCephCluster("default", "default"),
}
client := getClient(objects...)
clientset := test.New(t, 1)
configDir := t.TempDir()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
// clusterInfo is nil so we return err
@@ -90,6 +100,7 @@ func TestCheckHealth(t *testing.T) {
}
c.ClusterInfo.Context = ctx
err = c.checkHealth(ctx)
assert.Nil(t, err)
logger.Infof("mons after checkHealth: %v", c.ClusterInfo.InternalMonitors)
@@ -288,7 +299,28 @@ func TestEvictMonOnSameNode(t *testing.T) {
return "{\"key\":\"mysecurekey\"}", nil
},
}
context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: executor}
fakeCluster := &cephv1.CephCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "default",
Namespace: "default",
},
Spec: cephv1.ClusterSpec{
Security: cephv1.ClusterSecuritySpec{
CephX: cephv1.ClusterCephxConfig{
RBDMirrorPeer: cephv1.CephxConfig{},
},
},
},
Status: cephv1.ClusterStatus{
Cephx: &cephv1.ClusterCephxStatus{
RBDMirrorPeer: &cephv1.CephxStatus{},
},
},
}
client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects(fakeCluster).Build()
context := &clusterd.Context{Clientset: clientset, Client: client, ConfigDir: configDir, Executor: executor}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0}, "myversion")
@@ -415,13 +447,25 @@ func TestCheckHealthNotFound(t *testing.T) {
return clienttest.MonInQuorumResponse(), nil
},
}
objects := []runtime.Object{
getCephCluster("default", "default"),
}
client := getClient(objects...)
clientset := test.New(t, 1)
configDir := t.TempDir()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
// mock out the scheduler to return node0
waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) {
node, _ := clientset.CoreV1().Nodes().Get(ctx, "node0", metav1.GetOptions{})
return SchedulingResult{Node: node}, nil
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion")
@@ -476,12 +520,24 @@ func TestAddRemoveMons(t *testing.T) {
return monQuorumResponse, nil
},
}
objects := []runtime.Object{
getCephCluster("default", "default"),
}
client := getClient(objects...)
clientset := test.New(t, 1)
configDir := t.TempDir()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
// mock out the scheduler to return node0
waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) {
node, _ := clientset.CoreV1().Nodes().Get(ctx, "node0", metav1.GetOptions{})
return SchedulingResult{Node: node}, nil
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
@@ -795,13 +851,42 @@ func TestExternalMons_notInSpec_InQuorum(t *testing.T) {
return monQuorumResponse, nil
},
}
fakeCluster := &cephv1.CephCluster{
ObjectMeta: metav1.ObjectMeta{
Name: "default",
Namespace: "default",
},
Spec: cephv1.ClusterSpec{
Security: cephv1.ClusterSecuritySpec{
CephX: cephv1.ClusterCephxConfig{
RBDMirrorPeer: cephv1.CephxConfig{},
},
},
},
Status: cephv1.ClusterStatus{
Cephx: &cephv1.ClusterCephxStatus{
RBDMirrorPeer: &cephv1.CephxStatus{},
},
},
}
clientset := test.New(t, 1)
configDir := t.TempDir()
client := fake.NewClientBuilder().WithScheme(scheme.Scheme).WithRuntimeObjects(fakeCluster).Build()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
// mock out the scheduler to return node0
waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) {
node, _ := clientset.CoreV1().Nodes().Get(ctx, "node0", metav1.GetOptions{})
return SchedulingResult{Node: node}, nil
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
@@ -926,13 +1011,26 @@ func TestExternalMons_inSpec_notInQuorum(t *testing.T) {
return monQuorumResponse, nil
},
}
objects := []runtime.Object{
getCephCluster("default", "default"),
}
client := getClient(objects...)
clientset := test.New(t, 1)
configDir := t.TempDir()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
// mock out the scheduler to return node0
waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) {
node, _ := clientset.CoreV1().Nodes().Get(ctx, "node0", metav1.GetOptions{})
return SchedulingResult{Node: node}, nil
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
@@ -1049,13 +1147,25 @@ func TestExternalMons_inSpec_inQuorum(t *testing.T) {
return monQuorumResponse, nil
},
}
objects := []runtime.Object{
getCephCluster("default", "default"),
}
client := getClient(objects...)
clientset := test.New(t, 1)
configDir := t.TempDir()
context := &clusterd.Context{
Clientset: clientset,
ConfigDir: configDir,
Executor: executor,
Client: client,
}
// mock out the scheduler to return node0
waitForMonitorScheduling = func(c *Cluster, d *apps.Deployment) (SchedulingResult, error) {
node, _ := clientset.CoreV1().Nodes().Get(ctx, "node0", metav1.GetOptions{})
return SchedulingResult{Node: node}, nil
}
ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef()
c := New(ctx, context, "ns", cephv1.ClusterSpec{}, ownerInfo)
setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion")
@@ -1186,3 +1296,32 @@ func TestExternalMons_inSpec_inQuorum(t *testing.T) {
}
}
}
func getClient(obj ...runtime.Object) client.Client {
// Register operator types with the runtime scheme.
scheme := scheme.Scheme
scheme.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephCluster{})
client := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(obj...).Build()
return client
}
func getCephCluster(name, namespace string) *cephv1.CephCluster {
return &cephv1.CephCluster{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
},
Spec: cephv1.ClusterSpec{
Security: cephv1.ClusterSecuritySpec{
CephX: cephv1.ClusterCephxConfig{
RBDMirrorPeer: cephv1.CephxConfig{},
},
},
},
Status: cephv1.ClusterStatus{
Cephx: &cephv1.ClusterCephxStatus{
RBDMirrorPeer: &cephv1.CephxStatus{},
},
},
}
}
+57 -1
View File
@@ -26,10 +26,13 @@ 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"
"github.com/rook/rook/pkg/operator/ceph/reporting"
"github.com/rook/rook/pkg/operator/k8sutil"
v1 "k8s.io/api/core/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/util/retry"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -71,12 +74,25 @@ func CreateBootstrapPeerSecret(ctx *clusterd.Context, clusterInfo *cephclient.Cl
case *cephv1.CephCluster:
ns = objectType.Namespace
daemonType = "cluster-rbd"
// check if rbd-mirror-peer user needs to be rotated
shouldRotateKeys, err := shouldRotateMirrorPeerKeys(ctx, clusterInfo)
if err != nil {
return ImmediateRetryResult, errors.Wrap(err, "failed to check if rbd-mirror-peer keys should be rotated or not")
}
// Create rbd mirror bootstrap peer token
bootstrapToken, err = cephclient.CreateRBDMirrorBootstrapPeerWithoutPool(ctx, clusterInfo)
bootstrapToken, err = cephclient.CreateRBDMirrorBootstrapPeerWithoutPool(ctx, clusterInfo, shouldRotateKeys)
if err != nil {
return ImmediateRetryResult, errors.Wrapf(err, "failed to create %s-mirror bootstrap peer", daemonType)
}
// update rbd mirror peer cephx status in cephCluster resource
err = updateCephClusterCephxRbdMirrorStatus(ctx, clusterInfo, shouldRotateKeys)
if err != nil {
return ImmediateRetryResult, errors.Wrapf(err, "failed to update rbd-mirror-peer cephx status in the ceph cluster")
}
// Add additional information to the peer token
bootstrapToken, err = expandBootstrapPeerToken(ctx, clusterInfo, bootstrapToken)
if err != nil {
@@ -222,3 +238,43 @@ func expandBootstrapPeerToken(ctx *clusterd.Context, clusterInfo *cephclient.Clu
// Return the base64 encoded token
return []byte(base64.StdEncoding.EncodeToString(decodedTokenBackToJSON)), nil
}
func shouldRotateMirrorPeerKeys(c *clusterd.Context, clusterInfo *cephclient.ClusterInfo) (bool, error) {
cephObj := &cephv1.CephCluster{}
if err := c.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephObj); err != nil {
return false, errors.Wrapf(err, "failed to get cluster %v to get the cephx keys to rotate.", clusterInfo.NamespacedName())
}
// TODO: for rotation WithCephVersionUpdate fix this to have the right runningCephVersion and desiredCephVersion
desiredCephVersion := clusterInfo.CephVersion
runningCephVersion := clusterInfo.CephVersion
shouldRotateKeys, err := keyring.ShouldRotateCephxKeys(cephObj.Spec.Security.CephX.RBDMirrorPeer, runningCephVersion, desiredCephVersion, *cephObj.Status.Cephx.RBDMirrorPeer)
if err != nil {
return false, errors.Wrap(err, "failed to check if mirror peer keys should be rotated or not")
}
return shouldRotateKeys, nil
}
// updateCephClusterCephxRbdMirrorStatus fetches the latest cephCluster instance and updates mirror peer cephx status
func updateCephClusterCephxRbdMirrorStatus(c *clusterd.Context, clusterInfo *cephclient.ClusterInfo, didRotate bool) error {
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
cluster := &cephv1.CephCluster{}
if err := c.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cluster); err != nil {
return errors.Wrapf(err, "failed to get cluster %v to update the conditions.", clusterInfo.NamespacedName())
}
updatedStatus := keyring.UpdatedCephxStatus(didRotate, cluster.Spec.Security.CephX.RBDMirrorPeer, clusterInfo.CephVersion, *cluster.Status.Cephx.RBDMirrorPeer)
cluster.Status.Cephx.RBDMirrorPeer = &updatedStatus
logger.Debugf("updating rbd-mirror cephx status to %+v", cluster.Status.Cephx.RBDMirrorPeer)
if err := reporting.UpdateStatus(c.Client, cluster); err != nil {
return errors.Wrap(err, "failed to update cluster cephx status for rbd-mirror")
}
return nil
})
if err != nil {
return err
}
logger.Debugf("successfully updated rbd-mirror cephx status on the cluster %q", clusterInfo.NamespacedName())
return nil
}
+33
View File
@@ -488,3 +488,36 @@ func GetSpec(obj client.Object) interface{} {
return spec.Interface()
}
func WatchPeerTokenSecretPredicate[T *corev1.Secret]() predicate.TypedFuncs[T] {
return predicate.TypedFuncs[T]{
CreateFunc: func(e event.TypedCreateEvent[T]) bool {
newSecret := (*corev1.Secret)(e.Object)
// reconcile when secret is created
if strings.Contains(newSecret.GetName(), clusterMirrorBootstrapPeerSecretName) {
logger.Debugf("peer token create event for secret %q in the namespace %q", newSecret.GetName(), newSecret.GetNamespace())
return true
}
return false
},
UpdateFunc: func(e event.TypedUpdateEvent[T]) bool {
newSecret := (*corev1.Secret)(e.ObjectNew)
oldSecret := (*corev1.Secret)(e.ObjectOld)
if !strings.Contains(newSecret.GetName(), clusterMirrorBootstrapPeerSecretName) {
return false
}
// reconcile if the peer token data has changed
newData := newSecret.Data["token"]
oldData := oldSecret.Data["token"]
if string(newData) != string(oldData) {
logger.Debugf("peer token update event for secret %q in the namespace %q", newSecret.GetName(), newSecret.GetNamespace())
return true
}
return false
},
DeleteFunc: func(e event.TypedDeleteEvent[T]) bool {
// Do not reconcile when secret is deleted
return false
},
}
}
+41 -9
View File
@@ -129,7 +129,7 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco
// Build Handler function to return the list of ceph block pool
// This is used by the watchers below
handlerFunc, err := opcontroller.ObjectToCRMapper[*cephv1.CephBlockPoolList, *corev1.ConfigMap](
configHandler, err := opcontroller.ObjectToCRMapper[*cephv1.CephBlockPoolList, *corev1.ConfigMap](
opManagerContext,
mgr.GetClient(),
&cephv1.CephBlockPoolList{},
@@ -144,7 +144,7 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco
source.Kind(
mgr.GetCache(),
&corev1.ConfigMap{TypeMeta: metav1.TypeMeta{Kind: "ConfigMap", APIVersion: corev1.SchemeGroupVersion.String()}},
handler.TypedEnqueueRequestsFromMapFunc(handlerFunc),
handler.TypedEnqueueRequestsFromMapFunc(configHandler),
mon.PredicateMonEndpointChanges(),
),
)
@@ -152,6 +152,30 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco
return err
}
// Build Handler function to return the list of ceph block pool
// This is used by the watchers below
secretHandler, err := opcontroller.ObjectToCRMapper[*cephv1.CephBlockPoolList, *corev1.Secret](
opManagerContext,
mgr.GetClient(),
&cephv1.CephBlockPoolList{},
mgr.GetScheme(),
)
if err != nil {
return err
}
// Watch for updates to the secret triggered by changes in the peer pool token
err = c.Watch(
source.Kind(
mgr.GetCache(),
&corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: corev1.SchemeGroupVersion.String()}},
handler.TypedEnqueueRequestsFromMapFunc(secretHandler),
opcontroller.WatchPeerTokenSecretPredicate(),
),
)
if err != nil {
return err
}
return nil
}
@@ -203,9 +227,10 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
var statusErr error
// The CR was just created, initializing status fields
if cephBlockPool.Status == nil {
err = r.updateStatus(request.NamespacedName, cephv1.ConditionProgressing, k8sutil.ObservedGenerationNotAvailable)
// The pool is not available so let's not build the status Info yet
err = r.updateStatus(request.NamespacedName, cephv1.ConditionProgressing, k8sutil.ObservedGenerationNotAvailable, &cephv1.CephxStatus{})
if err != nil {
logger.Errorf("failed to update %q status to %q: %v", request.NamespacedName, cephv1.ConditionProgressing, err)
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(err, "failed to update %q status to %q", request.NamespacedName, cephv1.ConditionProgressing)
}
}
@@ -313,7 +338,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephBlockPool, nil
}
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionFailure, k8sutil.ObservedGenerationNotAvailable)
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionFailure, k8sutil.ObservedGenerationNotAvailable, nil)
if statusErr != nil {
logger.Errorf("failed to update %q status to %q: %v", request.NamespacedName, cephv1.ConditionFailure, statusErr)
}
@@ -331,13 +356,19 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// Always create a bootstrap peer token in case another cluster wants to add us as a peer
reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, clusterInfo, cephBlockPool, k8sutil.NewOwnerInfo(cephBlockPool, r.scheme))
if err != nil {
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionFailure, k8sutil.ObservedGenerationNotAvailable)
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionFailure, k8sutil.ObservedGenerationNotAvailable, nil)
if statusErr != nil {
logger.Errorf("failed to update %q status to %q: %v", request.NamespacedName, cephv1.ConditionFailure, statusErr)
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(statusErr, "failed to update %q status to %q", request.NamespacedName, cephv1.ConditionFailure)
}
return reconcileResponse, *cephBlockPool, errors.Wrapf(err, "failed to create rbd-mirror bootstrap peer for pool %q.", cephBlockPool.GetName())
}
// update rbdMirror cephXStatus immediately after bootstrapping the peer token
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionProgressing, observedGeneration, cephCluster.Status.Cephx.RBDMirrorPeer)
if statusErr != nil {
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(statusErr, "failed to update %q status to %q", request.NamespacedName, cephv1.ConditionProgressing)
}
// Check if rbd-mirror CR and daemons are running
logger.Debug("listing rbd-mirror CR")
@@ -356,7 +387,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// update ObservedGeneration in status at the end of reconcile
// Set Ready status, we are done reconciling
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionReady, observedGeneration)
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionReady, observedGeneration, nil)
if cephBlockPool.Spec.StatusCheck.Mirror.Disabled {
// Stop monitoring the mirroring status of this pool
@@ -389,7 +420,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
}
// update ObservedGeneration in status at the end of reconcile
// Set Ready status, we are done reconciling
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionReady, observedGeneration)
statusErr = r.updateStatus(request.NamespacedName, cephv1.ConditionReady, observedGeneration, &cephv1.CephxStatus{})
// Stop monitoring the mirroring status of this pool
if blockPoolContextsExists && r.blockPoolContexts[blockPoolChannelKey].started {
@@ -581,6 +612,7 @@ func configureRBDStats(clusterContext *clusterd.Context, clusterInfo *cephclient
cephBlockPoolList := &cephv1.CephBlockPoolList{}
var rookStatsPools []string
var removePools []string
err := clusterContext.Client.List(clusterInfo.Context, cephBlockPoolList, namespaceListOpt)
if err != nil {
return errors.Wrap(err, "failed to retrieve list of CephBlockPool")
+203
View File
@@ -317,6 +317,7 @@ func TestCephBlockPoolController(t *testing.T) {
CephStatus: &cephv1.CephStatus{
Health: "",
},
Cephx: &cephv1.ClusterCephxStatus{},
},
}
@@ -974,3 +975,205 @@ func TestGenerateStatsPoolList(t *testing.T) {
})
}
}
func TestMirrorPeerKeyRotationStatus(t *testing.T) {
ctx := context.TODO()
// Set DEBUG logging
t.Setenv("ROOK_LOG_LEVEL", "DEBUG")
var (
name = "my-pool"
namespace = "rook-ceph"
replicas uint = 3
)
cephCluster := &cephv1.CephCluster{
ObjectMeta: metav1.ObjectMeta{
Name: namespace,
Namespace: namespace,
},
Status: cephv1.ClusterStatus{
Phase: "",
CephVersion: &cephv1.ClusterVersion{
Version: "20.2.0-0",
},
CephStatus: &cephv1.CephStatus{
Health: "HEALTH_OK",
},
Cephx: &cephv1.ClusterCephxStatus{},
},
}
// A Pool resource with metadata and spec.
pool := &cephv1.CephBlockPool{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
UID: types.UID("c47cac40-9bee-4d52-823b-ccd803ba5bfe"),
Finalizers: []string{"cephblockpool.ceph.rook.io"},
},
TypeMeta: metav1.TypeMeta{
Kind: "CephBlockPool",
},
Spec: cephv1.NamedBlockPoolSpec{
PoolSpec: cephv1.PoolSpec{
Replicated: cephv1.ReplicatedSpec{
Size: replicas,
},
Mirroring: cephv1.MirroringSpec{
Enabled: true,
Mode: "image",
Peers: &cephv1.MirroringPeerSpec{},
},
StatusCheck: cephv1.MirrorHealthCheckSpec{
Mirror: cephv1.HealthCheckSpec{
Disabled: true,
},
},
},
},
Status: &cephv1.CephBlockPoolStatus{
Phase: "",
},
}
// Objects to track in the fake client.
object := []runtime.Object{
pool, cephCluster,
}
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_OK"},"pgmap":{"num_pgs":100,"pgs_by_state":[{"state_name":"active+clean","count":100}]}}`, nil
}
if args[0] == "mirror" && args[1] == "pool" && args[2] == "peer" && args[3] == "bootstrap" && args[4] == "create" {
return `eyJmc2lkIjoiYzZiMDg3ZjItNzgyOS00ZGJiLWJjZmMtNTNkYzM0ZTBiMzVkIiwiY2xpZW50X2lkIjoicmJkLW1pcnJvci1wZWVyIiwia2V5IjoiQVFBV1lsWmZVQ1Q2RGhBQVBtVnAwbGtubDA5YVZWS3lyRVV1NEE9PSIsIm1vbl9ob3N0IjoiW3YyOjE5Mi4xNjguMTExLjEwOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTA6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjEyOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTI6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjExOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTE6Njc4OV0ifQ==`, nil
}
if args[0] == "mirror" && args[1] == "pool" && args[2] == "info" {
return "{}", nil
}
return "", nil
},
}
// Register operator types with the runtime scheme.
s := scheme.Scheme
s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephBlockPool{})
s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephCluster{}, &cephv1.CephClusterList{})
s.AddKnownTypes(cephv1.SchemeGroupVersion, &cephv1.CephBlockPoolList{})
// Create a fake client to mock API calls.
cl := fake.NewClientBuilder().WithRuntimeObjects(object...).Build()
c := &clusterd.Context{
Client: cl,
Executor: executor,
Clientset: testop.New(t, 1),
RookClientset: rookclient.NewSimpleClientset(),
}
// 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)
// Mock request to simulate Reconcile() being called on an event for a
// watched resource .
req := reconcile.Request{
NamespacedName: types.NamespacedName{
Name: name,
Namespace: namespace,
},
}
// Create a ReconcileCephBlockPool object with the scheme and fake client.
r := &ReconcileCephBlockPool{
client: cl,
scheme: s,
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
t.Setenv("POD_NAME", "test")
t.Setenv("POD_NAMESPACE", namespace)
p := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "test",
Namespace: namespace,
OwnerReferences: []metav1.OwnerReference{
{
Kind: "ReplicaSet",
Name: "testReplicaSet",
},
},
},
}
// Create fake pod
_, err = r.context.Clientset.CoreV1().Pods(namespace).Create(context.TODO(), p, metav1.CreateOptions{})
assert.NoError(t, err)
replicaSet := &appsv1.ReplicaSet{
ObjectMeta: metav1.ObjectMeta{
Name: "testReplicaSet",
Namespace: namespace,
OwnerReferences: []metav1.OwnerReference{
{
Kind: "Deployment",
},
},
},
}
// Create fake replicaset
_, err = r.context.Clientset.AppsV1().ReplicaSets(namespace).Create(context.TODO(), replicaSet, metav1.CreateOptions{})
assert.NoError(t, err)
t.Run("first reconcile - PeerToken cephx status is set to empty", func(t *testing.T) {
_, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
pool := &cephv1.CephBlockPool{}
err = r.client.Get(context.TODO(), req.NamespacedName, pool)
assert.NoError(t, err)
assert.Equal(t, cephv1.CephxStatus{}, pool.Status.Cephx.PeerToken)
})
t.Run("subsequent reconcile - retain PeerToken cephx status", func(t *testing.T) {
_, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
pool := &cephv1.CephBlockPool{}
err = r.client.Get(context.TODO(), req.NamespacedName, pool)
assert.NoError(t, err)
assert.Equal(t, cephv1.CephxStatus{}, pool.Status.Cephx.PeerToken)
})
t.Run("PeerToken cephx status should be updated based on the cephCluster cephx status", func(t *testing.T) {
RDBMirrorPeerStatus := &cephv1.CephxStatus{
KeyGeneration: 2,
KeyCephVersion: "20.2.0-0",
}
cephCluster.Status.Cephx.RBDMirrorPeer = RDBMirrorPeerStatus
err := r.client.Update(context.TODO(), cephCluster)
assert.NoError(t, err)
_, err = r.Reconcile(ctx, req)
assert.NoError(t, err)
pool := &cephv1.CephBlockPool{}
err = r.client.Get(context.TODO(), req.NamespacedName, pool)
assert.NoError(t, err)
assert.Equal(t, *RDBMirrorPeerStatus, pool.Status.Cephx.PeerToken)
})
}
+41 -28
View File
@@ -26,40 +26,53 @@ import (
"github.com/rook/rook/pkg/operator/k8sutil"
kerrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/util/retry"
)
// updateStatus updates a pool CR with the given status
func (r *ReconcileCephBlockPool) updateStatus(poolName types.NamespacedName, status cephv1.ConditionType, observedGeneration int64) error {
pool := &cephv1.CephBlockPool{}
err := r.client.Get(r.opManagerContext, poolName, pool)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephBlockPool resource not found. Ignoring since object must be deleted.")
return nil
func (r *ReconcileCephBlockPool) updateStatus(poolName types.NamespacedName, status cephv1.ConditionType, observedGeneration int64, cephx *cephv1.CephxStatus) error {
err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
pool := &cephv1.CephBlockPool{}
err := r.client.Get(r.opManagerContext, poolName, pool)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephBlockPool resource not found. Ignoring since object must be deleted.")
return nil
}
logger.Warningf("failed to retrieve pool %q to update status to %q. %v", poolName, status, err)
return errors.Wrapf(err, "failed to retrieve pool %q to update status to %q", poolName, status)
}
logger.Warningf("failed to retrieve pool %q to update status to %q. %v", poolName, status, err)
return errors.Wrapf(err, "failed to retrieve pool %q to update status to %q", poolName, status)
if pool.Status == nil {
pool.Status = &cephv1.CephBlockPoolStatus{}
}
// add pool ID to the status
if status == cephv1.ConditionReady && pool.Status.PoolID == 0 {
r.updatePoolID(pool)
}
pool.Status.Phase = status
updateStatusInfo(pool)
if observedGeneration != k8sutil.ObservedGenerationNotAvailable {
pool.Status.ObservedGeneration = observedGeneration
}
if cephx != nil {
pool.Status.Cephx.PeerToken = *cephx
}
if err := reporting.UpdateStatus(r.client, pool); err != nil {
logger.Warningf("failed to set pool %q status to %q. %v", pool.Name, status, err)
return errors.Wrapf(err, "failed to set pool %q status to %q", pool.Name, status)
}
logger.Debugf("pool %q status updated to %q", poolName, status)
return nil
})
if err != nil {
logger.Error(err)
}
if pool.Status == nil {
pool.Status = &cephv1.CephBlockPoolStatus{}
}
// add pool ID to the status
if status == cephv1.ConditionReady && pool.Status.PoolID == 0 {
r.updatePoolID(pool)
}
pool.Status.Phase = status
updateStatusInfo(pool)
if observedGeneration != k8sutil.ObservedGenerationNotAvailable {
pool.Status.ObservedGeneration = observedGeneration
}
if err := reporting.UpdateStatus(r.client, pool); err != nil {
logger.Warningf("failed to set pool %q status to %q. %v", pool.Name, status, err)
return errors.Wrapf(err, "failed to set pool %q status to %q", pool.Name, status)
}
logger.Debugf("pool %q status updated to %q", poolName, status)
return nil
}