cephxmirroringStatus-(Appears on:CephClientStatus, CephxStatusWithKeyCount, ClusterCephxStatus, LocalCephxStatus) +(Appears on:CephClientStatus, CephxStatusWithKeyCount, ClusterCephxStatus, LocalCephxStatus, PeerTokenCephxStatus)
rbdMirrorPeerRBDMirrorPeer 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.
csi+(Appears on:CephBlockPoolStatus) +
+PeerTokenCephxStatus represents the cephx key rotation status for peer tokens
+| Field | +Description | +
|---|---|
+peerToken+ + +CephxStatus + + + |
+
+ PeerToken shows the rotation status of the peer token associated with the |
+
diff --git a/deploy/charts/rook-ceph/templates/resources.yaml b/deploy/charts/rook-ceph/templates/resources.yaml index 3bebd7afb..17db9db4a 100644 --- a/deploy/charts/rook-ceph/templates/resources.yaml +++ b/deploy/charts/rook-ceph/templates/resources.yaml @@ -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. diff --git a/deploy/examples/crds.yaml b/deploy/examples/crds.yaml index da344e03d..90d947ff9 100644 --- a/deploy/examples/crds.yaml +++ b/deploy/examples/crds.yaml @@ -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. diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index d83c1a0ec..d70aeb596 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/pkg/apis/ceph.rook.io/v1/types.go @@ -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 diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go index 70be2d75f..bdb91d624 100644 --- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go +++ b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go @@ -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 diff --git a/pkg/daemon/ceph/client/mirror.go b/pkg/daemon/ceph/client/mirror.go index 96c5b54c5..6bab01205 100644 --- a/pkg/daemon/ceph/client/mirror.go +++ b/pkg/daemon/ceph/client/mirror.go @@ -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) diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index 4e92de553..80392f37f 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/pkg/operator/ceph/cluster/mon/health_test.go @@ -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{}, + }, + }, + } +} diff --git a/pkg/operator/ceph/controller/mirror_peer.go b/pkg/operator/ceph/controller/mirror_peer.go index 3f0ba5550..cb8ff9c24 100644 --- a/pkg/operator/ceph/controller/mirror_peer.go +++ b/pkg/operator/ceph/controller/mirror_peer.go @@ -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 +} diff --git a/pkg/operator/ceph/controller/predicate.go b/pkg/operator/ceph/controller/predicate.go index 3e25d3d0e..4acc32ba4 100644 --- a/pkg/operator/ceph/controller/predicate.go +++ b/pkg/operator/ceph/controller/predicate.go @@ -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 + }, + } +} diff --git a/pkg/operator/ceph/pool/controller.go b/pkg/operator/ceph/pool/controller.go index 7b44c840b..505e80599 100644 --- a/pkg/operator/ceph/pool/controller.go +++ b/pkg/operator/ceph/pool/controller.go @@ -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") diff --git a/pkg/operator/ceph/pool/controller_test.go b/pkg/operator/ceph/pool/controller_test.go index 861f80afb..560e7695f 100644 --- a/pkg/operator/ceph/pool/controller_test.go +++ b/pkg/operator/ceph/pool/controller_test.go @@ -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) + }) +} diff --git a/pkg/operator/ceph/pool/status.go b/pkg/operator/ceph/pool/status.go index d66274fc9..24e226181 100644 --- a/pkg/operator/ceph/pool/status.go +++ b/pkg/operator/ceph/pool/status.go @@ -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 }