From 354c51e440a5078418d82f247a9c2054bb5a1cef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 16 Jul 2020 10:27:12 +0200 Subject: [PATCH 1/9] ceph: do not monitor osd on external mode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If the cluster is external, we must not run the osd healthcheck go routine. Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/monitoring.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index fe1ad04ee..a2723a22c 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/pkg/operator/ceph/cluster/monitoring.go @@ -117,9 +117,11 @@ func (c *ClusterController) startMonitoringCheck(cluster *cluster, clusterInfo * go healthChecker.Check(cluster.monitoringChannels[daemon].stopChan) case "osd": - c.osdChecker = osd.NewOSDHealthMonitor(c.context, clusterInfo, cluster.Spec.RemoveOSDsIfOutAndSafeToRemove, cluster.Spec.HealthCheck) - logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) - go c.osdChecker.Start(cluster.monitoringChannels[daemon].stopChan) + if !cluster.Spec.External.Enable { + c.osdChecker = osd.NewOSDHealthMonitor(c.context, clusterInfo, cluster.Spec.RemoveOSDsIfOutAndSafeToRemove, cluster.Spec.HealthCheck) + logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) + go c.osdChecker.Start(cluster.monitoringChannels[daemon].stopChan) + } case "status": cephChecker := newCephStatusChecker(c.context, clusterInfo, cluster.Spec.HealthCheck) From 28c415636f527405517ea155c12e83eb8da26dfc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 16 Jul 2020 10:31:39 +0200 Subject: [PATCH 2/9] ceph: remove unnecessary comment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This comment is not needed. Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/monitoring.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index a2723a22c..106238b60 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/pkg/operator/ceph/cluster/monitoring.go @@ -55,10 +55,6 @@ func (c *ClusterController) configureCephMonitoring(cluster *cluster, clusterInf } } } else { - // If the mon does not exist in the map, this is a first deployment or an operator restart - // So we check the desired state from the CR and run it if necessary - // - // If the mon monitoring is enabled if !isDisabled { cluster.monitoringChannels[daemon] = &clusterHealth{ stopChan: make(chan struct{}), From 89f815bf9b26b1f008dfa2298b8042cc3f167a4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 16 Jul 2020 13:43:17 +0200 Subject: [PATCH 3/9] ceph: start rgw healthcheck immediately MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When we run the goroutine, we should run start it immediately instead of waiting for the configured interval to start. With that we can get the Status Phase reported much faster in the CRD. Signed-off-by: Sébastien Han --- pkg/operator/ceph/object/health.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/pkg/operator/ceph/object/health.go b/pkg/operator/ceph/object/health.go index 300e427e5..921d5013a 100644 --- a/pkg/operator/ceph/object/health.go +++ b/pkg/operator/ceph/object/health.go @@ -77,6 +77,12 @@ func newBucketChecker(context *clusterd.Context, objContext *Context, serviceIP, // checkObjectStore periodically checks the health of the cluster func (c *bucketChecker) checkObjectStore(stopCh chan struct{}) { + // check the object store health immediately before starting the loop + err := c.checkObjectStoreHealth() + if err != nil { + updateStatusBucket(c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) + logger.Debugf("failed to check rgw health for object store %q. %v", c.namespacedName.Name, err) + } for { select { @@ -88,11 +94,11 @@ func (c *bucketChecker) checkObjectStore(stopCh chan struct{}) { return case <-time.After(c.interval): - logger.Debug("checking rgw health of object store %q", c.namespacedName.Name) + logger.Debugf("checking rgw health of object store %q", c.namespacedName.Name) err := c.checkObjectStoreHealth() if err != nil { updateStatusBucket(c.client, c.namespacedName, cephv1.ConditionFailure, err.Error()) - logger.Warningf("failed to check rgw health for object store %q. %v", c.namespacedName.Name, err) + logger.Debugf("failed to check rgw health for object store %q. %v", c.namespacedName.Name, err) } } } From 8729a206f359cf94fa8d69c7d2c4e682e506d3e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 16 Jul 2020 13:45:43 +0200 Subject: [PATCH 4/9] ceph: let the healthcheck set the status phase MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instead of setting the Phase of the CR once the reconcile is done, let's actually set in from the healthcheck so it is more accurate. Closes: https://github.com/rook/rook/issues/5249 Signed-off-by: Sébastien Han --- pkg/apis/ceph.rook.io/v1/types.go | 1 - pkg/operator/ceph/cluster/cephstatus.go | 46 ++++++++++++-- pkg/operator/ceph/cluster/cephstatus_test.go | 60 +++++++++++++++++-- pkg/operator/ceph/cluster/cluster_external.go | 3 - pkg/operator/ceph/cluster/monitoring.go | 2 +- pkg/operator/ceph/object/controller.go | 6 +- pkg/operator/ceph/object/controller_test.go | 2 +- pkg/operator/ceph/object/health.go | 6 +- pkg/operator/ceph/object/status.go | 1 + tests/integration/ceph_base_object_test.go | 2 +- 10 files changed, 107 insertions(+), 22 deletions(-) diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index 3ace76e1b..c808e7ca9 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/pkg/apis/ceph.rook.io/v1/types.go @@ -230,7 +230,6 @@ const ( ConditionFailure ConditionType = "Failure" ConditionUpgrading ConditionType = "Upgrading" ConditionDeleting ConditionType = "Deleting" - ConditionHealthy ConditionType = "Healthy" // DefaultFailureDomain for PoolSpec DefaultFailureDomain = "host" ) diff --git a/pkg/operator/ceph/cluster/cephstatus.go b/pkg/operator/ceph/cluster/cephstatus.go index 584ccb9a5..16e33766b 100644 --- a/pkg/operator/ceph/cluster/cephstatus.go +++ b/pkg/operator/ceph/cluster/cephstatus.go @@ -26,8 +26,10 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" "github.com/rook/rook/pkg/clusterd" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" + "github.com/rook/rook/pkg/operator/ceph/config" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" cephver "github.com/rook/rook/pkg/operator/ceph/version" + v1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" "sigs.k8s.io/controller-runtime/pkg/client" ) @@ -43,21 +45,23 @@ type cephStatusChecker struct { clusterInfo *cephclient.ClusterInfo interval time.Duration client client.Client + isExternal bool } // newCephStatusChecker creates a new HealthChecker object -func newCephStatusChecker(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, healthCheck cephv1.CephClusterHealthCheckSpec) *cephStatusChecker { +func newCephStatusChecker(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo, clusterSpec *cephv1.ClusterSpec) *cephStatusChecker { c := &cephStatusChecker{ context: context, clusterInfo: clusterInfo, interval: defaultStatusCheckInterval, client: context.Client, + isExternal: clusterSpec.External.Enable, } // allow overriding the check interval with an env var on the operator // Keep the existing behavior var checkInterval string - checkIntervalCRSetting := healthCheck.DaemonHealth.Status.Interval + checkIntervalCRSetting := clusterSpec.HealthCheck.DaemonHealth.Status.Interval checkIntervalEnv := os.Getenv("ROOK_CEPH_STATUS_CHECK_INTERVAL") if checkIntervalEnv != "" { checkInterval = checkIntervalEnv @@ -106,20 +110,22 @@ func (c *cephStatusChecker) checkStatus() { status, err = cephclient.StatusWithUser(c.context, c.clusterInfo) if err != nil { logger.Errorf("failed to get ceph status. %v", err) - if err := c.updateCephStatus(cephStatusOnError(err.Error())); err != nil { + condition, reason, message := c.conditionMessageReason(cephv1.ConditionFailure) + if err := c.updateCephStatus(cephStatusOnError(err.Error()), condition, reason, message); err != nil { logger.Errorf("failed to query cluster status in namespace %q. %v", c.clusterInfo.Namespace, err) } return } logger.Debugf("cluster status: %+v", status) - if err := c.updateCephStatus(&status); err != nil { + condition, reason, message := c.conditionMessageReason(cephv1.ConditionReady) + if err := c.updateCephStatus(&status, condition, reason, message); err != nil { logger.Errorf("failed to query cluster status in namespace %q. %v", c.clusterInfo.Namespace, err) } } // updateStatus updates an object with a given status -func (c *cephStatusChecker) updateCephStatus(status *cephclient.CephStatus) error { +func (c *cephStatusChecker) updateCephStatus(status *cephclient.CephStatus, condition cephv1.ConditionType, reason, message string) error { clusterName := c.clusterInfo.NamespacedName() cephCluster := &cephv1.CephCluster{} err := c.client.Get(context.TODO(), clusterName, cephCluster) @@ -131,12 +137,17 @@ func (c *cephStatusChecker) updateCephStatus(status *cephclient.CephStatus) erro return errors.Wrapf(err, "failed to retrieve ceph cluster %q in namespace %q to update status to %+v", clusterName.Name, clusterName.Namespace, status) } + // Update with Ceph Status cephCluster.Status.CephStatus = toCustomResourceStatus(cephCluster.Status, status) + cephCluster.Status.Phase = condition if err := opcontroller.UpdateStatus(c.client, cephCluster); err != nil { return errors.Wrapf(err, "failed to update cluster %q status", clusterName.Namespace) } - logger.Debugf("ceph cluster %q status updated to %+v", clusterName.Namespace, status) + // Update condition + config.ConditionExport(c.context, c.clusterInfo.NamespacedName(), condition, v1.ConditionTrue, reason, message) + + logger.Debugf("ceph cluster %q status and condition updated to %+v, %v, %s, %s", clusterName.Namespace, status, v1.ConditionTrue, reason, message) return nil } @@ -211,3 +222,26 @@ func cephStatusOnError(errorMessage string) *cephclient.CephStatus { }, } } + +func (c *cephStatusChecker) conditionMessageReason(condition cephv1.ConditionType) (cephv1.ConditionType, string, string) { + var reason, message string + + switch condition { + case cephv1.ConditionFailure: + reason = "ClusterFailure" + message = "Failed to configure ceph cluster" + if c.isExternal { + message = "Failed to configure external ceph cluster" + } + case cephv1.ConditionReady: + reason = "ClusterCreated" + message = "Cluster created successfully" + if c.isExternal { + condition = cephv1.ConditionConnected + reason = "ClusterConnected" + message = "Cluster connected successfully" + } + } + + return condition, reason, message +} diff --git a/pkg/operator/ceph/cluster/cephstatus_test.go b/pkg/operator/ceph/cluster/cephstatus_test.go index f42e7c64d..ff3d72873 100644 --- a/pkg/operator/ceph/cluster/cephstatus_test.go +++ b/pkg/operator/ceph/cluster/cephstatus_test.go @@ -27,6 +27,7 @@ import ( "github.com/rook/rook/pkg/daemon/ceph/client" cephclient "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/stretchr/testify/assert" + controllerclient "sigs.k8s.io/controller-runtime/pkg/client" ) func TestCephStatus(t *testing.T) { @@ -100,21 +101,72 @@ func TestNewCephStatusChecker(t *testing.T) { type args struct { context *clusterd.Context - healthCheck cephv1.CephClusterHealthCheckSpec + clusterInfo *cephclient.ClusterInfo + clusterSpec *cephv1.ClusterSpec } tests := []struct { name string args args want *cephStatusChecker }{ - {"default-interval", args{c, cephv1.CephClusterHealthCheckSpec{}}, &cephStatusChecker{c, clusterInfo, defaultStatusCheckInterval, c.Client}}, - {"default-interval", args{c, cephv1.CephClusterHealthCheckSpec{DaemonHealth: cephv1.DaemonHealthSpec{Status: cephv1.HealthCheckSpec{Interval: "10s"}}}}, &cephStatusChecker{c, clusterInfo, time10s, c.Client}}, + {"default-interval", args{c, clusterInfo, &cephv1.ClusterSpec{}}, &cephStatusChecker{c, clusterInfo, defaultStatusCheckInterval, c.Client, false}}, + {"10s-interval", args{c, clusterInfo, &cephv1.ClusterSpec{HealthCheck: cephv1.CephClusterHealthCheckSpec{DaemonHealth: cephv1.DaemonHealthSpec{Status: cephv1.HealthCheckSpec{Interval: "10s"}}}}}, &cephStatusChecker{c, clusterInfo, time10s, c.Client, false}}, + {"10s-interval-external", args{c, clusterInfo, &cephv1.ClusterSpec{External: cephv1.ExternalSpec{Enable: true}, HealthCheck: cephv1.CephClusterHealthCheckSpec{DaemonHealth: cephv1.DaemonHealthSpec{Status: cephv1.HealthCheckSpec{Interval: "10s"}}}}}, &cephStatusChecker{c, clusterInfo, time10s, c.Client, true}}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := newCephStatusChecker(tt.args.context, clusterInfo, tt.args.healthCheck); !reflect.DeepEqual(got, tt.want) { + if got := newCephStatusChecker(tt.args.context, tt.args.clusterInfo, tt.args.clusterSpec); !reflect.DeepEqual(got, tt.want) { t.Errorf("newCephStatusChecker() = %v, want %v", got, tt.want) } }) } } + +func Test_cephStatusChecker_conditionMessageReason(t *testing.T) { + c := &clusterd.Context{} + clusterInfo := client.AdminClusterInfo("ns") + type fields struct { + context *clusterd.Context + clusterInfo *cephclient.ClusterInfo + interval time.Duration + client controllerclient.Client + isExternal bool + } + type args struct { + condition cephv1.ConditionType + } + tests := []struct { + name string + fields fields + args args + want cephv1.ConditionType + want1 string + want2 string + }{ + {"failure-converged", fields{c, clusterInfo, defaultStatusCheckInterval, c.Client, false}, args{cephv1.ConditionFailure}, cephv1.ConditionFailure, "ClusterFailure", "Failed to configure ceph cluster"}, + {"failure-external", fields{c, clusterInfo, defaultStatusCheckInterval, c.Client, true}, args{cephv1.ConditionFailure}, cephv1.ConditionFailure, "ClusterFailure", "Failed to configure external ceph cluster"}, + {"success-converged", fields{c, clusterInfo, defaultStatusCheckInterval, c.Client, false}, args{cephv1.ConditionReady}, cephv1.ConditionReady, "ClusterCreated", "Cluster created successfully"}, + {"success-external", fields{c, clusterInfo, defaultStatusCheckInterval, c.Client, true}, args{cephv1.ConditionReady}, cephv1.ConditionConnected, "ClusterConnected", "Cluster connected successfully"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + c := &cephStatusChecker{ + context: tt.fields.context, + clusterInfo: tt.fields.clusterInfo, + interval: tt.fields.interval, + client: tt.fields.client, + isExternal: tt.fields.isExternal, + } + got, got1, got2 := c.conditionMessageReason(tt.args.condition) + if !reflect.DeepEqual(got, tt.want) { + t.Errorf("cephStatusChecker.conditionMessageReason() got = %v, want %v", got, tt.want) + } + if got1 != tt.want1 { + t.Errorf("cephStatusChecker.conditionMessageReason() got1 = %v, want %v", got1, tt.want1) + } + if got2 != tt.want2 { + t.Errorf("cephStatusChecker.conditionMessageReason() got2 = %v, want %v", got2, tt.want2) + } + }) + } +} diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go index 31bfea95e..0313f5c56 100644 --- a/pkg/operator/ceph/cluster/cluster_external.go +++ b/pkg/operator/ceph/cluster/cluster_external.go @@ -139,9 +139,6 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error } } - // Everything went well so let's update the CR's status to "connected" - config.ConditionExport(c.context, c.namespacedName, cephv1.ConditionConnected, v1.ConditionTrue, "ClusterConnected", "Cluster connected successfully") - // Mark initialization has done cluster.initCompleted = true diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index 106238b60..1dbe6da3a 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/pkg/operator/ceph/cluster/monitoring.go @@ -120,7 +120,7 @@ func (c *ClusterController) startMonitoringCheck(cluster *cluster, clusterInfo * } case "status": - cephChecker := newCephStatusChecker(c.context, clusterInfo, cluster.Spec.HealthCheck) + cephChecker := newCephStatusChecker(c.context, clusterInfo, cluster.Spec) logger.Infof("enabling ceph %s monitoring goroutine for cluster %q", daemon, cluster.Namespace) go cephChecker.checkCephStatus(cluster.monitoringChannels[daemon].stopChan) } diff --git a/pkg/operator/ceph/object/controller.go b/pkg/operator/ceph/object/controller.go index da47ca1b3..8c21991b4 100644 --- a/pkg/operator/ceph/object/controller.go +++ b/pkg/operator/ceph/object/controller.go @@ -279,8 +279,8 @@ func (r *ReconcileCephObjectStore) reconcile(request reconcile.Request) (reconci return r.setFailedStatus(request.NamespacedName, "failed to create object store deployments", err) } - // Set Ready status, we are done reconciling - updateStatus(r.client, request.NamespacedName, cephv1.ConditionReady, buildStatusInfo(cephObjectStore)) + // Set Progressing status, we are done reconciling, the health check go routine will update the status + updateStatus(r.client, request.NamespacedName, cephv1.ConditionProgressing, buildStatusInfo(cephObjectStore)) // Return and do not requeue logger.Debug("done reconciling") @@ -479,7 +479,7 @@ func (r *ReconcileCephObjectStore) startMonitoring(objectstore *cephv1.CephObjec port = strconv.Itoa(int(objectstore.Spec.Gateway.SecurePort)) } - rgwChecker := newBucketChecker(r.context, objContext, serviceIP, port, r.client, namespacedName, &objectstore.Spec.HealthCheck) + rgwChecker := newBucketChecker(r.context, objContext, serviceIP, port, r.client, namespacedName, &objectstore.Spec.HealthCheck, r.cephClusterSpec.External.Enable) logger.Info("starting rgw healthcheck") go rgwChecker.checkObjectStore(r.objectStoreChannels[objectstore.Name].stopChan) } diff --git a/pkg/operator/ceph/object/controller_test.go b/pkg/operator/ceph/object/controller_test.go index 0c2a3bba5..bc50a46ad 100644 --- a/pkg/operator/ceph/object/controller_test.go +++ b/pkg/operator/ceph/object/controller_test.go @@ -328,7 +328,7 @@ func TestCephObjectStoreController(t *testing.T) { assert.False(t, res.Requeue) err = r.client.Get(context.TODO(), req.NamespacedName, objectStore) assert.NoError(t, err) - assert.Equal(t, cephv1.ConditionReady, objectStore.Status.Phase, objectStore) + assert.Equal(t, cephv1.ConditionProgressing, objectStore.Status.Phase, objectStore) assert.NotEmpty(t, objectStore.Status.Info["endpoint"], objectStore) assert.Equal(t, "http://rook-ceph-rgw-my-store.rook-ceph:80", objectStore.Status.Info["endpoint"], objectStore) logger.Info("PHASE 3 DONE") diff --git a/pkg/operator/ceph/object/health.go b/pkg/operator/ceph/object/health.go index 921d5013a..16fcb9401 100644 --- a/pkg/operator/ceph/object/health.go +++ b/pkg/operator/ceph/object/health.go @@ -48,10 +48,11 @@ type bucketChecker struct { client client.Client namespacedName types.NamespacedName healthCheckSpec *cephv1.BucketHealthCheckSpec + isExternal bool } // newbucketChecker creates a new HealthChecker object -func newBucketChecker(context *clusterd.Context, objContext *Context, serviceIP, port string, client client.Client, namespacedName types.NamespacedName, healthCheckSpec *cephv1.BucketHealthCheckSpec) *bucketChecker { +func newBucketChecker(context *clusterd.Context, objContext *Context, serviceIP, port string, client client.Client, namespacedName types.NamespacedName, healthCheckSpec *cephv1.BucketHealthCheckSpec, isExternal bool) *bucketChecker { c := &bucketChecker{ context: context, objContext: objContext, @@ -61,6 +62,7 @@ func newBucketChecker(context *clusterd.Context, objContext *Context, serviceIP, namespacedName: namespacedName, client: client, healthCheckSpec: healthCheckSpec, + isExternal: isExternal, } // allow overriding the check interval @@ -169,7 +171,7 @@ func (c *bucketChecker) checkObjectStoreHealth() error { logger.Debug("successfully checked object store endpoint for object store %q", c.namespacedName.Name) // Update the EndpointStatus in the CR to reflect the healthyness - updateStatusBucket(c.client, c.namespacedName, cephv1.ConditionHealthy, "") + updateStatusBucket(c.client, c.namespacedName, cephv1.ConditionConnected, "") return nil } diff --git a/pkg/operator/ceph/object/status.go b/pkg/operator/ceph/object/status.go index 97656e67b..9d77e6f5a 100644 --- a/pkg/operator/ceph/object/status.go +++ b/pkg/operator/ceph/object/status.go @@ -71,6 +71,7 @@ func updateStatusBucket(client client.Client, name types.NamespacedName, phase c } objectStore.Status.BucketStatus = toCustomResourceStatus(objectStore.Status.BucketStatus, details, phase) + objectStore.Status.Phase = phase if err := opcontroller.UpdateStatus(client, objectStore); err != nil { logger.Errorf("failed to set object store %q status to %v. %v", name, phase, err) return diff --git a/tests/integration/ceph_base_object_test.go b/tests/integration/ceph_base_object_test.go index 3a1ed75c6..6fe84e74e 100644 --- a/tests/integration/ceph_base_object_test.go +++ b/tests/integration/ceph_base_object_test.go @@ -86,7 +86,7 @@ func runObjectE2ETest(helper *clients.TestClient, k8sh *utils.K8sHelper, s suite // Check object store status objectStore, err := k8sh.RookClientset.CephV1().CephObjectStores(namespace).Get(storeName, metav1.GetOptions{}) assert.Nil(s.T(), err) - assert.Equal(s.T(), cephv1.ConditionHealthy, objectStore.Status.BucketStatus.Health) + assert.Equal(s.T(), cephv1.ConditionConnected, objectStore.Status.BucketStatus.Health) // Info field has the endpoint in it assert.NotEmpty(s.T(), objectStore.Status.Info) assert.NotEmpty(s.T(), objectStore.Status.Info["endpoint"]) From 39931b1edfde0433f3764f9512ccab872dc7da1e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Thu, 16 Jul 2020 15:19:32 +0200 Subject: [PATCH 5/9] ceph: fix conditions not being applied MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Somehow, the client.Update was not updating the cluster object, using client.Status().Update actually fixed it. Signed-off-by: Sébastien Han --- cluster/charts/rook-ceph/templates/resources.yaml | 2 ++ pkg/operator/ceph/config/conditions.go | 2 +- tests/framework/installer/ceph_manifests.go | 2 ++ 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/cluster/charts/rook-ceph/templates/resources.yaml b/cluster/charts/rook-ceph/templates/resources.yaml index fabf6ce5c..4fe541511 100644 --- a/cluster/charts/rook-ceph/templates/resources.yaml +++ b/cluster/charts/rook-ceph/templates/resources.yaml @@ -204,6 +204,8 @@ spec: type: string description: Ceph Health JSONPath: .status.ceph.health + subresources: + status: {} --- apiVersion: apiextensions.k8s.io/v1beta1 kind: CustomResourceDefinition diff --git a/pkg/operator/ceph/config/conditions.go b/pkg/operator/ceph/config/conditions.go index 3ea519730..844bffcd8 100644 --- a/pkg/operator/ceph/config/conditions.go +++ b/pkg/operator/ceph/config/conditions.go @@ -89,7 +89,7 @@ func setCondition(c *clusterd.Context, namespaceName types.NamespacedName, newCo logger.Infof("CephCluster %q status: %q. %q", namespaceName.Namespace, cluster.Status.Phase, cluster.Status.Message) } - err = c.Client.Update(context.TODO(), cluster) + err = c.Client.Status().Update(context.TODO(), cluster) if err != nil { logger.Errorf("failed to update cluster condition to %+v. %v", newCondition, err) } diff --git a/tests/framework/installer/ceph_manifests.go b/tests/framework/installer/ceph_manifests.go index a8e610204..f1fc55bb6 100644 --- a/tests/framework/installer/ceph_manifests.go +++ b/tests/framework/installer/ceph_manifests.go @@ -287,6 +287,8 @@ spec: type: string description: Ceph Health JSONPath: .status.ceph.health + subresources: + status: {} --- apiVersion: apiextensions.k8s.io/v1beta1 kind: CustomResourceDefinition From 9b9141ee84f115fa3190bf2407cc5e14575b7edb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Fri, 17 Jul 2020 12:12:49 +0200 Subject: [PATCH 6/9] ceph: stop monitoring health on cluster deletion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stop the go routines on cluster deletion. Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/controller.go | 21 +++++++++++++++++---- pkg/operator/ceph/cluster/monitoring.go | 7 +++++-- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 2b81f8a69..988294710 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -368,10 +368,23 @@ func (c *ClusterController) requestClusterDelete(cluster *cephv1.CephCluster) (r logger.Infof("delete event for cluster %q in namespace %q", cluster.Name, cluster.Namespace) - if cluster, ok := c.clusterMap[cluster.Namespace]; ok && !cluster.closedStopCh { - // close the goroutines watching the health of the cluster (mons, osds, ceph status, etc) - close(cluster.stopCh) - cluster.closedStopCh = true + if cluster, ok := c.clusterMap[cluster.Namespace]; ok { + // if not already stopped, stop clientcontroller and bucketController + if !cluster.closedStopCh { + close(cluster.stopCh) + cluster.closedStopCh = true + } + + // close the goroutines watching the health of the cluster (mons, osds, ceph status) + var isDisabled bool + for _, daemon := range monitorDaemonList { + isDisabled = isMonitoringDisabled(daemon, cluster.Spec) + if _, ok := cluster.monitoringChannels[daemon]; ok { + if !isDisabled { + close(cluster.monitoringChannels[daemon].stopChan) + } + } + } } err := c.checkIfVolumesExist(cluster) diff --git a/pkg/operator/ceph/cluster/monitoring.go b/pkg/operator/ceph/cluster/monitoring.go index 1dbe6da3a..de09c6475 100644 --- a/pkg/operator/ceph/cluster/monitoring.go +++ b/pkg/operator/ceph/cluster/monitoring.go @@ -26,11 +26,14 @@ import ( "github.com/rook/rook/pkg/operator/ceph/object/bucket" ) +var ( + monitorDaemonList = []string{"mon", "osd", "status"} +) + func (c *ClusterController) configureCephMonitoring(cluster *cluster, clusterInfo *cephclient.ClusterInfo) { var isDisabled bool - daemons := []string{"mon", "osd", "status"} - for _, daemon := range daemons { + for _, daemon := range monitorDaemonList { // Is the monitoring enabled for that daemon? isDisabled = isMonitoringDisabled(daemon, cluster.Spec) if health, ok := cluster.monitoringChannels[daemon]; ok { From 17e9f1a89019260dc6962ea31a6c66110b4b6cae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Fri, 17 Jul 2020 14:04:29 +0200 Subject: [PATCH 7/9] ceph: remove unexisting function MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Leftover from a previous patch, this function does not exist anymore. Signed-off-by: Sébastien Han --- cluster/examples/kubernetes/ceph/import-external-cluster.sh | 1 - 1 file changed, 1 deletion(-) diff --git a/cluster/examples/kubernetes/ceph/import-external-cluster.sh b/cluster/examples/kubernetes/ceph/import-external-cluster.sh index feceec2f2..8d65cfbf5 100644 --- a/cluster/examples/kubernetes/ceph/import-external-cluster.sh +++ b/cluster/examples/kubernetes/ceph/import-external-cluster.sh @@ -137,7 +137,6 @@ function importCsiCephFSProvisionerSecret() { checkEnvVars importSecret importConfigMap -importCheckerSecret importCsiRBDNodeSecret importCsiRBDProvisionerSecret importCsiCephFSNodeSecret From 5a621a15919837275486d1b73e924351b56ac562 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Mon, 20 Jul 2020 14:37:19 +0200 Subject: [PATCH 8/9] ceph: use infof MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We need infof since we pass a variable. Signed-off-by: Sébastien Han --- pkg/apis/ceph.rook.io/v1/validate_clusters.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/apis/ceph.rook.io/v1/validate_clusters.go b/pkg/apis/ceph.rook.io/v1/validate_clusters.go index 4cb20d0f8..5de327c0b 100644 --- a/pkg/apis/ceph.rook.io/v1/validate_clusters.go +++ b/pkg/apis/ceph.rook.io/v1/validate_clusters.go @@ -47,7 +47,7 @@ func (c *CephCluster) ValidateCreate() error { } func (c *CephCluster) ValidateUpdate(old runtime.Object) error { - logger.Info("validate update cephcluster %q", c.ObjectMeta.Name) + logger.Infof("validate update cephcluster %q", c.ObjectMeta.Name) if err := validateCommon(*c); err != nil { return err From e517ac96aac7f67ea10c422afe8ea2ba57e5fc3a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Mon, 20 Jul 2020 14:38:22 +0200 Subject: [PATCH 9/9] ceph: fail if the pool fails to be deleted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Let's assert if the pool is still present. Signed-off-by: Sébastien Han --- tests/framework/clients/block.go | 8 +++++++- tests/integration/ceph_flex_test.go | 6 ++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/framework/clients/block.go b/tests/framework/clients/block.go index 96a97a4b7..adb64776e 100644 --- a/tests/framework/clients/block.go +++ b/tests/framework/clients/block.go @@ -22,6 +22,7 @@ import ( "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/rook/rook/tests/framework/installer" "github.com/rook/rook/tests/framework/utils" + "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -85,7 +86,12 @@ func (b *BlockOperation) DeletePVC(namespace, claimName string) error { func (b *BlockOperation) DeleteStorageClass(storageClassName string) error { logger.Infof("deleting storage class %q", storageClassName) - return b.k8sClient.Clientset.StorageV1().StorageClasses().Delete(storageClassName, &metav1.DeleteOptions{}) + err := b.k8sClient.Clientset.StorageV1().StorageClasses().Delete(storageClassName, &metav1.DeleteOptions{}) + if err != nil && !errors.IsNotFound(err) { + return fmt.Errorf("failed to delete storage class %q. %v", storageClassName, err) + } + + return nil } // BlockDelete Function to delete a Block using Rook diff --git a/tests/integration/ceph_flex_test.go b/tests/integration/ceph_flex_test.go index a785d1e34..6540685dd 100644 --- a/tests/integration/ceph_flex_test.go +++ b/tests/integration/ceph_flex_test.go @@ -173,8 +173,10 @@ func (s *CephFlexDriverSuite) statefulSetDataCleanup(poolName, storageClassName, // Delete all PVCs s.kh.DeletePvcWithLabel(defaultNamespace, statefulSetName) // Delete storageclass and pool - s.testClient.PoolClient.DeletePool(s.testClient.BlockClient, s.clusterInfo, poolName) - s.testClient.BlockClient.DeleteStorageClass(storageClassName) + err := s.testClient.PoolClient.DeletePool(s.testClient.BlockClient, s.clusterInfo, poolName) + require.Nil(s.T(), err) + err = s.testClient.BlockClient.DeleteStorageClass(storageClassName) + require.Nil(s.T(), err) } func (s *CephFlexDriverSuite) setupPVCs() {