Merge pull request #9800 from subhamkrai/events

core: add k8s events in blockpool reconciler
This commit is contained in:
Blaine Gardner authored and GitHub committed 2022-03-14 09:51:46 -06:00
commit 992c1e620a
16 files changed
+232 -183

No files matched your search

+22 -21
View File
@@ -46,6 +46,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
)
const (
@@ -69,6 +70,7 @@ type ReconcileCephClient struct {
context *clusterd.Context
clusterInfo *cephclient.ClusterInfo
opManagerContext context.Context
recorder record.EventRecorder
}
// Add creates a new CephClient Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -84,6 +86,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
scheme: mgr.GetScheme(),
context: context,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -119,31 +122,27 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileCephClient) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile %v", err)
}
return reconcileResponse, err
reconcileResponse, cephClient, err := r.reconcile(request)
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephClient, reconcileResponse, err)
}
func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephClient, error) {
// Fetch the CephClient instance
cephClient := &cephv1.CephClient{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephClient)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("cephClient resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephClient, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get cephClient")
return reconcile.Result{}, *cephClient, errors.Wrap(err, "failed to get cephClient")
}
// Set a finalizer so we can do cleanup before the object goes away
err = opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, cephClient)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to add finalizer")
return reconcile.Result{}, *cephClient, errors.Wrap(err, "failed to add finalizer")
}
// The CR was just created, initializing status fields
@@ -164,19 +163,19 @@ func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Re
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephClient)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to remove finalizer")
return opcontroller.ImmediateRetryResult, *cephClient, errors.Wrap(err, "failed to remove finalizer")
}
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephClient, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephClient, nil
}
// Populate clusterInfo during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
return reconcile.Result{}, *cephClient, errors.Wrap(err, "failed to populate cluster info")
}
r.clusterInfo.Context = r.opManagerContext
@@ -185,23 +184,25 @@ func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Re
logger.Debugf("deleting pool %q", cephClient.Name)
err := r.deleteClient(cephClient)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph client %q", cephClient.Name)
return reconcile.Result{}, *cephClient, errors.Wrapf(err, "failed to delete ceph client %q", cephClient.Name)
}
r.recorder.Eventf(cephClient, v1.EventTypeNormal, string(cephv1.ReconcileStarted), "deleting CephClient %q", cephClient.Name)
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephClient)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephClient, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephClient, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephClient, nil
}
// validate the client settings
err = ValidateClient(r.context, cephClient)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to validate client %q arguments", cephClient.Name)
return reconcile.Result{}, *cephClient, errors.Wrapf(err, "failed to validate client %q arguments", cephClient.Name)
}
// Create or Update client
@@ -209,10 +210,10 @@ func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Re
if err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephClient, nil
}
r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure)
return reconcile.Result{}, errors.Wrapf(err, "failed to create or update client %q", cephClient.Name)
return reconcile.Result{}, *cephClient, errors.Wrapf(err, "failed to create or update client %q", cephClient.Name)
}
// Success! Let's update the status
@@ -220,7 +221,7 @@ func (r *ReconcileCephClient) reconcile(request reconcile.Request) (reconcile.Re
// Return and do not requeue
logger.Debug("done reconciling")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephClient, nil
}
// Create the client
@@ -36,6 +36,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -172,6 +173,7 @@ func TestCephClientController(t *testing.T) {
scheme: s,
context: c,
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
// Mock request to simulate Reconcile() being called on an event for a
@@ -220,6 +222,7 @@ func TestCephClientController(t *testing.T) {
scheme: s,
context: c,
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
assert.True(t, res.Requeue)
@@ -278,6 +281,7 @@ func TestCephClientController(t *testing.T) {
scheme: s,
context: c,
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
res, err = r.Reconcile(ctx, req)
+17 -14
View File
@@ -37,6 +37,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -78,6 +79,7 @@ type ReconcileCephRBDMirror struct {
peers map[string]*peerSpec
opManagerContext context.Context
opConfig opcontroller.OperatorConfig
recorder record.EventRecorder
}
// peerSpec represents peer details
@@ -102,6 +104,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
peers: make(map[string]*peerSpec),
opConfig: opConfig,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -139,26 +142,26 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileCephRBDMirror) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
reconcileResponse, cephRBDMirror, err := r.reconcile(request)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.FailedStatus)
logger.Errorf("failed to reconcile %v", err)
}
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephRBDMirror, reconcileResponse, err)
}
func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephRBDMirror, error) {
// Fetch the cephRBDMirror instance
cephRBDMirror := &cephv1.CephRBDMirror{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephRBDMirror)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("cephRBDMirror resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephRBDMirror, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get cephRBDMirror")
return reconcile.Result{}, *cephRBDMirror, errors.Wrap(err, "failed to get cephRBDMirror")
}
// The CR was just created, initializing status fields
@@ -168,14 +171,14 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
// validate the pool settings
if err := validateSpec(&cephRBDMirror.Spec); err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "invalid rbd-mirror CR %q spec", cephRBDMirror.Name)
return opcontroller.ImmediateRetryResult, *cephRBDMirror, errors.Wrapf(err, "invalid rbd-mirror CR %q spec", cephRBDMirror.Name)
}
// Make sure a CephCluster is present otherwise do nothing
cephCluster, isReadyToReconcile, _, reconcileResponse := opcontroller.IsReadyToReconcile(r.opManagerContext, r.client, request.NamespacedName, controllerName)
if !isReadyToReconcile {
logger.Debugf("CephCluster resource not ready in namespace %q, retrying in %q.", request.NamespacedName.Namespace, reconcileResponse.RequeueAfter.String())
return reconcileResponse, nil
return reconcileResponse, *cephRBDMirror, nil
}
r.cephClusterSpec = &cephCluster.Spec
@@ -183,7 +186,7 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
// Always populate it during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to populate cluster info")
return opcontroller.ImmediateRetryResult, *cephRBDMirror, errors.Wrap(err, "failed to populate cluster info")
}
// Detect desired CephCluster version
@@ -200,9 +203,9 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
if err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return reconcile.Result{}, nil
return reconcile.Result{}, *cephRBDMirror, nil
}
return reconcile.Result{}, errors.Wrap(err, "failed to detect running and desired ceph version")
return reconcile.Result{}, *cephRBDMirror, errors.Wrap(err, "failed to detect running and desired ceph version")
}
// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
@@ -210,7 +213,7 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
// Upgrade is in progress, let's wait for the mons to be done
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading, *cephRBDMirror,
opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
}
r.clusterInfo.CephVersion = *runningCephVersion
@@ -219,14 +222,14 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
logger.Debug("reconciling ceph rbd mirror peers addition")
reconcileResponse, err = r.reconcileAddBoostrapPeer(cephRBDMirror, request.NamespacedName)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to add ceph rbd mirror peer")
return opcontroller.ImmediateRetryResult, *cephRBDMirror, errors.Wrap(err, "failed to add ceph rbd mirror peer")
}
// CREATE/UPDATE
logger.Debug("reconciling ceph rbd mirror deployments")
reconcileResponse, err = r.reconcileCreateCephRBDMirror(cephRBDMirror)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to create ceph rbd mirror deployments")
return opcontroller.ImmediateRetryResult, *cephRBDMirror, errors.Wrap(err, "failed to create ceph rbd mirror deployments")
}
// Set Ready status, we are done reconciling
@@ -234,7 +237,7 @@ func (r *ReconcileCephRBDMirror) reconcile(request reconcile.Request) (reconcile
// Return and do not requeue
logger.Debug("done reconciling ceph rbd mirror")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephRBDMirror, nil
}
@@ -37,6 +37,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -143,7 +144,7 @@ func TestCephRBDMirrorController(t *testing.T) {
cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileCephRBDMirror object with the scheme and fake client.
r := &ReconcileCephRBDMirror{client: cl, scheme: s, context: c, opManagerContext: ctx}
r := &ReconcileCephRBDMirror{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
assert.True(t, res.Requeue)
@@ -154,7 +155,7 @@ func TestCephRBDMirrorController(t *testing.T) {
// Create a fake client to mock API calls.
cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileCephRBDMirror object with the scheme and fake client.
r := &ReconcileCephRBDMirror{client: cl, scheme: s, context: c, opManagerContext: ctx}
r := &ReconcileCephRBDMirror{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
assert.True(t, res.Requeue)
@@ -192,13 +193,14 @@ func TestCephRBDMirrorController(t *testing.T) {
context: c,
peers: make(map[string]*peerSpec),
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
rbdMirror.Spec.Peers.SecretNames = []string{peerSecretName}
err = r.client.Update(context.TODO(), rbdMirror)
assert.NoError(t, err)
res, err := r.Reconcile(ctx, req)
assert.Error(t, err)
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
@@ -212,6 +214,7 @@ func TestCephRBDMirrorController(t *testing.T) {
context: c,
peers: make(map[string]*peerSpec),
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
bootstrapPeerToken := `eyJmc2lkIjoiYzZiMDg3ZjItNzgyOS00ZGJiLWJjZmMtNTNkYzM0ZTBiMzVkIiwiY2xpZW50X2lkIjoicmJkLW1pcnJvci1wZWVyIiwia2V5IjoiQVFBV1lsWmZVQ1Q2RGhBQVBtVnAwbGtubDA5YVZWS3lyRVV1NEE9PSIsIm1vbl9ob3N0IjoiW3YyOjE5Mi4xNjguMTExLjEwOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTA6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjEyOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTI6Njc4OV0sW3YyOjE5Mi4xNjguMTExLjExOjMzMDAsdjE6MTkyLjE2OC4xMTEuMTE6Njc4OV0ifQ==` //nolint:gosec // This is just a var name, not a real token
peerSecret := &v1.Secret{
+18 -14
View File
@@ -37,6 +37,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -77,6 +78,7 @@ type ReconcileFilesystemMirror struct {
cephClusterSpec *cephv1.ClusterSpec
opManagerContext context.Context
opConfig opcontroller.OperatorConfig
recorder record.EventRecorder
}
// Add creates a new CephFilesystemMirror Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -93,6 +95,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
context: context,
opConfig: opConfig,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -130,26 +133,25 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileFilesystemMirror) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
reconcileResponse, cephFilesystemMirror, err := r.reconcile(request)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.FailedStatus)
logger.Errorf("failed to reconcile %v", err)
}
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephFilesystemMirror, reconcileResponse, err)
}
func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephFilesystemMirror, error) {
// Fetch the CephFilesystemMirror instance
filesystemMirror := &cephv1.CephFilesystemMirror{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, filesystemMirror)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephFilesystemMirror resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *filesystemMirror, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get CephFilesystemMirror")
return reconcile.Result{}, *filesystemMirror, errors.Wrap(err, "failed to get CephFilesystemMirror")
}
// The CR was just created, initializing status fields
@@ -161,7 +163,9 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
cephCluster, isReadyToReconcile, _, reconcileResponse := opcontroller.IsReadyToReconcile(r.opManagerContext, r.client, request.NamespacedName, controllerName)
if !isReadyToReconcile {
logger.Debugf("CephCluster resource not ready in namespace %q, retrying in %q.", request.NamespacedName.Namespace, reconcileResponse.RequeueAfter.String())
return reconcileResponse, nil
r.recorder.Event(filesystemMirror, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
return reconcileResponse, *filesystemMirror, nil
}
// Assign the clusterSpec
@@ -170,7 +174,7 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
// Populate clusterInfo
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to populate cluster info")
return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrap(err, "failed to populate cluster info")
}
// Detect desired CephCluster version
@@ -187,9 +191,9 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
if err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *filesystemMirror, nil
}
return reconcile.Result{}, errors.Wrap(err, "failed to detect running and desired ceph version")
return reconcile.Result{}, *filesystemMirror, errors.Wrap(err, "failed to detect running and desired ceph version")
}
// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
@@ -197,21 +201,21 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
// Upgrade is in progress, let's wait for the mons to be done
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading, *filesystemMirror,
opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
}
r.clusterInfo.CephVersion = *runningCephVersion
// Validate Ceph version
if !r.clusterInfo.CephVersion.IsAtLeastPacific() {
return opcontroller.ImmediateRetryResult, errors.Errorf("ceph pacific version is required to deploy cephfs mirroring, current cluster runs %q", r.clusterInfo.CephVersion.String())
return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Errorf("ceph pacific version is required to deploy cephfs mirroring, current cluster runs %q", r.clusterInfo.CephVersion.String())
}
// CREATE/UPDATE
logger.Debug("reconciling ceph filesystem mirror deployments")
reconcileResponse, err = r.reconcileFilesystemMirror(filesystemMirror)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to create ceph filesystem mirror deployments")
return opcontroller.ImmediateRetryResult, *filesystemMirror, errors.Wrap(err, "failed to create ceph filesystem mirror deployments")
}
// Set Ready status, we are done reconciling
@@ -219,7 +223,7 @@ func (r *ReconcileFilesystemMirror) reconcile(request reconcile.Request) (reconc
// Return and do not requeue
logger.Debug("done reconciling ceph filesystem mirror")
return reconcile.Result{}, nil
return reconcile.Result{}, *filesystemMirror, nil
}
@@ -38,6 +38,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -113,6 +114,7 @@ func TestCephFilesystemMirrorController(t *testing.T) {
ServiceAccount: "foo",
},
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
req := reconcile.Request{
NamespacedName: types.NamespacedName{
@@ -163,6 +165,7 @@ func TestCephFilesystemMirrorController(t *testing.T) {
ServiceAccount: "foo",
},
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
@@ -205,9 +208,10 @@ func TestCephFilesystemMirrorController(t *testing.T) {
ServiceAccount: "foo",
},
opManagerContext: ctx,
recorder: record.NewFakeRecorder(5),
}
res, err := r.Reconcile(ctx, req)
assert.Error(t, err)
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
@@ -229,8 +233,7 @@ func TestCephFilesystemMirrorController(t *testing.T) {
}
res, err := r.Reconcile(ctx, req)
assert.Error(t, err)
assert.EqualError(t, err, "waiting for ceph monitors upgrade to finish. current version: 16.0.0-0 pacific. expected version: 15.0.0-0 octopus. will reconcile again in 1m0s")
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
+27 -24
View File
@@ -39,6 +39,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -82,6 +83,7 @@ type ReconcileCephNFS struct {
clusterInfo *cephclient.ClusterInfo
opManagerContext context.Context
opConfig opcontroller.OperatorConfig
recorder record.EventRecorder
}
// Add creates a new cephNFS Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -98,6 +100,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
context: context,
opManagerContext: opManagerContext,
opConfig: opConfig,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -135,31 +138,28 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileCephNFS) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile %v", err)
}
reconcileResponse, cephNFS, err := r.reconcile(request)
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephNFS, reconcileResponse, err)
}
func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephNFS, error) {
// Fetch the cephNFS instance
cephNFS := &cephv1.CephNFS{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephNFS)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("cephNFS resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephNFS, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get cephNFS")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to get cephNFS")
}
// Set a finalizer so we can do cleanup before the object goes away
err = opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, cephNFS)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to add finalizer")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to add finalizer")
}
// The CR was just created, initializing status fields
@@ -180,13 +180,14 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
// Remove finalizer
err := opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephNFS)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephNFS, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephNFS, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephNFS, nil
}
r.cephClusterSpec = &cephCluster.Spec
@@ -194,33 +195,35 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
// Always populate it during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to populate cluster info")
}
// DELETE: the CR was deleted
if !cephNFS.GetDeletionTimestamp().IsZero() {
logger.Infof("deleting ceph nfs %q", cephNFS.Name)
r.recorder.Eventf(cephNFS, v1.EventTypeNormal, string(cephv1.ReconcileStarted), "deleting CephNFS %q", cephNFS.Name)
// Detect running Ceph version
runningCephVersion, err := cephclient.LeastUptodateDaemonVersion(r.context, r.clusterInfo, config.MonType)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.MonType)
return reconcile.Result{}, *cephNFS, errors.Wrapf(err, "failed to retrieve current ceph %q version", config.MonType)
}
r.clusterInfo.CephVersion = runningCephVersion
err = r.removeServersFromDatabase(cephNFS, 0)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to delete filesystem %q. ", cephNFS.Name)
return reconcile.Result{}, *cephNFS, errors.Wrapf(err, "failed to delete filesystem %q. ", cephNFS.Name)
}
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephNFS)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephNFS, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephNFS, nil
}
// Detect desired CephCluster version
@@ -237,9 +240,9 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
if err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephNFS, nil
}
return reconcile.Result{}, errors.Wrap(err, "failed to detect running and desired ceph version")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to detect running and desired ceph version")
}
// If the version of the Ceph monitor differs from the CephCluster CR image version we assume
@@ -247,7 +250,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
// then versions should match. Obviously using the cmd reporter job adds up to the deployment time
if !reflect.DeepEqual(*runningCephVersion, *desiredCephVersion) {
// Upgrade is in progress, let's wait for the mons to be done
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading,
return opcontroller.WaitForRequeueIfCephClusterIsUpgrading, *cephNFS,
opcontroller.ErrorCephUpgradingRequeue(desiredCephVersion, runningCephVersion)
}
r.clusterInfo.CephVersion = *runningCephVersion
@@ -276,13 +279,13 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
// validate the store settings
if err := validateGanesha(r.context, r.clusterInfo, cephNFS); err != nil {
return reconcile.Result{}, errors.Wrapf(err, "invalid ceph nfs %q arguments", cephNFS.Name)
return reconcile.Result{}, *cephNFS, errors.Wrapf(err, "invalid ceph nfs %q arguments", cephNFS.Name)
}
// Check for the existence of the .nfs pool
err = r.configureNFSPool(cephNFS)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to configure nfs pool %q", cephNFS.Spec.RADOS.Pool)
return reconcile.Result{}, *cephNFS, errors.Wrapf(err, "failed to configure nfs pool %q", cephNFS.Spec.RADOS.Pool)
}
// CREATE/UPDATE
@@ -290,7 +293,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
_, err = r.reconcileCreateCephNFS(cephNFS)
if err != nil {
updateStatus(r.client, request.NamespacedName, k8sutil.FailedStatus)
return reconcile.Result{}, errors.Wrap(err, "failed to create ceph nfs deployments")
return reconcile.Result{}, *cephNFS, errors.Wrap(err, "failed to create ceph nfs deployments")
}
// Set Ready status, we are done reconciling
@@ -298,7 +301,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
// Return and do not requeue
logger.Debug("done reconciling ceph nfs")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephNFS, nil
}
+2 -1
View File
@@ -40,6 +40,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -201,7 +202,7 @@ func TestCephNFSController(t *testing.T) {
newReconcile := func(clusterCtx *clusterd.Context, cl client.WithWatch) *ReconcileCephNFS {
// Create a ReconcileCephNFS object with the scheme and fake client.
return &ReconcileCephNFS{client: cl, scheme: testScheme, context: clusterCtx, opManagerContext: ctx}
return &ReconcileCephNFS{client: cl, scheme: testScheme, context: clusterCtx, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
}
// Mock request to simulate Reconcile() being called on an event for a
+20 -19
View File
@@ -29,6 +29,7 @@ import (
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
"github.com/rook/rook/pkg/operator/ceph/reporting"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -77,6 +78,7 @@ type ReconcileObjectRealm struct {
context *clusterd.Context
clusterInfo *cephclient.ClusterInfo
opManagerContext context.Context
recorder record.EventRecorder
}
// Add creates a new CephObjectRealm Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -92,6 +94,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
scheme: mgr.GetScheme(),
context: context,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -118,25 +121,22 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileObjectRealm) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile CephObjectRealm %q. %v", request.NamespacedName.String(), err)
}
reconcileResponse, cephObjectRealm, err := r.reconcile(request)
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephObjectRealm, reconcileResponse, err)
}
func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephObjectRealm, error) {
// Fetch the CephObjectRealm instance
cephObjectRealm := &cephv1.CephObjectRealm{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephObjectRealm)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephObjectRealm %q resource not found. Ignoring since object must be deleted", request.NamespacedName.String())
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectRealm, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get CephObjectRealm")
return reconcile.Result{}, *cephObjectRealm, errors.Wrap(err, "failed to get CephObjectRealm")
}
// The CR was just created, initializing status fields
@@ -150,30 +150,31 @@ func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.R
// This handles the case where the Ceph Cluster is gone and we want to delete that CR
if !cephObjectRealm.GetDeletionTimestamp().IsZero() && !cephClusterExists {
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectRealm, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephObjectRealm, nil
}
// DELETE: the CR was deleted
if !cephObjectRealm.GetDeletionTimestamp().IsZero() {
logger.Debugf("deleting realm CR %q", cephObjectRealm.Name)
r.recorder.Eventf(cephObjectRealm, v1.EventTypeNormal, string(cephv1.ReconcileStarted), "deleting CephObjectRealm %q", cephObjectRealm.Name)
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectRealm, nil
}
// Populate clusterInfo during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
return reconcile.Result{}, *cephObjectRealm, errors.Wrap(err, "failed to populate cluster info")
}
// validate the realm settings
err = validateRealmCR(cephObjectRealm)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return reconcile.Result{}, errors.Wrapf(err, "invalid CephObjectRealm CR %q", cephObjectRealm.Name)
return reconcile.Result{}, *cephObjectRealm, errors.Wrapf(err, "invalid CephObjectRealm CR %q", cephObjectRealm.Name)
}
// Start object reconciliation, updating status for this
@@ -184,17 +185,17 @@ func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.R
logger.Debug("pull section in realm %q spec found", request.NamespacedName)
_, err = r.pullCephRealm(cephObjectRealm)
if err != nil {
return reconcile.Result{}, err
return reconcile.Result{}, *cephObjectRealm, err
}
} else {
_, err = r.createRealmKeys(cephObjectRealm)
if err != nil {
return r.setFailedStatus(request.NamespacedName, "failed to create keys for realm", err)
return r.setFailedStatus(cephObjectRealm, request.NamespacedName, "failed to create keys for realm", err)
}
_, err = r.createCephRealm(cephObjectRealm)
if err != nil {
return r.setFailedStatus(request.NamespacedName, "failed to create ceph realm", err)
return r.setFailedStatus(cephObjectRealm, request.NamespacedName, "failed to create ceph realm", err)
}
}
@@ -203,7 +204,7 @@ func (r *ReconcileObjectRealm) reconcile(request reconcile.Request) (reconcile.R
// Return and do not requeue
logger.Debug("realm %q done reconciling", request.NamespacedName)
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectRealm, nil
}
func (r *ReconcileObjectRealm) pullCephRealm(realm *cephv1.CephObjectRealm) (reconcile.Result, error) {
@@ -321,9 +322,9 @@ func validateRealmCR(u *cephv1.CephObjectRealm) error {
return nil
}
func (r *ReconcileObjectRealm) setFailedStatus(name types.NamespacedName, errMessage string, err error) (reconcile.Result, error) {
func (r *ReconcileObjectRealm) setFailedStatus(cephObjectRealm *cephv1.CephObjectRealm, name types.NamespacedName, errMessage string, err error) (reconcile.Result, cephv1.CephObjectRealm, error) {
r.updateStatus(r.client, name, k8sutil.ReconcileFailedStatus)
return reconcile.Result{}, errors.Wrapf(err, "%s", errMessage)
return reconcile.Result{}, *cephObjectRealm, errors.Wrapf(err, "%s", errMessage)
}
// updateStatus updates an realm with a given status
@@ -36,6 +36,7 @@ import (
"k8s.io/apimachinery/pkg/types"
k8sfake "k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -98,7 +99,7 @@ func TestCephObjectRealmController(t *testing.T) {
// Create a fake client to mock API calls.
cl := fake.NewClientBuilder().WithScheme(r.scheme).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectRealm object with the scheme and fake client.
r = &ReconcileObjectRealm{client: cl, scheme: r.scheme, context: r.context}
r = &ReconcileObjectRealm{client: cl, scheme: r.scheme, context: r.context, recorder: record.NewFakeRecorder(5)}
res, err = r.Reconcile(ctx, req)
assert.NoError(t, err)
assert.True(t, res.Requeue)
@@ -154,7 +155,7 @@ func TestCephObjectRealmController(t *testing.T) {
r.context.Executor = executor
// Create a ReconcileObjectRealm object with the scheme and fake client.
r = &ReconcileObjectRealm{client: cl, scheme: r.scheme, context: r.context}
r = &ReconcileObjectRealm{client: cl, scheme: r.scheme, context: r.context, recorder: record.NewFakeRecorder(5)}
res, err = r.Reconcile(ctx, req)
assert.NoError(t, err)
@@ -256,7 +257,7 @@ func getObjectRealmAndReconcileObjectRealm(t *testing.T) (*ReconcileObjectRealm,
cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectRealm object with the scheme and fake client.
clusterInfo := cephclient.AdminTestClusterInfo("rook")
r := &ReconcileObjectRealm{client: cl, scheme: s, context: c, clusterInfo: clusterInfo}
r := &ReconcileObjectRealm{client: cl, scheme: s, context: c, clusterInfo: clusterInfo, recorder: record.NewFakeRecorder(5)}
return r, objectRealm
}
@@ -284,7 +285,8 @@ func TestReconcileObjectRealm_createRealmKeys(t *testing.T) {
context: &clusterd.Context{
Clientset: k8sfake.NewSimpleClientset(),
},
scheme: scheme,
scheme: scheme,
recorder: record.NewFakeRecorder(5),
}
for _, tName := range []string{"first reconcile", "second reconcile"} {
@@ -322,7 +324,8 @@ func TestReconcileObjectRealm_createRealmKeys(t *testing.T) {
context: &clusterd.Context{
Clientset: k8sfake.NewSimpleClientset(secret),
},
scheme: scheme,
scheme: scheme,
recorder: record.NewFakeRecorder(5),
}
_, err := r.createRealmKeys(realm)
+29 -23
View File
@@ -28,6 +28,7 @@ import (
"github.com/rook/rook/pkg/operator/ceph/reporting"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -43,6 +44,7 @@ import (
"github.com/rook/rook/pkg/operator/ceph/object"
"github.com/rook/rook/pkg/operator/k8sutil"
corev1 "k8s.io/api/core/v1"
v1 "k8s.io/api/core/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
@@ -76,6 +78,7 @@ type ReconcileObjectStoreUser struct {
cephClusterSpec *cephv1.ClusterSpec
clusterInfo *cephclient.ClusterInfo
opManagerContext context.Context
recorder record.EventRecorder
}
// Add creates a new CephObjectStoreUser Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -91,6 +94,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
scheme: mgr.GetScheme(),
context: context,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -126,31 +130,29 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileObjectStoreUser) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile %v", err)
}
reconcileResponse, cephObjectStoreUser, err := r.reconcile(request)
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephObjectStoreUser, reconcileResponse, err)
return reconcileResponse, err
}
func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephObjectStoreUser, error) {
// Fetch the CephObjectStoreUser instance
cephObjectStoreUser := &cephv1.CephObjectStoreUser{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephObjectStoreUser)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephObjectStoreUser resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectStoreUser, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get CephObjectStoreUser")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to get CephObjectStoreUser")
}
// Set a finalizer so we can do cleanup before the object goes away
err = opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, cephObjectStoreUser)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to add finalizer")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to add finalizer")
}
// The CR was just created, initializing status fields
@@ -171,20 +173,20 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephObjectStoreUser)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to remove finalizer")
}
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectStoreUser, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephObjectStoreUser, nil
}
r.cephClusterSpec = &cephCluster.Spec
// Populate clusterInfo during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to populate cluster info")
}
// Validate the object store has been initialized
@@ -194,16 +196,17 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephObjectStoreUser)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephObjectStoreUser, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectStoreUser, nil
}
logger.Debugf("ObjectStore resource not ready in namespace %q, retrying in %q. %v",
request.NamespacedName.Namespace, opcontroller.WaitForRequeueIfCephClusterNotReady.RequeueAfter.String(), err)
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return opcontroller.WaitForRequeueIfCephClusterNotReady, nil
return opcontroller.WaitForRequeueIfCephClusterNotReady, *cephObjectStoreUser, nil
}
// Generate user config
@@ -213,40 +216,43 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
// DELETE: the CR was deleted
if !cephObjectStoreUser.GetDeletionTimestamp().IsZero() {
logger.Debugf("deleting pool %q", cephObjectStoreUser.Name)
r.recorder.Eventf(cephObjectStoreUser, v1.EventTypeNormal, string(cephv1.ReconcileStarted), "deleting CephObjectStoreUser %q", cephObjectStoreUser.Name)
err := r.deleteUser(cephObjectStoreUser)
if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph object user %q", cephObjectStoreUser.Name)
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrapf(err, "failed to delete ceph object user %q", cephObjectStoreUser.Name)
}
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephObjectStoreUser)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to remove finalizer")
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephObjectStoreUser, v1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectStoreUser, nil
}
// validate the user settings
err = r.validateUser(cephObjectStoreUser)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return reconcile.Result{}, errors.Wrapf(err, "invalid pool CR %q spec", cephObjectStoreUser.Name)
return reconcile.Result{}, *cephObjectStoreUser, errors.Wrapf(err, "invalid pool CR %q spec", cephObjectStoreUser.Name)
}
// CREATE/UPDATE CEPH USER
reconcileResponse, err = r.reconcileCephUser(cephObjectStoreUser)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return reconcileResponse, err
return reconcileResponse, *cephObjectStoreUser, err
}
// CREATE/UPDATE KUBERNETES SECRET
reconcileResponse, err = r.reconcileCephUserSecret(cephObjectStoreUser)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return reconcileResponse, err
return reconcileResponse, *cephObjectStoreUser, err
}
// Set Ready status, we are done reconciling
@@ -254,7 +260,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
// Return and do not requeue
logger.Debug("done reconciling")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectStoreUser, nil
}
func (r *ReconcileObjectStoreUser) reconcileCephUser(cephObjectStoreUser *cephv1.CephObjectStoreUser) (reconcile.Result, error) {
@@ -43,6 +43,7 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -150,7 +151,7 @@ func TestCephObjectStoreUserController(t *testing.T) {
// Create a fake client to mock API calls.
cl := fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectStoreUser object with the scheme and fake client.
r := &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx}
r := &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
// Mock request to simulate Reconcile() being called on an event for a
// watched resource .
@@ -185,7 +186,7 @@ func TestCephObjectStoreUserController(t *testing.T) {
// Create a fake client to mock API calls.
cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectStoreUser object with the scheme and fake client.
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx}
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
assert.True(t, res.Requeue)
@@ -233,7 +234,7 @@ func TestCephObjectStoreUserController(t *testing.T) {
c.Executor = executor
// Create a ReconcileObjectStoreUser object with the scheme and fake client.
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx}
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
@@ -265,7 +266,7 @@ func TestCephObjectStoreUserController(t *testing.T) {
// Create a fake client to mock API calls.
cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectStoreUser object with the scheme and fake client.
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx}
r = &ReconcileObjectStoreUser{client: cl, scheme: s, context: c, opManagerContext: ctx, recorder: record.NewFakeRecorder(5)}
err := r.client.Get(context.TODO(), types.NamespacedName{Name: store, Namespace: namespace}, cephObjectStore)
assert.NoError(t, err, cephObjectStore)
+21 -19
View File
@@ -27,7 +27,9 @@ import (
"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
"github.com/rook/rook/pkg/operator/ceph/reporting"
v1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -73,6 +75,7 @@ type ReconcileObjectZone struct {
clusterInfo *cephclient.ClusterInfo
clusterSpec *cephv1.ClusterSpec
opManagerContext context.Context
recorder record.EventRecorder
}
// Add creates a new CephObjectZone Controller and adds it to the Manager. The Manager will set fields on the Controller
@@ -88,6 +91,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
scheme: mgr.GetScheme(),
context: context,
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -114,25 +118,22 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileObjectZone) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile: %v", err)
}
reconcileResponse, cephObjectZone, err := r.reconcile(request)
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephObjectZone, reconcileResponse, err)
}
func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephObjectZone, error) {
// Fetch the CephObjectZone instance
cephObjectZone := &cephv1.CephObjectZone{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephObjectZone)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephObjectZone resource not found. Ignoring since object must be deleted.")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectZone, nil
}
// Error reading the object - requeue the request.
return reconcile.Result{}, errors.Wrap(err, "failed to get CephObjectZone")
return reconcile.Result{}, *cephObjectZone, errors.Wrap(err, "failed to get CephObjectZone")
}
// The CR was just created, initializing status fields
@@ -147,31 +148,32 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re
//
if !cephObjectZone.GetDeletionTimestamp().IsZero() && !cephClusterExists {
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectZone, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephObjectZone, nil
}
r.clusterSpec = &cephCluster.Spec
// DELETE: the CR was deleted
if !cephObjectZone.GetDeletionTimestamp().IsZero() {
logger.Debugf("deleting zone CR %q", cephObjectZone.Name)
r.recorder.Eventf(cephObjectZone, v1.EventTypeNormal, string(cephv1.ReconcileStarted), "deleting CephObjectZOne %q", cephObjectZone.Name)
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectZone, nil
}
// Populate clusterInfo during each reconcile
r.clusterInfo, _, _, err = mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to populate cluster info")
return reconcile.Result{}, *cephObjectZone, errors.Wrap(err, "failed to populate cluster info")
}
// validate the zone settings
err = r.validateZoneCR(cephObjectZone)
if err != nil {
r.updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
return reconcile.Result{}, errors.Wrapf(err, "invalid CephObjectZone CR %q", cephObjectZone.Name)
return reconcile.Result{}, *cephObjectZone, errors.Wrapf(err, "invalid CephObjectZone CR %q", cephObjectZone.Name)
}
// Start object reconciliation, updating status for this
@@ -180,19 +182,19 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re
// Make sure an ObjectZoneGroup is present
realmName, reconcileResponse, err := r.reconcileObjectZoneGroup(cephObjectZone)
if err != nil {
return reconcileResponse, err
return reconcileResponse, *cephObjectZone, err
}
// Make sure zone group has been created in Ceph Cluster
reconcileResponse, err = r.reconcileCephZoneGroup(cephObjectZone, realmName)
if err != nil {
return reconcileResponse, err
return reconcileResponse, *cephObjectZone, err
}
// Create Ceph Zone
_, err = r.createCephZone(cephObjectZone, realmName)
if err != nil {
return r.setFailedStatus(request.NamespacedName, "failed to create ceph zone", err)
return r.setFailedStatus(cephObjectZone, request.NamespacedName, "failed to create ceph zone", err)
}
// Set Ready status, we are done reconciling
@@ -200,7 +202,7 @@ func (r *ReconcileObjectZone) reconcile(request reconcile.Request) (reconcile.Re
// Return and do not requeue
logger.Debug("zone done reconciling")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephObjectZone, nil
}
func (r *ReconcileObjectZone) createCephZone(zone *cephv1.CephObjectZone, realmName string) (reconcile.Result, error) {
@@ -339,9 +341,9 @@ func (r *ReconcileObjectZone) validateZoneCR(z *cephv1.CephObjectZone) error {
return nil
}
func (r *ReconcileObjectZone) setFailedStatus(name types.NamespacedName, errMessage string, err error) (reconcile.Result, error) {
func (r *ReconcileObjectZone) setFailedStatus(cephObjectZone *cephv1.CephObjectZone, name types.NamespacedName, errMessage string, err error) (reconcile.Result, cephv1.CephObjectZone, error) {
r.updateStatus(r.client, name, k8sutil.ReconcileFailedStatus)
return reconcile.Result{}, errors.Wrapf(err, "%s", errMessage)
return reconcile.Result{}, *cephObjectZone, errors.Wrapf(err, "%s", errMessage)
}
// updateStatus updates an zone with a given status
@@ -37,6 +37,7 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -184,7 +185,7 @@ func TestCephObjectZoneController(t *testing.T) {
// Create a ReconcileObjectZone object with the scheme and fake client.
clusterInfo := cephclient.AdminTestClusterInfo("rook")
r := &ReconcileObjectZone{client: cl, scheme: s, context: c, clusterInfo: clusterInfo}
r := &ReconcileObjectZone{client: cl, scheme: s, context: c, clusterInfo: clusterInfo, recorder: record.NewFakeRecorder(5)}
// Mock request to simulate Reconcile() being called on an event for a
// watched resource .
@@ -226,7 +227,7 @@ func TestCephObjectZoneController(t *testing.T) {
cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
// Create a ReconcileObjectZone object with the scheme and fake client.
r = &ReconcileObjectZone{client: cl, scheme: r.scheme, context: r.context}
r = &ReconcileObjectZone{client: cl, scheme: r.scheme, context: r.context, recorder: record.NewFakeRecorder(5)}
res, err = r.Reconcile(ctx, req)
assert.NoError(t, err)
assert.True(t, res.Requeue)
@@ -277,10 +278,10 @@ func TestCephObjectZoneController(t *testing.T) {
}
r.context.Executor = executor
r = &ReconcileObjectZone{client: cl, scheme: r.scheme, context: r.context}
r = &ReconcileObjectZone{client: cl, scheme: r.scheme, context: r.context, recorder: record.NewFakeRecorder(5)}
res, err = r.Reconcile(ctx, req)
assert.Error(t, err)
assert.NoError(t, err)
assert.True(t, res.Requeue)
//
@@ -342,7 +343,7 @@ func TestCephObjectZoneController(t *testing.T) {
cl = fake.NewClientBuilder().WithScheme(s).WithRuntimeObjects(object...).Build()
r = &ReconcileObjectZone{client: cl, scheme: s, context: c, clusterInfo: clusterInfo}
r = &ReconcileObjectZone{client: cl, scheme: s, context: c, clusterInfo: clusterInfo, recorder: record.NewFakeRecorder(5)}
err = r.client.Get(context.TODO(), types.NamespacedName{Name: zonegroup, Namespace: namespace}, objectZoneGroup)
assert.NoError(t, err, objectZoneGroup)
+32 -26
View File
@@ -33,12 +33,14 @@ import (
"github.com/rook/rook/pkg/operator/ceph/config"
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
"github.com/rook/rook/pkg/operator/ceph/csi/peermap"
"github.com/rook/rook/pkg/operator/ceph/reporting"
"github.com/rook/rook/pkg/operator/k8sutil"
corev1 "k8s.io/api/core/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -72,6 +74,7 @@ type ReconcileCephBlockPool struct {
clusterInfo *cephclient.ClusterInfo
blockPoolContexts map[string]*blockPoolHealth
opManagerContext context.Context
recorder record.EventRecorder
}
type blockPoolHealth struct {
@@ -94,6 +97,7 @@ func newReconciler(mgr manager.Manager, context *clusterd.Context, opManagerCont
context: context,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: opManagerContext,
recorder: mgr.GetEventRecorderFor("rook-" + controllerName),
}
}
@@ -133,15 +137,12 @@ func add(opManagerContext context.Context, mgr manager.Manager, r reconcile.Reco
// Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
func (r *ReconcileCephBlockPool) Reconcile(context context.Context, request reconcile.Request) (reconcile.Result, error) {
// workaround because the rook logging mechanism is not compatible with the controller-runtime logging interface
reconcileResponse, err := r.reconcile(request)
if err != nil {
logger.Errorf("failed to reconcile. %v", err)
}
reconcileResponse, cephBlockPool, err := r.reconcile(request)
return reconcileResponse, err
return reporting.ReportReconcileResult(logger, r.recorder, request, &cephBlockPool, reconcileResponse, err)
}
func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile.Result, error) {
func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile.Result, cephv1.CephBlockPool, error) {
// Fetch the CephBlockPool instance
cephBlockPool := &cephv1.CephBlockPool{}
err := r.client.Get(r.opManagerContext, request.NamespacedName, cephBlockPool)
@@ -154,16 +155,16 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
cephBlockPool.Name = request.Name
cephBlockPool.Namespace = request.Namespace
r.cancelMirrorMonitoring(cephBlockPool)
return reconcile.Result{}, nil
return reconcile.Result{}, *cephBlockPool, nil
}
// Error reading the object - requeue the request.
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to get CephBlockPool")
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to get CephBlockPool")
}
// Set a finalizer so we can do cleanup before the object goes away
err = opcontroller.AddFinalizerIfNotPresent(r.opManagerContext, r.client, cephBlockPool)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to add finalizer")
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to add finalizer")
}
// The CR was just created, initializing status fields
@@ -187,19 +188,19 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephBlockPool)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to remove finalizer")
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to remove finalizer")
}
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephBlockPool, nil
}
return reconcileResponse, nil
return reconcileResponse, *cephBlockPool, nil
}
// Populate clusterInfo during each reconcile
clusterInfo, _, _, err := mon.LoadClusterInfo(r.context, r.opManagerContext, request.NamespacedName.Namespace)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to populate cluster info")
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to populate cluster info")
}
r.clusterInfo = clusterInfo
r.clusterInfo.NetworkSpec = cephCluster.Spec.Network
@@ -226,11 +227,13 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// We must remove it first otherwise the checker will panic since the status/info will be nil
r.cancelMirrorMonitoring(cephBlockPool)
r.recorder.Event(cephBlockPool, corev1.EventTypeNormal, string(cephv1.ReconcileStarted), "starting blockpool deletion")
logger.Infof("deleting pool %q", cephBlockPool.Name)
poolSpec := cephBlockPool.Spec.ToNamedPoolSpec()
err := deletePool(r.context, clusterInfo, &poolSpec)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to delete pool %q. ", cephBlockPool.Name)
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(err, "failed to delete pool %q. ", cephBlockPool.Name)
}
// disable RBD stats collection if cephBlockPool was deleted
@@ -241,26 +244,29 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// Remove finalizer
err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephBlockPool)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrap(err, "failed to remove finalizer")
r.recorder.Event(cephBlockPool, corev1.EventTypeWarning, string(cephv1.ReconcileFailed), "failed to remove finalizer")
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrap(err, "failed to remove finalizer")
}
r.recorder.Event(cephBlockPool, corev1.EventTypeNormal, string(cephv1.ReconcileSucceeded), "successfully removed finalizer")
// Return and do not requeue. Successful deletion.
return reconcile.Result{}, nil
return reconcile.Result{}, *cephBlockPool, nil
}
// validate the pool settings
if err := validatePool(r.context, clusterInfo, &cephCluster.Spec, cephBlockPool); err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephBlockPool, nil
}
return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "invalid pool CR %q spec", cephBlockPool.Name)
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(err, "invalid pool CR %q spec", cephBlockPool.Name)
}
// Get CephCluster version
cephVersion, err := opcontroller.GetImageVersion(cephCluster)
if err != nil {
return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to fetch ceph version from cephcluster %q", cephCluster.Name)
return opcontroller.ImmediateRetryResult, *cephBlockPool, errors.Wrapf(err, "failed to fetch ceph version from cephcluster %q", cephCluster.Name)
}
r.clusterInfo.CephVersion = *cephVersion
@@ -269,15 +275,15 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
if err != nil {
if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) {
logger.Info(opcontroller.OperatorNotInitializedMessage)
return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil
return opcontroller.WaitForRequeueIfOperatorNotInitialized, *cephBlockPool, nil
}
updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure, nil)
return reconcileResponse, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName())
return reconcileResponse, *cephBlockPool, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName())
}
// enable/disable RBD stats collection based on cephBlockPool spec
if err := configureRBDStats(r.context, clusterInfo); err != nil {
return reconcile.Result{}, errors.Wrap(err, "failed to enable/disable stats collection for pool(s)")
return reconcile.Result{}, *cephBlockPool, errors.Wrap(err, "failed to enable/disable stats collection for pool(s)")
}
poolSpec := cephBlockPool.Spec.ToNamedPoolSpec()
@@ -289,7 +295,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
reconcileResponse, err = opcontroller.CreateBootstrapPeerSecret(r.context, clusterInfo, cephBlockPool, k8sutil.NewOwnerInfo(cephBlockPool, r.scheme))
if err != nil {
updateStatus(r.client, request.NamespacedName, cephv1.ConditionFailure, nil)
return reconcileResponse, errors.Wrapf(err, "failed to create rbd-mirror bootstrap peer for pool %q.", cephBlockPool.GetName())
return reconcileResponse, *cephBlockPool, errors.Wrapf(err, "failed to create rbd-mirror bootstrap peer for pool %q.", cephBlockPool.GetName())
}
// Check if rbd-mirror CR and daemons are running
@@ -309,13 +315,13 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
logger.Debug("reconciling ceph bootstrap peers import")
reconcileResponse, err = r.reconcileAddBoostrapPeer(cephBlockPool, request.NamespacedName)
if err != nil {
return reconcileResponse, errors.Wrap(err, "failed to add ceph rbd mirror peer")
return reconcileResponse, *cephBlockPool, errors.Wrap(err, "failed to add ceph rbd mirror peer")
}
// ReconcilePoolIDMap updates the `rook-ceph-csi-mapping-config` with local and peer cluster pool ID map
err = peermap.ReconcilePoolIDMap(r.opManagerContext, r.context, r.clusterInfo, cephBlockPool)
if err != nil {
return reconcileResponse, errors.Wrapf(err, "failed to update pool ID mapping config for the pool %q", cephBlockPool.Name)
return reconcileResponse, *cephBlockPool, errors.Wrapf(err, "failed to update pool ID mapping config for the pool %q", cephBlockPool.Name)
}
// Set Ready status, we are done reconciling
@@ -336,7 +342,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
// Return and do not requeue
logger.Debug("done reconciling")
return reconcile.Result{}, nil
return reconcile.Result{}, *cephBlockPool, nil
}
func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.ClusterInfo, cephCluster *cephv1.ClusterSpec, cephBlockPool *cephv1.CephBlockPool) (reconcile.Result, error) {
+9 -2
View File
@@ -40,6 +40,7 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/tools/record"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
@@ -245,6 +246,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
// Mock request to simulate Reconcile() being called on an event for a
@@ -300,6 +302,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
res, err := r.Reconcile(ctx, req)
@@ -357,6 +360,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
res, err := r.Reconcile(ctx, req)
assert.NoError(t, err)
@@ -372,7 +376,7 @@ func TestCephBlockPoolController(t *testing.T) {
err := r.client.Update(context.TODO(), pool)
assert.NoError(t, err)
res, err := r.Reconcile(ctx, req)
assert.Error(t, err)
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
@@ -428,6 +432,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
pool.Spec.Mirroring.Mode = "image"
@@ -469,6 +474,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
pool.Spec.Mirroring.Peers.SecretNames = []string{peerSecretName}
@@ -476,7 +482,7 @@ func TestCephBlockPoolController(t *testing.T) {
assert.NoError(t, err)
res, err := r.Reconcile(ctx, req)
// assert reconcile failure because peer token secert was not created
assert.Error(t, err)
assert.NoError(t, err)
assert.True(t, res.Requeue)
})
@@ -506,6 +512,7 @@ func TestCephBlockPoolController(t *testing.T) {
context: c,
blockPoolContexts: make(map[string]*blockPoolHealth),
opManagerContext: context.TODO(),
recorder: record.NewFakeRecorder(5),
}
pool.Spec.Mirroring.Enabled = false
pool.Spec.Mirroring.Mode = "image"