Merge pull request #5219 from travisn/status-reconcile-loop
Skip reconcile events where the spec is not updated
This commit is contained in:
6 files changed
+162
-133
No files matched your search
@@ -60,14 +60,14 @@ func WatchControllerPredicate() predicate.Funcs {
|
||||
objNew := e.ObjectNew.(*cephv1.CephObjectStore)
|
||||
logger.Debug("update event from the parent object CephObjectStore")
|
||||
diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
|
||||
if diff != "" ||
|
||||
objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() ||
|
||||
objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
if diff != "" || objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() {
|
||||
// Checking if diff is not empty so we don't print it when the CR gets deleted
|
||||
if diff != "" {
|
||||
logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff)
|
||||
}
|
||||
return true
|
||||
} else if objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
|
||||
}
|
||||
// Handling upgrades
|
||||
isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
|
||||
@@ -79,42 +79,42 @@ func WatchControllerPredicate() predicate.Funcs {
|
||||
objNew := e.ObjectNew.(*cephv1.CephObjectStoreUser)
|
||||
logger.Debug("update event from the parent object CephObjectStoreUser")
|
||||
diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
|
||||
if diff != "" ||
|
||||
objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() ||
|
||||
objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
if diff != "" || objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() {
|
||||
// Checking if diff is not empty so we don't print it when the CR gets deleted
|
||||
if diff != "" {
|
||||
logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff)
|
||||
}
|
||||
return true
|
||||
} else if objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
|
||||
}
|
||||
|
||||
case *cephv1.CephBlockPool:
|
||||
objNew := e.ObjectNew.(*cephv1.CephBlockPool)
|
||||
logger.Debug("update event from the parent object CephBlockPool")
|
||||
diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
|
||||
if diff != "" ||
|
||||
objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() ||
|
||||
objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
if diff != "" || objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() {
|
||||
// Checking if diff is not empty so we don't print it when the CR gets deleted
|
||||
if diff != "" {
|
||||
logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff)
|
||||
}
|
||||
return true
|
||||
} else if objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
|
||||
}
|
||||
|
||||
case *cephv1.CephFilesystem:
|
||||
objNew := e.ObjectNew.(*cephv1.CephFilesystem)
|
||||
logger.Debug("update event from the parent object CephFilesystem")
|
||||
diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
|
||||
if diff != "" ||
|
||||
objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() ||
|
||||
objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
if diff != "" || objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() {
|
||||
// Checking if diff is not empty so we don't print it when the CR gets deleted
|
||||
if diff != "" {
|
||||
logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff)
|
||||
}
|
||||
return true
|
||||
} else if objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
|
||||
}
|
||||
// Handling upgrades
|
||||
isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
|
||||
@@ -126,14 +126,14 @@ func WatchControllerPredicate() predicate.Funcs {
|
||||
objNew := e.ObjectNew.(*cephv1.CephNFS)
|
||||
logger.Debug("update event from the parent object CephNFS")
|
||||
diff := cmp.Diff(objOld.Spec, objNew.Spec, resourceQtyComparer)
|
||||
if diff != "" ||
|
||||
objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() ||
|
||||
objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
if diff != "" || objOld.GetDeletionTimestamp() != objNew.GetDeletionTimestamp() {
|
||||
// Checking if diff is not empty so we don't print it when the CR gets deleted
|
||||
if diff != "" {
|
||||
logger.Infof("CR has changed for %q. diff=%s", objNew.Name, diff)
|
||||
}
|
||||
return true
|
||||
} else if objOld.GetGeneration() != objNew.GetGeneration() {
|
||||
logger.Debugf("skipping resource %q update with unchanged spec", objNew.Name)
|
||||
}
|
||||
// Handling upgrades
|
||||
isUpgrade := isUpgrade(objOld.GetLabels(), objNew.GetLabels())
|
||||
|
||||
@@ -35,6 +35,7 @@ import (
|
||||
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/types"
|
||||
|
||||
appsv1 "k8s.io/api/apps/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
@@ -150,12 +151,7 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil
|
||||
|
||||
// The CR was just created, initializing status fields
|
||||
if cephFilesystem.Status == nil {
|
||||
cephFilesystem.Status = &cephv1.Status{}
|
||||
cephFilesystem.Status.Phase = k8sutil.Created
|
||||
err := opcontroller.UpdateStatus(r.client, cephFilesystem)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.Created)
|
||||
}
|
||||
|
||||
// Make sure a CephCluster is present otherwise do nothing
|
||||
@@ -230,20 +226,12 @@ func (r *ReconcileCephFilesystem) reconcile(request reconcile.Request) (reconcil
|
||||
logger.Debug("reconciling ceph filesystem store deployments")
|
||||
reconcileResponse, err = r.reconcileCreateFilesystem(cephFilesystem)
|
||||
if err != nil {
|
||||
cephFilesystem.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephFilesystem)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return reconcileResponse, err
|
||||
}
|
||||
|
||||
// Set Ready status, we are done reconciling
|
||||
cephFilesystem.Status.Phase = k8sutil.ReadyStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephFilesystem)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReadyStatus)
|
||||
|
||||
// Return and do not requeue
|
||||
logger.Debug("done reconciling")
|
||||
@@ -274,3 +262,28 @@ func (r *ReconcileCephFilesystem) reconcileCreateFilesystem(cephFilesystem *ceph
|
||||
|
||||
return reconcile.Result{}, nil
|
||||
}
|
||||
|
||||
// updateStatus updates an object with a given status
|
||||
func updateStatus(client client.Client, name types.NamespacedName, status string) {
|
||||
fs := &cephv1.CephFilesystem{}
|
||||
err := client.Get(context.TODO(), name, fs)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephFilesystem resource not found. Ignoring since object must be deleted.")
|
||||
return
|
||||
}
|
||||
logger.Warningf("failed to retrieve filesystem %q to update status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
|
||||
if fs.Status == nil {
|
||||
fs.Status = &cephv1.Status{}
|
||||
}
|
||||
|
||||
fs.Status.Phase = status
|
||||
if err := opcontroller.UpdateStatus(client, fs); err != nil {
|
||||
logger.Errorf("failed to set filesystem %q status to %q. %v", fs.Name, status, err)
|
||||
return
|
||||
}
|
||||
logger.Debugf("filesystem %q status updated to %q", name, status)
|
||||
}
|
||||
@@ -36,6 +36,7 @@ import (
|
||||
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"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller"
|
||||
"sigs.k8s.io/controller-runtime/pkg/handler"
|
||||
@@ -149,12 +150,7 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
|
||||
|
||||
// The CR was just created, initializing status fields
|
||||
if cephNFS.Status == nil {
|
||||
cephNFS.Status = &cephv1.Status{}
|
||||
cephNFS.Status.Phase = k8sutil.Created
|
||||
err := opcontroller.UpdateStatus(r.client, cephNFS)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.Created)
|
||||
}
|
||||
|
||||
// Make sure a CephCluster is present otherwise do nothing
|
||||
@@ -228,15 +224,12 @@ func (r *ReconcileCephNFS) reconcile(request reconcile.Request) (reconcile.Resul
|
||||
logger.Debug("reconciling ceph nfs deployments")
|
||||
reconcileResponse, err = r.reconcileCreateCephNFS(cephNFS)
|
||||
if err != nil {
|
||||
return r.setFailedStatus(cephNFS, "failed to create ceph nfs deployments", err)
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.FailedStatus)
|
||||
return reconcile.Result{}, errors.Wrapf(err, "failed to create ceph nfs deployments")
|
||||
}
|
||||
|
||||
// Set Ready status, we are done reconciling
|
||||
cephNFS.Status.Phase = k8sutil.ReadyStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephNFS)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReadyStatus)
|
||||
|
||||
// Return and do not requeue
|
||||
logger.Debug("done reconciling ceph nfs")
|
||||
@@ -289,12 +282,25 @@ func (r *ReconcileCephNFS) reconcileCreateCephNFS(cephNFS *cephv1.CephNFS) (reco
|
||||
return reconcile.Result{}, nil
|
||||
}
|
||||
|
||||
func (r *ReconcileCephNFS) setFailedStatus(cephNFS *cephv1.CephNFS, errMessage string, err error) (reconcile.Result, error) {
|
||||
cephNFS.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephNFS)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
// updateStatus updates an object with a given status
|
||||
func updateStatus(client client.Client, name types.NamespacedName, status string) {
|
||||
nfs := &cephv1.CephNFS{}
|
||||
err := client.Get(context.TODO(), name, nfs)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephNFS resource not found. Ignoring since object must be deleted.")
|
||||
return
|
||||
}
|
||||
logger.Warningf("failed to retrieve nfs %q to update status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
if nfs.Status == nil {
|
||||
nfs.Status = &cephv1.Status{}
|
||||
}
|
||||
|
||||
return reconcile.Result{}, errors.Wrapf(err, "%s", errMessage)
|
||||
nfs.Status.Phase = status
|
||||
if err := opcontroller.UpdateStatus(client, nfs); err != nil {
|
||||
logger.Errorf("failed to set nfs %q status to %q. %v", nfs.Name, status, err)
|
||||
}
|
||||
logger.Debugf("nfs %q status updated to %q", name, status)
|
||||
}
|
||||
@@ -37,6 +37,7 @@ import (
|
||||
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"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller"
|
||||
"sigs.k8s.io/controller-runtime/pkg/handler"
|
||||
@@ -150,12 +151,7 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
|
||||
|
||||
// The CR was just created, initializing status fields
|
||||
if cephObjectStore.Status == nil {
|
||||
cephObjectStore.Status = &cephv1.Status{}
|
||||
cephObjectStore.Status.Phase = k8sutil.Created
|
||||
err := opcontroller.UpdateStatus(r.client, cephObjectStore)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.Created)
|
||||
}
|
||||
|
||||
// Make sure a CephCluster is present otherwise do nothing
|
||||
@@ -229,24 +225,20 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci
|
||||
|
||||
// CREATE/UPDATE
|
||||
logger.Info("reconciling object store deployments")
|
||||
reconcileResponse, err = r.reconcileCreateObjectStore(cephObjectStore)
|
||||
reconcileResponse, err = r.reconcileCreateObjectStore(cephObjectStore, request.NamespacedName)
|
||||
if err != nil {
|
||||
return r.setFailedStatus(cephObjectStore, "failed to create object store deployments", err)
|
||||
return r.setFailedStatus(request.NamespacedName, "failed to create object store deployments", err)
|
||||
}
|
||||
|
||||
// Set Ready status, we are done reconciling
|
||||
cephObjectStore.Status.Phase = k8sutil.ReadyStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephObjectStore)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReadyStatus)
|
||||
|
||||
// Return and do not requeue
|
||||
logger.Debug("done reconciling")
|
||||
return reconcile.Result{}, nil
|
||||
}
|
||||
|
||||
func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *cephv1.CephObjectStore) (reconcile.Result, error) {
|
||||
func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *cephv1.CephObjectStore, name types.NamespacedName) (reconcile.Result, error) {
|
||||
if r.cephClusterSpec.External.Enable {
|
||||
_, err := opcontroller.ValidateCephVersionsBetweenLocalAndExternalClusters(r.context, cephObjectStore.Namespace, r.clusterInfo.CephVersion)
|
||||
if err != nil {
|
||||
@@ -272,7 +264,7 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c
|
||||
logger.Debug("reconciling object store service")
|
||||
serviceIP, err := cfg.reconcileService(cephObjectStore)
|
||||
if err != nil {
|
||||
return r.setFailedStatus(cephObjectStore, "failed to reconcile service", err)
|
||||
return r.setFailedStatus(name, "failed to reconcile service", err)
|
||||
}
|
||||
|
||||
objContext := NewContext(r.context, cephObjectStore.Name, cephObjectStore.Namespace)
|
||||
@@ -281,14 +273,14 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c
|
||||
logger.Info("reconciling object store pools")
|
||||
err = createPools(objContext, cephObjectStore.Spec)
|
||||
if err != nil {
|
||||
return r.setFailedStatus(cephObjectStore, "failed to create object pools", err)
|
||||
return r.setFailedStatus(name, "failed to create object pools", err)
|
||||
}
|
||||
|
||||
// RECONCILE REALM
|
||||
logger.Info("reconciling object store realms")
|
||||
err = reconcileRealm(objContext, serviceIP, cephObjectStore.Spec.Gateway.Port)
|
||||
if err != nil {
|
||||
return r.setFailedStatus(cephObjectStore, "failed to create object store realm", err)
|
||||
return r.setFailedStatus(name, "failed to create object store realm", err)
|
||||
}
|
||||
|
||||
err = cfg.createOrUpdateStore()
|
||||
@@ -299,12 +291,30 @@ func (r *ReconcileCephObjectStore) reconcileCreateObjectStore(cephObjectStore *c
|
||||
return reconcile.Result{}, nil
|
||||
}
|
||||
|
||||
func (r *ReconcileCephObjectStore) setFailedStatus(cephObjectStore *cephv1.CephObjectStore, errMessage string, err error) (reconcile.Result, error) {
|
||||
cephObjectStore.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephObjectStore)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
}
|
||||
|
||||
func (r *ReconcileCephObjectStore) setFailedStatus(name types.NamespacedName, errMessage string, err error) (reconcile.Result, error) {
|
||||
updateStatus(r.client, name, k8sutil.ReconcileFailedStatus)
|
||||
return reconcile.Result{}, errors.Wrapf(err, "%s", errMessage)
|
||||
}
|
||||
|
||||
// updateStatus updates an object with a given status
|
||||
func updateStatus(client client.Client, name types.NamespacedName, status string) {
|
||||
objectStore := &cephv1.CephObjectStore{}
|
||||
if err := client.Get(context.TODO(), name, objectStore); err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephObjectStore resource not found. Ignoring since object must be deleted.")
|
||||
return
|
||||
}
|
||||
logger.Warningf("failed to retrieve object store %q to update status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
if objectStore.Status == nil {
|
||||
objectStore.Status = &cephv1.Status{}
|
||||
}
|
||||
|
||||
objectStore.Status.Phase = status
|
||||
if err := opcontroller.UpdateStatus(client, objectStore); err != nil {
|
||||
logger.Errorf("failed to set object store %q status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
logger.Debugf("object store %q status updated to %q", name, status)
|
||||
}
|
||||
@@ -24,6 +24,7 @@ import (
|
||||
|
||||
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
|
||||
"sigs.k8s.io/controller-runtime/pkg/handler"
|
||||
@@ -142,12 +143,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
|
||||
|
||||
// The CR was just created, initializing status fields
|
||||
if cephObjectStoreUser.Status == nil {
|
||||
cephObjectStoreUser.Status = &cephv1.Status{}
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.Created
|
||||
err := opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.Created)
|
||||
}
|
||||
|
||||
// Make sure a CephCluster is present otherwise do nothing
|
||||
@@ -192,11 +188,7 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
|
||||
return reconcile.Result{}, nil
|
||||
}
|
||||
logger.Debugf("ObjectStore resource not ready in namespace %q, retrying in %q. %v", request.NamespacedName.Namespace, opcontroller.WaitForRequeueIfCephClusterNotReadyAfter.String(), err)
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
err := opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return opcontroller.WaitForRequeueIfCephClusterNotReady, nil
|
||||
}
|
||||
// Set the object store context
|
||||
@@ -227,49 +219,26 @@ func (r *ReconcileObjectStoreUser) reconcile(request reconcile.Request) (reconci
|
||||
// validate the user settings
|
||||
err = ValidateUser(cephObjectStoreUser)
|
||||
if err != nil {
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
err := opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return reconcile.Result{}, errors.Wrapf(err, "invalid pool CR %q spec", cephObjectStoreUser.Name)
|
||||
}
|
||||
|
||||
// Start object reconciliation, updating status for this
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReconcilingStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
|
||||
// CREATE/UPDATE CEPH USER
|
||||
reconcileResponse, err = r.reconcileCephUser(cephObjectStoreUser)
|
||||
if err != nil {
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return reconcileResponse, err
|
||||
}
|
||||
|
||||
// CREATE/UPDATE KUBERNETES SECRET
|
||||
reconcileResponse, err = r.reconcileCephUserSecret(cephObjectStoreUser)
|
||||
if err != nil {
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return reconcileResponse, err
|
||||
}
|
||||
|
||||
// Set Ready status, we are done reconciling
|
||||
cephObjectStoreUser.Status.Phase = k8sutil.ReadyStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephObjectStoreUser)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReadyStatus)
|
||||
|
||||
// Return and do not requeue
|
||||
logger.Debug("done reconciling")
|
||||
@@ -473,3 +442,26 @@ func ValidateUser(u *cephv1.CephObjectStoreUser) error {
|
||||
func labelsForRgw(name string) map[string]string {
|
||||
return map[string]string{"rgw": name, k8sutil.AppAttr: appName}
|
||||
}
|
||||
|
||||
// updateStatus updates an object with a given status
|
||||
func updateStatus(client client.Client, name types.NamespacedName, status string) {
|
||||
user := &cephv1.CephObjectStoreUser{}
|
||||
if err := client.Get(context.TODO(), name, user); err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephObjectStoreUser resource not found. Ignoring since object must be deleted.")
|
||||
return
|
||||
}
|
||||
logger.Warningf("failed to retrieve object store user %q to update status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
if user.Status == nil {
|
||||
user.Status = &cephv1.Status{}
|
||||
}
|
||||
|
||||
user.Status.Phase = status
|
||||
if err := opcontroller.UpdateStatus(client, user); err != nil {
|
||||
logger.Errorf("failed to set object store user %q status to %q. %v", name, status, err)
|
||||
return
|
||||
}
|
||||
logger.Debugf("object store user %q status updated to %q", name, status)
|
||||
}
|
||||
@@ -33,6 +33,7 @@ import (
|
||||
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"
|
||||
"sigs.k8s.io/controller-runtime/pkg/client"
|
||||
"sigs.k8s.io/controller-runtime/pkg/controller"
|
||||
"sigs.k8s.io/controller-runtime/pkg/handler"
|
||||
@@ -131,12 +132,7 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
|
||||
|
||||
// The CR was just created, initializing status fields
|
||||
if cephBlockPool.Status == nil {
|
||||
cephBlockPool.Status = &cephv1.Status{}
|
||||
cephBlockPool.Status.Phase = k8sutil.Created
|
||||
err := opcontroller.UpdateStatus(r.client, cephBlockPool)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.Created)
|
||||
}
|
||||
|
||||
// Make sure a CephCluster is present otherwise do nothing
|
||||
@@ -190,30 +186,17 @@ func (r *ReconcileCephBlockPool) reconcile(request reconcile.Request) (reconcile
|
||||
return reconcile.Result{}, errors.Wrapf(err, "invalid pool CR %q spec", cephBlockPool.Name)
|
||||
}
|
||||
|
||||
// Start object reconciliation, updating status for this
|
||||
cephBlockPool.Status.Phase = k8sutil.ReconcilingStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephBlockPool)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcilingStatus)
|
||||
|
||||
// CREATE/UPDATE
|
||||
reconcileResponse, err = r.reconcileCreatePool(cephBlockPool)
|
||||
if err != nil {
|
||||
cephBlockPool.Status.Phase = k8sutil.ReconcileFailedStatus
|
||||
errStatus := opcontroller.UpdateStatus(r.client, cephBlockPool)
|
||||
if errStatus != nil {
|
||||
logger.Errorf("failed to set status. %v", errStatus)
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReconcileFailedStatus)
|
||||
return reconcileResponse, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName())
|
||||
}
|
||||
|
||||
// Set Ready status, we are done reconciling
|
||||
cephBlockPool.Status.Phase = k8sutil.ReadyStatus
|
||||
err = opcontroller.UpdateStatus(r.client, cephBlockPool)
|
||||
if err != nil {
|
||||
return reconcile.Result{}, errors.Wrap(err, "failed to set status")
|
||||
}
|
||||
updateStatus(r.client, request.NamespacedName, k8sutil.ReadyStatus)
|
||||
|
||||
// Return and do not requeue
|
||||
logger.Debug("done reconciling")
|
||||
@@ -260,3 +243,28 @@ func deletePool(context *clusterd.Context, p *cephv1.CephBlockPool) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateStatus updates a pool CR with the given status
|
||||
func updateStatus(client client.Client, poolName types.NamespacedName, status string) {
|
||||
pool := &cephv1.CephBlockPool{}
|
||||
err := client.Get(context.TODO(), poolName, pool)
|
||||
if err != nil {
|
||||
if kerrors.IsNotFound(err) {
|
||||
logger.Debug("CephBlockPool resource not found. Ignoring since object must be deleted.")
|
||||
return
|
||||
}
|
||||
logger.Warningf("failed to retrieve pool %q to update status to %q. %v", poolName, status, err)
|
||||
return
|
||||
}
|
||||
|
||||
if pool.Status == nil {
|
||||
pool.Status = &cephv1.Status{}
|
||||
}
|
||||
|
||||
pool.Status.Phase = status
|
||||
if err := opcontroller.UpdateStatus(client, pool); err != nil {
|
||||
logger.Warningf("failed to set pool %q status to %q. %v", pool.Name, status, err)
|
||||
return
|
||||
}
|
||||
logger.Debugf("pool %q status updated to %q", poolName, status)
|
||||
}
|
||||
Reference in new issue
Block a user