Merge pull request #16217 from rook/mergify/bp/release-1.18/pr-16029
OSD: implement automated cephx key rotation for osds (backport #16029)
This commit is contained in:
13 files changed
+560
-31
No files matched your search
@@ -5058,6 +5058,19 @@ CephxStatus
|
||||
</tr>
|
||||
<tr>
|
||||
<td>
|
||||
<code>osd</code><br/>
|
||||
<em>
|
||||
<a href="#ceph.rook.io/v1.CephxStatus">
|
||||
CephxStatus
|
||||
</a>
|
||||
</em>
|
||||
</td>
|
||||
<td>
|
||||
<p>OSD shows the CephX key status of of OSDs</p>
|
||||
</td>
|
||||
</tr>
|
||||
<tr>
|
||||
<td>
|
||||
<code>rbdMirrorPeer</code><br/>
|
||||
<em>
|
||||
<a href="#ceph.rook.io/v1.CephxStatus">
|
||||
|
||||
@@ -5996,6 +5996,28 @@ spec:
|
||||
format: int32
|
||||
type: integer
|
||||
type: object
|
||||
osd:
|
||||
description: OSD shows the CephX key status of of OSDs
|
||||
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
|
||||
rbdMirrorPeer:
|
||||
description: RBDMirrorPeer represents the cephx key rotation status of the `rbd-mirror-peer` user
|
||||
properties:
|
||||
|
||||
@@ -5994,6 +5994,28 @@ spec:
|
||||
format: int32
|
||||
type: integer
|
||||
type: object
|
||||
osd:
|
||||
description: OSD shows the CephX key status of of OSDs
|
||||
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
|
||||
rbdMirrorPeer:
|
||||
description: RBDMirrorPeer represents the cephx key rotation status of the `rbd-mirror-peer` user
|
||||
properties:
|
||||
|
||||
@@ -744,6 +744,8 @@ type LocalCephxStatus struct {
|
||||
type ClusterCephxStatus struct {
|
||||
// Mgr represents the cephx key rotation status of the ceph manager daemon
|
||||
Mgr *CephxStatus `json:"mgr,omitempty"`
|
||||
// OSD shows the CephX key status of of OSDs
|
||||
OSD *CephxStatus `json:"osd,omitempty"`
|
||||
// RBDMirrorPeer represents the cephx key rotation status of the `rbd-mirror-peer` user
|
||||
RBDMirrorPeer *CephxStatus `json:"rbdMirrorPeer,omitempty"`
|
||||
// CSI shows the CephX key status for Ceph-CSI components.
|
||||
|
||||
@@ -2080,6 +2080,11 @@ func (in *ClusterCephxStatus) DeepCopyInto(out *ClusterCephxStatus) {
|
||||
*out = new(CephxStatus)
|
||||
**out = **in
|
||||
}
|
||||
if in.OSD != nil {
|
||||
in, out := &in.OSD, &out.OSD
|
||||
*out = new(CephxStatus)
|
||||
**out = **in
|
||||
}
|
||||
if in.RBDMirrorPeer != nil {
|
||||
in, out := &in.RBDMirrorPeer, &out.RBDMirrorPeer
|
||||
*out = new(CephxStatus)
|
||||
|
||||
@@ -798,6 +798,7 @@ func (c *cluster) fetchSecretValue(selector v1.SecretKeySelector) (string, error
|
||||
func initClusterCephxStatus(c *clusterd.Context, cluster *cephv1.CephCluster) error {
|
||||
uninitializedStatus := keyring.UninitializedCephxStatus()
|
||||
cluster.Status.Cephx = &cephv1.ClusterCephxStatus{
|
||||
OSD: &uninitializedStatus,
|
||||
RBDMirrorPeer: &uninitializedStatus,
|
||||
Mgr: &uninitializedStatus,
|
||||
CSI: &cephv1.CephxStatusWithKeyCount{
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
kms "github.com/rook/rook/pkg/daemon/ceph/osd/kms"
|
||||
osdconfig "github.com/rook/rook/pkg/operator/ceph/cluster/osd/config"
|
||||
"github.com/rook/rook/pkg/operator/ceph/config/keyring"
|
||||
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
@@ -94,6 +95,12 @@ func (c *createConfig) createNewOSDsFromStatus(
|
||||
logger.Debugf("not creating deployment for OSD %d which already exists", osd.ID)
|
||||
continue
|
||||
}
|
||||
|
||||
// osd prepare jobs don't generate cephx status info for OSDs. since this is a new OSD
|
||||
// deployment, generate a first-deploy cephx status
|
||||
status.OSDs[i].CephxStatus = keyring.UpdatedCephxStatus(false, c.cluster.spec.Security.CephX.Daemon,
|
||||
c.cluster.clusterInfo.CephVersion, keyring.UninitializedCephxStatus())
|
||||
|
||||
if status.PvcBackedOSD {
|
||||
logger.Infof("creating OSD %d on PVC %q", osd.ID, nodeOrPVCName)
|
||||
err := createDaemonOnPVCFunc(c.cluster, &status.OSDs[i], nodeOrPVCName, c.provisionConfig)
|
||||
|
||||
@@ -20,8 +20,8 @@ package osd
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -46,6 +46,7 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/util/retry"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -75,6 +76,9 @@ const (
|
||||
osdStore = "osd-store"
|
||||
deviceType = "device-type"
|
||||
encrypted = "encrypted"
|
||||
|
||||
// CephxStatus is applied to each OSD deployment as value of this annotation key
|
||||
cephxStatusAnnotationKey = "cephx-status"
|
||||
)
|
||||
|
||||
// Cluster keeps track of the OSDs
|
||||
@@ -119,12 +123,13 @@ type OSDInfo struct {
|
||||
CVMode string `json:"lv-mode"`
|
||||
Store string `json:"store"`
|
||||
// Ensure the OSD daemon has affinity with the same topology from the OSD prepare pod
|
||||
TopologyAffinity string `json:"topologyAffinity"`
|
||||
Encrypted bool `json:"encrypted"`
|
||||
ExportService bool `json:"exportService"`
|
||||
NodeName string `json:"nodeName"`
|
||||
PVCName string `json:"pvcName"`
|
||||
DeviceType string `json:"device-type"`
|
||||
TopologyAffinity string `json:"topologyAffinity"`
|
||||
Encrypted bool `json:"encrypted"`
|
||||
ExportService bool `json:"exportService"`
|
||||
NodeName string `json:"nodeName"`
|
||||
PVCName string `json:"pvcName"`
|
||||
DeviceType string `json:"device-type"`
|
||||
CephxStatus cephv1.CephxStatus `json:"cephxStatus"`
|
||||
}
|
||||
|
||||
// OrchestrationStatus represents the status of an OSD orchestration
|
||||
@@ -296,7 +301,7 @@ func (c *Cluster) Start() error {
|
||||
return errors.Wrap(err, "failed post reconcile of osd properties")
|
||||
}
|
||||
|
||||
err = c.updateCephStorageStatus()
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "failed to update ceph storage status")
|
||||
}
|
||||
@@ -724,6 +729,15 @@ func (c *Cluster) getOSDInfo(d *appsv1.Deployment) (OSDInfo, error) {
|
||||
osd.PVCName = d.Labels[OSDOverPVCLabelKey]
|
||||
}
|
||||
|
||||
cephxStatus := cephv1.CephxStatus{}
|
||||
cephxRaw, ok := d.Spec.Template.Annotations[cephxStatusAnnotationKey]
|
||||
if ok {
|
||||
if err := json.Unmarshal([]byte(cephxRaw), &cephxStatus); err != nil {
|
||||
return OSDInfo{}, errors.Wrapf(err, "failed to unmarshal cephx status %q for deployment %q", cephxRaw, d.Name)
|
||||
}
|
||||
}
|
||||
osd.CephxStatus = cephxStatus
|
||||
|
||||
return osd, nil
|
||||
}
|
||||
|
||||
@@ -970,8 +984,7 @@ func (c *Cluster) waitForHealthyPGs() (bool, error) {
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (c *Cluster) updateCephStorageStatus() error {
|
||||
cephCluster := cephv1.CephCluster{}
|
||||
func (c *Cluster) updateCephOsdStorageStatus() error {
|
||||
cephClusterStorage := cephv1.CephStorage{}
|
||||
|
||||
deviceClasses, err := cephclient.GetDeviceClasses(c.context, c.clusterInfo)
|
||||
@@ -983,7 +996,7 @@ func (c *Cluster) updateCephStorageStatus() error {
|
||||
cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: deviceClass})
|
||||
}
|
||||
|
||||
osdStore, err := c.getOSDStoreStatus()
|
||||
osdStore, cephx, err := c.getOSDStoreStatus()
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "failed to get osd store status")
|
||||
}
|
||||
@@ -1002,44 +1015,79 @@ func (c *Cluster) updateCephStorageStatus() error {
|
||||
cephClusterStorage.OSD.MigrationStatus.Pending = len(migrationConfig.osds)
|
||||
}
|
||||
|
||||
err = c.context.Client.Get(c.clusterInfo.Context, c.clusterInfo.NamespacedName(), &cephCluster)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
|
||||
return nil
|
||||
err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
|
||||
cephCluster := cephv1.CephCluster{}
|
||||
err := c.context.Client.Get(c.clusterInfo.Context, c.clusterInfo.NamespacedName(), &cephCluster)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
|
||||
return nil
|
||||
}
|
||||
return errors.Wrapf(err, "failed to retrieve ceph cluster %q to update ceph Storage", c.clusterInfo.NamespacedName().Name)
|
||||
}
|
||||
return errors.Wrapf(err, "failed to retrieve ceph cluster %q to update ceph Storage", c.clusterInfo.NamespacedName().Name)
|
||||
}
|
||||
if !reflect.DeepEqual(cephCluster.Status.CephStorage, cephClusterStorage) {
|
||||
|
||||
cephCluster.Status.CephStorage = &cephClusterStorage
|
||||
if err := reporting.UpdateStatus(c.context.Client, &cephCluster); err != nil {
|
||||
return errors.Wrapf(err, "failed to update cluster %q Storage.", c.clusterInfo.NamespacedName().Name)
|
||||
|
||||
if cephx != nil {
|
||||
if cephCluster.Status.Cephx == nil {
|
||||
cephCluster.Status.Cephx = &cephv1.ClusterCephxStatus{}
|
||||
}
|
||||
cephCluster.Status.Cephx.OSD = cephx
|
||||
}
|
||||
|
||||
if err := reporting.UpdateStatus(c.context.Client, &cephCluster); err != nil {
|
||||
return errors.Wrapf(err, "failed to update cluster %q status", c.clusterInfo.NamespacedName().Name)
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cluster) getOSDStoreStatus() (*cephv1.OSDStatus, error) {
|
||||
func (c *Cluster) getOSDStoreStatus() (*cephv1.OSDStatus, *cephv1.CephxStatus, error) {
|
||||
label := fmt.Sprintf("%s=%s", k8sutil.AppAttr, AppName)
|
||||
osdDeployments, err := k8sutil.GetDeployments(c.clusterInfo.Context, c.context.Clientset, c.clusterInfo.Namespace, label)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
return nil, nil
|
||||
return nil, nil, nil
|
||||
}
|
||||
return nil, errors.Wrap(err, "failed to get osd deployments")
|
||||
return nil, nil, errors.Wrap(err, "failed to get osd deployments")
|
||||
}
|
||||
|
||||
storeType := map[string]int{}
|
||||
var minCephxStatus *cephv1.CephxStatus = nil
|
||||
for i := range osdDeployments.Items {
|
||||
if osdStore, ok := osdDeployments.Items[i].Labels[osdStore]; ok {
|
||||
d := osdDeployments.Items[i]
|
||||
|
||||
if osdStore, ok := d.Labels[osdStore]; ok {
|
||||
storeType[osdStore]++
|
||||
}
|
||||
|
||||
// determine min cephx status on daemons
|
||||
var cephxStatus cephv1.CephxStatus
|
||||
cephxRaw, ok := d.Spec.Template.Annotations[cephxStatusAnnotationKey]
|
||||
if !ok || cephxRaw == "" {
|
||||
cephxStatus = cephv1.CephxStatus{} // no annotation or empty annotation means empty status
|
||||
} else {
|
||||
err := json.Unmarshal([]byte(cephxRaw), &cephxStatus)
|
||||
if err != nil {
|
||||
return nil, nil, errors.Wrapf(err, "failed to unmarshal cephx status %q for deployment %q", cephxRaw, d.Name)
|
||||
}
|
||||
}
|
||||
if minCephxStatus == nil || cephxStatus.KeyGeneration < minCephxStatus.KeyGeneration {
|
||||
// assume that min key generation will also give the right min ceph version
|
||||
// may not be true in extreme corner cases, but it should be true almost all the time
|
||||
minCephxStatus = &cephxStatus
|
||||
}
|
||||
}
|
||||
|
||||
return &cephv1.OSDStatus{
|
||||
StoreType: storeType,
|
||||
}, nil
|
||||
}, minCephxStatus, nil
|
||||
}
|
||||
|
||||
func getOSDLocationFromArgs(args []string) (string, bool) {
|
||||
|
||||
@@ -842,7 +842,7 @@ func TestUpdateCephStorageStatus(t *testing.T) {
|
||||
c := New(context, clusterInfo, cephCluster.Spec, "myversion")
|
||||
|
||||
t.Run("verify ssd device class added to storage status", func(t *testing.T) {
|
||||
err := c.updateCephStorageStatus()
|
||||
err := c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
@@ -868,7 +868,7 @@ func TestUpdateCephStorageStatus(t *testing.T) {
|
||||
if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
|
||||
logger.Errorf("Error creating fake deployment: %v", err)
|
||||
}
|
||||
err := c.updateCephStorageStatus()
|
||||
err := c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
@@ -896,7 +896,7 @@ func TestUpdateCephStorageStatus(t *testing.T) {
|
||||
if _, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{}); err != nil {
|
||||
logger.Errorf("Error creating fake deployment: %v", err)
|
||||
}
|
||||
err := c.updateCephStorageStatus()
|
||||
err := c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
@@ -907,6 +907,211 @@ func TestUpdateCephStorageStatus(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func Test_updateCephOsdStorageStatus_cephx(t *testing.T) {
|
||||
ctx := context.TODO()
|
||||
clusterInfo := cephclient.AdminTestClusterInfo("fake")
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
|
||||
if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
|
||||
// Mock executor for OSD crush class list command, returning ssd as available device class
|
||||
return `["ssd"]`, nil
|
||||
}
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
|
||||
cephCluster := &cephv1.CephCluster{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "testing",
|
||||
Namespace: "fake",
|
||||
},
|
||||
Spec: cephv1.ClusterSpec{},
|
||||
}
|
||||
// Objects to track in the fake client.
|
||||
object := []runtime.Object{
|
||||
cephCluster,
|
||||
}
|
||||
s := scheme.Scheme
|
||||
// Create a fake client to mock API calls.
|
||||
client := clientfake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
|
||||
|
||||
context := &clusterd.Context{
|
||||
Executor: executor,
|
||||
Client: client,
|
||||
Clientset: testexec.New(t, 2),
|
||||
}
|
||||
|
||||
// Initializing an OSD monitoring
|
||||
c := New(context, clusterInfo, cephCluster.Spec, "myversion")
|
||||
|
||||
// Tests are not independent and rely on prior tests's changes
|
||||
|
||||
t.Run("no osd deployments, no cephcluster status", func(t *testing.T) {
|
||||
err := c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.Nil(t, cephCluster.Status.Cephx)
|
||||
})
|
||||
|
||||
t.Run("unset cephx status, no cephcluster status", func(t *testing.T) {
|
||||
deployment := &apps.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "osd0",
|
||||
Namespace: clusterInfo.Namespace,
|
||||
Labels: map[string]string{
|
||||
k8sutil.AppAttr: AppName,
|
||||
k8sutil.ClusterAttr: clusterInfo.Namespace,
|
||||
OsdIdLabelKey: "0",
|
||||
},
|
||||
},
|
||||
Spec: apps.DeploymentSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Annotations: map[string]string{},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, cephCluster.Status.Cephx)
|
||||
assert.Empty(t, cephCluster.Status.Cephx.OSD)
|
||||
})
|
||||
|
||||
t.Run("empty string cephx status, no cephcluster status", func(t *testing.T) {
|
||||
deployment := &apps.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "osd0",
|
||||
Namespace: clusterInfo.Namespace,
|
||||
Labels: map[string]string{
|
||||
k8sutil.AppAttr: AppName,
|
||||
k8sutil.ClusterAttr: clusterInfo.Namespace,
|
||||
OsdIdLabelKey: "0",
|
||||
},
|
||||
},
|
||||
Spec: apps.DeploymentSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Annotations: map[string]string{
|
||||
cephxStatusAnnotationKey: ``,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Update(ctx, deployment, metav1.UpdateOptions{})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, cephCluster.Status.Cephx)
|
||||
assert.Empty(t, cephCluster.Status.Cephx.OSD)
|
||||
})
|
||||
|
||||
t.Run("gen 1 cephx status", func(t *testing.T) {
|
||||
deployment := &apps.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "osd0",
|
||||
Namespace: clusterInfo.Namespace,
|
||||
Labels: map[string]string{
|
||||
k8sutil.AppAttr: AppName,
|
||||
k8sutil.ClusterAttr: clusterInfo.Namespace,
|
||||
OsdIdLabelKey: "0",
|
||||
},
|
||||
},
|
||||
Spec: apps.DeploymentSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Annotations: map[string]string{
|
||||
cephxStatusAnnotationKey: `{"keyGeneration": 1, "keyCephVersion": "v20"}`,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Update(ctx, deployment, metav1.UpdateOptions{})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, cephCluster.Status.Cephx)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyGeneration: 1, KeyCephVersion: "v20"}, *cephCluster.Status.Cephx.OSD)
|
||||
})
|
||||
|
||||
t.Run("gen 1 and unset cephx status", func(t *testing.T) {
|
||||
deployment := &apps.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "osd1",
|
||||
Namespace: clusterInfo.Namespace,
|
||||
Labels: map[string]string{
|
||||
k8sutil.AppAttr: AppName,
|
||||
k8sutil.ClusterAttr: clusterInfo.Namespace,
|
||||
OsdIdLabelKey: "1",
|
||||
},
|
||||
},
|
||||
Spec: apps.DeploymentSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Annotations: map[string]string{},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Create(ctx, deployment, metav1.CreateOptions{})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, cephCluster.Status.Cephx)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyGeneration: 0, KeyCephVersion: ""}, *cephCluster.Status.Cephx.OSD)
|
||||
})
|
||||
|
||||
t.Run("gen 1 and gen 2 cephx status", func(t *testing.T) {
|
||||
deployment := &apps.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "osd1",
|
||||
Namespace: clusterInfo.Namespace,
|
||||
Labels: map[string]string{
|
||||
k8sutil.AppAttr: AppName,
|
||||
k8sutil.ClusterAttr: clusterInfo.Namespace,
|
||||
OsdIdLabelKey: "1",
|
||||
},
|
||||
},
|
||||
Spec: apps.DeploymentSpec{
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Annotations: map[string]string{
|
||||
cephxStatusAnnotationKey: `{"keyGeneration": 2, "keyCephVersion": "v19"}`,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
_, err := context.Clientset.AppsV1().Deployments(clusterInfo.Namespace).Update(ctx, deployment, metav1.UpdateOptions{})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = c.updateCephOsdStorageStatus()
|
||||
assert.NoError(t, err)
|
||||
err = context.Client.Get(clusterInfo.Context, clusterInfo.NamespacedName(), cephCluster)
|
||||
assert.NoError(t, err)
|
||||
assert.NotNil(t, cephCluster.Status.Cephx)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyGeneration: 1, KeyCephVersion: "v20"}, *cephCluster.Status.Cephx.OSD)
|
||||
})
|
||||
}
|
||||
|
||||
func TestGetOSDLocationFromArgs(t *testing.T) {
|
||||
args := []string{"--id", "2", "--crush-location=root=default host=minikube"}
|
||||
osdLocaiton, locationFound := getOSDLocationFromArgs(args)
|
||||
|
||||
@@ -18,6 +18,7 @@ limitations under the License.
|
||||
package osd
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path"
|
||||
"path/filepath"
|
||||
@@ -641,11 +642,21 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd *OSDInfo, provision
|
||||
"",
|
||||
))
|
||||
|
||||
// apply status to each OSD so its key isn't rotated again after a rotation is applied via
|
||||
// daemon restart. important b/c osd update is often interrupted by reconcile requeues
|
||||
cephxStatusVal, err := json.Marshal(osd.CephxStatus)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "failed to marshal OSD info CephX status %#v for OSD deployment %q", osd.CephxStatus, deploymentName)
|
||||
}
|
||||
|
||||
envVars = append(envVars, controller.ApplyNetworkEnv(&c.spec)...)
|
||||
podTemplateSpec := v1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: AppName,
|
||||
Labels: osdLabels,
|
||||
Annotations: map[string]string{
|
||||
cephxStatusAnnotationKey: string(cephxStatusVal),
|
||||
},
|
||||
},
|
||||
Spec: v1.PodSpec{
|
||||
RestartPolicy: v1.RestartPolicyAlways,
|
||||
@@ -768,7 +779,7 @@ func (c *Cluster) makeDeployment(osdProps osdProperties, osd *OSDInfo, provision
|
||||
cephv1.GetOSDLabels(c.spec.Labels).ApplyToObjectMeta(&deployment.ObjectMeta)
|
||||
cephv1.GetOSDLabels(c.spec.Labels).ApplyToObjectMeta(&deployment.Spec.Template.ObjectMeta)
|
||||
controller.AddCephVersionLabelToDeployment(c.clusterInfo.CephVersion, deployment)
|
||||
err := c.clusterInfo.OwnerInfo.SetControllerReference(deployment)
|
||||
err = c.clusterInfo.OwnerInfo.SetControllerReference(deployment)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "failed to set owner reference to osd deployment %q", deployment.Name)
|
||||
}
|
||||
|
||||
@@ -166,7 +166,7 @@ func testPodDevices(t *testing.T, dataDir, deviceName string, allDevices bool) {
|
||||
assert.Equal(t, AppName, deployment.Spec.Template.ObjectMeta.Name)
|
||||
assert.Equal(t, AppName, deployment.Spec.Template.ObjectMeta.Labels["app"])
|
||||
assert.Equal(t, c.clusterInfo.Namespace, deployment.Spec.Template.ObjectMeta.Labels["rook_cluster"])
|
||||
assert.Equal(t, 1, len(deployment.Spec.Template.ObjectMeta.Annotations))
|
||||
assert.Equal(t, 2, len(deployment.Spec.Template.ObjectMeta.Annotations))
|
||||
|
||||
assert.Equal(t, 4, len(deployment.Spec.Template.Spec.InitContainers))
|
||||
initCont := deployment.Spec.Template.Spec.InitContainers[0]
|
||||
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
"github.com/pkg/errors"
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
|
||||
"github.com/rook/rook/pkg/operator/ceph/config/keyring"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
appsv1 "k8s.io/api/apps/v1"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
@@ -169,6 +170,13 @@ func (c *updateConfig) updateExistingOSDs(errs *provisionErrors) {
|
||||
continue
|
||||
}
|
||||
|
||||
cephxStatus, err := c.cluster.rotateCephxKey(osdInfo)
|
||||
if err != nil {
|
||||
// user-desired rotation failed, so report an error, but continue to try to update the OSD deployment
|
||||
errs.addError("%v", errors.Wrapf(err, "failed to rotate cephx key for OSD %d", osdID))
|
||||
}
|
||||
osdInfo.CephxStatus = cephxStatus // returned status is always correct
|
||||
|
||||
var updatedDep *appsv1.Deployment
|
||||
|
||||
if c.cluster.spec.Network.MultiClusterService.Enabled {
|
||||
@@ -390,3 +398,35 @@ func (c *Cluster) getOSDDeployments() (*appsv1.DeploymentList, error) {
|
||||
}
|
||||
return deps, nil
|
||||
}
|
||||
|
||||
// if needed, rotate cephx key for the OSD
|
||||
// always returns the cephx status that should be applied to the OSD annotation, even in error case
|
||||
func (c *Cluster) rotateCephxKey(osdInfo OSDInfo) (cephv1.CephxStatus, error) {
|
||||
// TODO: for rotation WithCephVersionUpdate fix this to have the right runningCephVersion and desiredCephVersion
|
||||
runningCephVersion := c.clusterInfo.CephVersion
|
||||
desiredCephVersion := c.clusterInfo.CephVersion
|
||||
shouldRotate, err := keyring.ShouldRotateCephxKeys(c.spec.Security.CephX.Daemon,
|
||||
runningCephVersion, desiredCephVersion, osdInfo.CephxStatus)
|
||||
if err != nil {
|
||||
return osdInfo.CephxStatus, errors.Wrapf(err, "failed to determine if cephx key for OSD %d needs rotated", osdInfo.ID)
|
||||
}
|
||||
|
||||
didRotateCephxStatus := keyring.UpdatedCephxStatus(true, c.spec.Security.CephX.Daemon,
|
||||
c.clusterInfo.CephVersion, osdInfo.CephxStatus)
|
||||
didNotRotateCephxStatus := keyring.UpdatedCephxStatus(false, c.spec.Security.CephX.Daemon,
|
||||
c.clusterInfo.CephVersion, osdInfo.CephxStatus)
|
||||
|
||||
if !shouldRotate {
|
||||
return didNotRotateCephxStatus, nil
|
||||
}
|
||||
|
||||
logger.Infof("rotating cephx key of OSD %d for CephCluster in namespace %q", osdInfo.ID, c.clusterInfo.Namespace)
|
||||
user := fmt.Sprintf("osd.%d", osdInfo.ID)
|
||||
// Note: OSD key is not stored in k8s secret; rotated key is picked up by OSD init container
|
||||
_, err = cephclient.AuthRotate(c.context, c.clusterInfo, user)
|
||||
if err != nil {
|
||||
return didNotRotateCephxStatus, errors.Wrapf(err, "failed to rotate cephx key for OSD %d", osdInfo.ID)
|
||||
}
|
||||
|
||||
return didRotateCephxStatus, nil
|
||||
}
|
||||
@@ -28,6 +28,7 @@ import (
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
|
||||
cephclientfake "github.com/rook/rook/pkg/daemon/ceph/client/fake"
|
||||
"github.com/rook/rook/pkg/operator/ceph/config/keyring"
|
||||
"github.com/rook/rook/pkg/operator/ceph/controller"
|
||||
cephver "github.com/rook/rook/pkg/operator/ceph/version"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
@@ -779,3 +780,155 @@ func Test_existenceList(t *testing.T) {
|
||||
assert.True(t, l.Exists(1))
|
||||
assert.Equal(t, 4, l.Len())
|
||||
}
|
||||
|
||||
func TestCluster_rotateCephxKey(t *testing.T) {
|
||||
// auth rotate returns an array instead of a single object
|
||||
rotatedKeyJson := `[{"key":"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=="}]`
|
||||
rotateCalledForEntity := ""
|
||||
sharedExecutor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
|
||||
t.Logf("command: %s %v", command, args)
|
||||
if command == "ceph" && args[0] == "auth" && args[1] == "rotate" {
|
||||
rotateCalledForEntity = args[2]
|
||||
return rotatedKeyJson, nil
|
||||
}
|
||||
panic(fmt.Sprintf("unexpected command %q %v", command, args))
|
||||
},
|
||||
}
|
||||
|
||||
newTest := func(daemonCephxConfig cephv1.CephxConfig) *Cluster {
|
||||
rotateCalledForEntity = "" // reset to empty each test
|
||||
|
||||
clusterd := clusterd.Context{
|
||||
Executor: sharedExecutor,
|
||||
}
|
||||
clusterInfo := cephclient.ClusterInfo{
|
||||
Context: context.TODO(),
|
||||
Namespace: "ns",
|
||||
CephVersion: cephver.CephVersion{Major: 20, Minor: 2},
|
||||
}
|
||||
return &Cluster{
|
||||
context: &clusterd,
|
||||
clusterInfo: &clusterInfo,
|
||||
spec: cephv1.ClusterSpec{
|
||||
Security: cephv1.ClusterSecuritySpec{
|
||||
CephX: cephv1.ClusterCephxConfig{
|
||||
Daemon: daemonCephxConfig,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("empty status and config", func(t *testing.T) {
|
||||
c := newTest(cephv1.CephxConfig{})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 1,
|
||||
CephxStatus: cephv1.CephxStatus{},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "")
|
||||
})
|
||||
|
||||
t.Run("empty status, disabled config", func(t *testing.T) { // brownfield, no rotation
|
||||
c := newTest(cephv1.CephxConfig{
|
||||
KeyRotationPolicy: "Disabled",
|
||||
KeyGeneration: 3,
|
||||
})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 1,
|
||||
CephxStatus: cephv1.CephxStatus{},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "")
|
||||
})
|
||||
|
||||
t.Run("empty status, enabled config", func(t *testing.T) { // brownfield with rotation
|
||||
c := newTest(cephv1.CephxConfig{
|
||||
KeyRotationPolicy: "KeyGeneration",
|
||||
KeyGeneration: 3,
|
||||
})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 1,
|
||||
CephxStatus: cephv1.CephxStatus{},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyCephVersion: "20.2.0-0", KeyGeneration: 3}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "osd.1")
|
||||
})
|
||||
|
||||
t.Run("undefined status, empty config", func(t *testing.T) { // greenfield, no rotation
|
||||
c := newTest(cephv1.CephxConfig{})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 1,
|
||||
CephxStatus: keyring.UninitializedCephxStatus(), // shouldn't happen in reality
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
// results in unnecessary OSD restart, but ensures that future rotations aren't blocked
|
||||
// in the unlikely event the uninitialized status is erroneously applied to osd deployment
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyCephVersion: "20.2.0-0", KeyGeneration: 1}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "")
|
||||
})
|
||||
|
||||
t.Run("set status, empty config", func(t *testing.T) {
|
||||
c := newTest(cephv1.CephxConfig{})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 1,
|
||||
CephxStatus: cephv1.CephxStatus{KeyGeneration: 1, KeyCephVersion: "19.2.6-0"},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyCephVersion: "19.2.6-0", KeyGeneration: 1}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "")
|
||||
})
|
||||
|
||||
t.Run("set status, enabled config", func(t *testing.T) {
|
||||
c := newTest(cephv1.CephxConfig{
|
||||
KeyRotationPolicy: "KeyGeneration",
|
||||
KeyGeneration: 4,
|
||||
})
|
||||
osdInfo := OSDInfo{
|
||||
ID: 2,
|
||||
CephxStatus: cephv1.CephxStatus{KeyGeneration: 1, KeyCephVersion: "19.2.6-0"},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyCephVersion: "20.2.0-0", KeyGeneration: 4}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "osd.2")
|
||||
})
|
||||
|
||||
t.Run("auth rotate failure", func(t *testing.T) {
|
||||
// this case shouldn't happen in code, but it could repair a botched greenfield deploy
|
||||
c := newTest(cephv1.CephxConfig{
|
||||
KeyRotationPolicy: "KeyGeneration",
|
||||
KeyGeneration: 4,
|
||||
})
|
||||
c.context.Executor = &exectest.MockExecutor{
|
||||
MockExecuteCommandWithOutput: func(command string, arg ...string) (string, error) {
|
||||
rotateCalledForEntity = arg[2]
|
||||
return "", fmt.Errorf("mock error")
|
||||
},
|
||||
}
|
||||
osdInfo := OSDInfo{
|
||||
ID: 2,
|
||||
CephxStatus: cephv1.CephxStatus{KeyGeneration: 1, KeyCephVersion: "19.2.6-0"},
|
||||
}
|
||||
|
||||
cephxStatus, err := c.rotateCephxKey(osdInfo)
|
||||
assert.Error(t, err)
|
||||
assert.Equal(t, cephv1.CephxStatus{KeyCephVersion: "19.2.6-0", KeyGeneration: 1}, cephxStatus)
|
||||
assert.Equal(t, rotateCalledForEntity, "osd.2")
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user