Merge pull request #2938 from travisn/save-cluster-health

Add ceph health info to the cephcluster custom resource

PendingReleaseNotes.md

@@ -13,6 +13,7 @@
 ### Ceph
 
 - Ceph Nautilus (`v14`) is now supported by Rook.
+- Ceph status is surfaced in the CephCluster CR and periodically updated by the operator (default is 60s). The interval can be configured with the `ROOK_CEPH_STATUS_CHECK_INTERVAL` env var.
 - A `CephNFS` CRD will start NFS daemon(s) for exporting CephFS volumes or RGW buckets. See the [NFS documentation](Documentation/ceph-nfs-crd.md).
 - Selinux labeling for mounts can now be toggled with the [ROOK_ENABLE_SELINUX_RELABELING](https://github.com/rook/rook/issues/2417) environment variable.
 - Recursive chown for mounts can now be toggled with the [ROOK_ENABLE_FSGROUP](https://github.com/rook/rook/issues/2254) environment variable.

cluster/charts/rook-ceph/templates/deployment.yaml

@@ -89,6 +89,10 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: metadata.namespace
+{{- if .Values.cephStatusCheckInterval }}
+        - name: ROOK_CEPH_STATUS_CHECK_INTERVAL
+          value: {{ .Values.cephStatusCheckInterval }}
+{{- end }}
 {{- if .Values.mon }}
 {{- if .Values.mon.healthCheckInterval }}
         - name: ROOK_MON_HEALTHCHECK_INTERVAL

cluster/charts/rook-ceph/values.yaml.tmpl

@@ -29,6 +29,9 @@ nodeSelector:
 # Tolerations for the rook-ceph-operator to allow it to run on nodes with particular taints
 tolerations: []
 
+# Interval at which to get the ceph status and update the cluster custom resource status
+cephStatusCheckInterval: "60s"
+
 mon:
   healthCheckInterval: "45s"
   monOutTimeout: "300s"

cluster/examples/kubernetes/ceph/common.yaml

@@ -97,6 +97,10 @@ spec:
       type: string
       description: Current State
       JSONPath: .status.state
+    - name: Health
+      type: string
+      description: Ceph Health
+      JSONPath: .status.ceph.health
 ---
 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition

cluster/examples/kubernetes/ceph/operator.yaml

@@ -86,6 +86,9 @@ spec:
         # The logging level for the operator: INFO | DEBUG
         - name: ROOK_LOG_LEVEL
           value: "INFO"
+        # The interval to check the health of the ceph cluster and update the status in the custom resource.
+        - name: ROOK_CEPH_STATUS_CHECK_INTERVAL
+          value: "60s"
         # The interval to check if every mon is in the quorum.
         - name: ROOK_MON_HEALTHCHECK_INTERVAL
           value: "45s"

pkg/apis/ceph.rook.io/v1/types.go

@@ -98,8 +98,22 @@ type DashboardSpec struct {
 }
 
 type ClusterStatus struct {
-	State   ClusterState `json:"state,omitempty"`
-	Message string       `json:"message,omitempty"`
+	State      ClusterState `json:"state,omitempty"`
+	Message    string       `json:"message,omitempty"`
+	CephStatus *CephStatus  `json:"ceph,omitempty"`
+}
+
+type CephStatus struct {
+	Health         string                       `json:"health,omitempty"`
+	Details        map[string]CephHealthMessage `json:"details,omitempty"`
+	LastChecked    string                       `json:"lastChecked,omitempty"`
+	LastChanged    string                       `json:"lastChanged,omitempty"`
+	PreviousHealth string                       `json:"previousHealth,omitempty"`
+}
+
+type CephHealthMessage struct {
+	Severity string `json:"severity"`
+	Message  string `json:"message"`
 }
 
 type ClusterState string

pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go

@@ -91,7 +91,7 @@ func (in *CephCluster) DeepCopyInto(out *CephCluster) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
-	out.Status = in.Status
+	in.Status.DeepCopyInto(&out.Status)
 	return
 }
 
@@ -206,6 +206,22 @@ func (in *CephFilesystemList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CephHealthMessage) DeepCopyInto(out *CephHealthMessage) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CephHealthMessage.
+func (in *CephHealthMessage) DeepCopy() *CephHealthMessage {
+	if in == nil {
+		return nil
+	}
+	out := new(CephHealthMessage)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CephNFS) DeepCopyInto(out *CephNFS) {
 	*out = *in
@@ -386,6 +402,29 @@ func (in *CephObjectStoreUserList) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *CephStatus) DeepCopyInto(out *CephStatus) {
+	*out = *in
+	if in.Details != nil {
+		in, out := &in.Details, &out.Details
+		*out = make(map[string]CephHealthMessage, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new CephStatus.
+func (in *CephStatus) DeepCopy() *CephStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(CephStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *CephVersionSpec) DeepCopyInto(out *CephVersionSpec) {
 	*out = *in
@@ -441,6 +480,11 @@ func (in *ClusterSpec) DeepCopy() *ClusterSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ClusterStatus) DeepCopyInto(out *ClusterStatus) {
 	*out = *in
+	if in.CephStatus != nil {
+		in, out := &in.CephStatus, &out.CephStatus
+		*out = new(CephStatus)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 

pkg/daemon/ceph/client/command.go

@@ -88,6 +88,11 @@ func ExecuteCephCommand(context *clusterd.Context, clusterName string, args []st
 	return executeCephCommandWithOutputFile(context, clusterName, false, args)
 }
 
+// ExecuteCephCommandDebug executes the 'ceph' command with debug output
+func ExecuteCephCommandDebug(context *clusterd.Context, clusterName string, debug bool, args []string) ([]byte, error) {
+	return executeCephCommandWithOutputFile(context, clusterName, debug, args)
+}
+
 // ExecuteCephCommandPlain executes the 'ceph' command and returns stdout in PLAIN format instead of JSON
 func ExecuteCephCommandPlain(context *clusterd.Context, clusterName string, args []string) ([]byte, error) {
 	command, args := FinalizeCephCommandArgs(CephTool, args, context.ConfigDir, clusterName)

pkg/daemon/ceph/client/mon.go

@@ -67,16 +67,6 @@ func GetMonStatus(context *clusterd.Context, clusterName string, debug bool) (Mo
 	return resp, nil
 }
 
-// MonStats is a subset of fields on the response from the mon command "status".  These fields
-// are focused on monitor stats.
-type MonStats struct {
-	Health struct {
-		Status string                  `json:"status"`
-		Checks map[string]CheckMessage `json:"checks"`
-	} `json:"health"`
-	Quorum []int `json:"quorum"`
-}
-
 type MonTimeStatus struct {
 	Skew   map[string]MonTimeSkewStatus `json:"time_skew_status"`
 	Checks struct {
@@ -92,23 +82,6 @@ type MonTimeSkewStatus struct {
 	Health  string      `json:"health"`
 }
 
-func GetMonStats(context *clusterd.Context, clusterName string) (*MonStats, error) {
-	// note this is another call to the mon command "status", but we'll be marshalling it into
-	// a type with a different subset of fields, scoped to monitor stats
-	args := []string{"status"}
-	buf, err := ExecuteCephCommand(context, clusterName, args)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get status: %+v", err)
-	}
-
-	var monStats MonStats
-	if err := json.Unmarshal(buf, &monStats); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal status response: %+v", err)
-	}
-
-	return &monStats, nil
-}
-
 func GetMonTimeStatus(context *clusterd.Context, clusterName string) (*MonTimeStatus, error) {
 	args := []string{"time-sync-status"}
 	buf, err := ExecuteCephCommand(context, clusterName, args)

pkg/daemon/ceph/client/status.go

@@ -123,9 +123,9 @@ type PgStateEntry struct {
 	Count     int    `json:"count"`
 }
 
-func Status(context *clusterd.Context, clusterName string) (CephStatus, error) {
+func Status(context *clusterd.Context, clusterName string, debug bool) (CephStatus, error) {
 	args := []string{"status"}
-	buf, err := ExecuteCephCommand(context, clusterName, args)
+	buf, err := ExecuteCephCommandDebug(context, clusterName, debug, args)
 	if err != nil {
 		return CephStatus{}, fmt.Errorf("failed to get status: %+v", err)
 	}
@@ -141,7 +141,7 @@ func Status(context *clusterd.Context, clusterName string) (CephStatus, error) {
 // IsClusterClean returns a value indicating if the cluster is fully clean yet (i.e., all placement
 // groups are in the active+clean state).
 func IsClusterClean(context *clusterd.Context, clusterName string) error {
-	status, err := Status(context, clusterName)
+	status, err := Status(context, clusterName, false)
 	if err != nil {
 		return err
 	}

pkg/operator/ceph/cluster/cephstatus.go

@@ -0,0 +1,140 @@
+/*
+Copyright 2019 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package cluster to manage Kubernetes storage.
+package cluster
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/clusterd"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	// defaultStatusCheckInterval is the interval to check the status of the ceph cluster
+	defaultStatusCheckInterval = 60 * time.Second
+)
+
+// cephStatusChecker aggregates the mon/cluster info needed to check the health of the monitors
+type cephStatusChecker struct {
+	context      *clusterd.Context
+	namespace    string
+	resourceName string
+	interval     time.Duration
+}
+
+// newCephStatusChecker creates a new HealthChecker object
+func newCephStatusChecker(context *clusterd.Context, namespace, resourceName string) *cephStatusChecker {
+	c := &cephStatusChecker{
+		context:      context,
+		namespace:    namespace,
+		resourceName: resourceName,
+		interval:     defaultStatusCheckInterval,
+	}
+
+	// allow overriding the check interval with an env var on the operator
+	checkInterval := os.Getenv("ROOK_CEPH_STATUS_CHECK_INTERVAL")
+	if checkInterval != "" {
+		if duration, err := time.ParseDuration(checkInterval); err == nil {
+			logger.Infof("ceph status check interval is %s", checkInterval)
+			c.interval = duration
+		}
+	}
+	return c
+}
+
+// checkCephStatus periodically checks the health of the cluster
+func (c *cephStatusChecker) checkCephStatus(stopCh chan struct{}) {
+	// check the status immediately before starting the loop
+	c.checkStatus()
+
+	for {
+		select {
+		case <-stopCh:
+			logger.Infof("Stopping monitoring of ceph status")
+			return
+
+		case <-time.After(c.interval):
+			c.checkStatus()
+		}
+	}
+}
+
+// checkStatus queries the status of ceph health then updates the CR status
+func (c *cephStatusChecker) checkStatus() {
+	logger.Debugf("checking health of cluster")
+	status, err := client.Status(c.context, c.namespace, true)
+	if err != nil {
+		logger.Errorf("failed to get ceph status. %+v", err)
+		return
+	}
+
+	logger.Debugf("Cluster status: %+v", status)
+	if err := c.updateCephStatus(&status); err != nil {
+		logger.Errorf("failed to query cluster status in namespace %s", c.namespace)
+	}
+}
+
+// updateCephStatus detects the latest health status from ceph and updates the CR status
+func (c *cephStatusChecker) updateCephStatus(status *client.CephStatus) error {
+
+	// get the most recent cluster CRD object
+	cluster, err := c.context.RookClientset.CephV1().CephClusters(c.namespace).Get(c.resourceName, metav1.GetOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to get cluster from namespace %s prior to updating its status: %+v", c.namespace, err)
+	}
+
+	// translate the ceph status struct to the crd status
+	cluster.Status.CephStatus = toCustomResourceStatus(cluster.Status, status)
+	if _, err := c.context.RookClientset.CephV1().CephClusters(c.namespace).Update(cluster); err != nil {
+		return fmt.Errorf("failed to update cluster %s status: %+v", c.namespace, err)
+	}
+
+	return nil
+}
+
+// toCustomResourceStatus converts the ceph status to the struct expected for the CephCluster CR status
+func toCustomResourceStatus(currentStatus cephv1.ClusterStatus, newStatus *client.CephStatus) *cephv1.CephStatus {
+	s := &cephv1.CephStatus{
+		Health:      newStatus.Health.Status,
+		LastChecked: formatTime(time.Now().UTC()),
+		Details:     make(map[string]cephv1.CephHealthMessage),
+	}
+	for name, message := range newStatus.Health.Checks {
+		s.Details[name] = cephv1.CephHealthMessage{
+			Severity: message.Severity,
+			Message:  message.Summary.Message,
+		}
+	}
+	if currentStatus.CephStatus != nil {
+		s.PreviousHealth = currentStatus.CephStatus.PreviousHealth
+		s.LastChanged = currentStatus.CephStatus.LastChanged
+		if currentStatus.CephStatus.Health != s.Health {
+			s.PreviousHealth = currentStatus.CephStatus.Health
+			s.LastChanged = s.LastChecked
+		}
+	}
+	return s
+}
+
+func formatTime(t time.Time) string {
+	return t.Format(time.RFC3339)
+}

pkg/operator/ceph/cluster/cephstatus_test.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2019 The Rook Authors. All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+	http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package cluster to manage Kubernetes storage.
+package cluster
+
+import (
+	"testing"
+	"time"
+
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
+	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCephStatus(t *testing.T) {
+	newStatus := &client.CephStatus{
+		Health: client.HealthStatus{Status: "HEALTH_OK"},
+	}
+
+	// Empty initial status will have no previous health
+	currentStatus := cephv1.ClusterStatus{}
+	aggregateStatus := toCustomResourceStatus(currentStatus, newStatus)
+	assert.NotNil(t, aggregateStatus)
+	assert.Equal(t, "HEALTH_OK", aggregateStatus.Health)
+	assert.NotEqual(t, "", aggregateStatus.LastChecked)
+	assert.Equal(t, "", aggregateStatus.LastChanged)
+	assert.Equal(t, "", aggregateStatus.PreviousHealth)
+	assert.Equal(t, 0, len(aggregateStatus.Details))
+
+	// Set the current status to the same as the new status and there will be no previous health
+	currentStatus.CephStatus = &cephv1.CephStatus{
+		Health: "HEALTH_OK",
+	}
+	aggregateStatus = toCustomResourceStatus(currentStatus, newStatus)
+	assert.NotNil(t, aggregateStatus)
+	assert.Equal(t, "HEALTH_OK", aggregateStatus.Health)
+	assert.NotEqual(t, "", aggregateStatus.LastChecked)
+	assert.Equal(t, "", aggregateStatus.LastChanged)
+	assert.Equal(t, "", aggregateStatus.PreviousHealth)
+	assert.Equal(t, 0, len(aggregateStatus.Details))
+
+	// Set the new status to something different and we should get a previous health
+	// Simulate the previous check a minute ago.
+	previousTime := formatTime(time.Now().Add(-time.Minute).UTC())
+	currentStatus.CephStatus.LastChecked = previousTime
+	newStatus.Health.Status = "HEALTH_WARN"
+	aggregateStatus = toCustomResourceStatus(currentStatus, newStatus)
+	assert.NotNil(t, aggregateStatus)
+	assert.Equal(t, "HEALTH_WARN", aggregateStatus.Health)
+	assert.NotEqual(t, "", aggregateStatus.LastChecked)
+	assert.Equal(t, aggregateStatus.LastChecked, aggregateStatus.LastChanged)
+	assert.Equal(t, "HEALTH_OK", aggregateStatus.PreviousHealth)
+	assert.Equal(t, 0, len(aggregateStatus.Details))
+
+	// Add some details to the warning
+	osdDownMsg := client.CheckMessage{Severity: "HEALTH_WARN"}
+	osdDownMsg.Summary.Message = "1 osd down"
+	pgAvailMsg := client.CheckMessage{Severity: "HEALTH_ERR"}
+	pgAvailMsg.Summary.Message = "'Reduced data availability: 100 pgs stale'"
+	newStatus.Health.Checks = map[string]client.CheckMessage{
+		"OSD_DOWN":        osdDownMsg,
+		"PG_AVAILABILITY": pgAvailMsg,
+	}
+	newStatus.Health.Status = "HEALTH_ERR"
+	aggregateStatus = toCustomResourceStatus(currentStatus, newStatus)
+	assert.NotNil(t, aggregateStatus)
+	assert.Equal(t, "HEALTH_ERR", aggregateStatus.Health)
+	assert.NotEqual(t, "", aggregateStatus.LastChecked)
+	assert.Equal(t, aggregateStatus.LastChecked, aggregateStatus.LastChanged)
+	assert.Equal(t, "HEALTH_OK", aggregateStatus.PreviousHealth)
+	assert.Equal(t, 2, len(aggregateStatus.Details))
+	assert.Equal(t, osdDownMsg.Summary.Message, aggregateStatus.Details["OSD_DOWN"].Message)
+	assert.Equal(t, osdDownMsg.Severity, aggregateStatus.Details["OSD_DOWN"].Severity)
+	assert.Equal(t, pgAvailMsg.Summary.Message, aggregateStatus.Details["PG_AVAILABILITY"].Message)
+	assert.Equal(t, pgAvailMsg.Severity, aggregateStatus.Details["PG_AVAILABILITY"].Severity)
+}

pkg/operator/ceph/cluster/controller.go

@@ -309,6 +309,10 @@ func (c *ClusterController) onAdd(obj interface{}) {
 	osdChecker := osd.NewMonitor(c.context, cluster.Namespace)
 	go osdChecker.Start(cluster.stopCh)
 
+	// Start the ceph status checker
+	cephChecker := newCephStatusChecker(c.context, cluster.Namespace, clusterObj.Name)
+	go cephChecker.checkCephStatus(cluster.stopCh)
+
 	// add the finalizer to the crd
 	err = c.addFinalizer(clusterObj)
 	if err != nil {
@@ -396,7 +400,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
 		return
 	}
 
-	logger.Infof("update event for cluster %s", newClust.Namespace)
+	logger.Debugf("update event for cluster %s", newClust.Namespace)
 
 	// Check if the cluster is being deleted. This code path is called when a finalizer is specified in the crd.
 	// When a cluster is requested for deletion, K8s will only set the deletion timestamp if there are any finalizers in the list.
@@ -420,7 +424,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
 
 	changed, _ := clusterChanged(oldClust.Spec, newClust.Spec, cluster)
 	if !changed {
-		logger.Infof("update event for cluster %s is not supported", newClust.Namespace)
+		logger.Debugf("update event for cluster %s is not supported", newClust.Namespace)
 		return
 	}
 
@@ -659,7 +663,10 @@ func (c *ClusterController) removeFinalizer(obj interface{}) {
 	logger.Warningf("giving up from removing the %s cluster finalizer", fname)
 }
 
+// updateClusterStatus updates the status of the cluster custom resource, whether it is being updated or is completed
 func (c *ClusterController) updateClusterStatus(namespace, name string, state cephv1.ClusterState, message string) error {
+	logger.Infof("CephCluster %s status: %s", namespace, state)
+
 	// get the most recent cluster CRD object
 	cluster, err := c.context.RookClientset.CephV1().CephClusters(namespace).Get(name, metav1.GetOptions{})
 	if err != nil {
@@ -667,9 +674,11 @@ func (c *ClusterController) updateClusterStatus(namespace, name string, state ce
 	}
 
 	// update the status on the retrieved cluster object
-	cluster.Status = cephv1.ClusterStatus{State: state, Message: message}
-	if _, err := c.context.RookClientset.CephV1().CephClusters(cluster.Namespace).Update(cluster); err != nil {
-		return fmt.Errorf("failed to update cluster %s status: %+v", cluster.Namespace, err)
+	// do not overwrite the ceph status that is updated in a separate goroutine
+	cluster.Status.State = state
+	cluster.Status.Message = message
+	if _, err := c.context.RookClientset.CephV1().CephClusters(namespace).Update(cluster); err != nil {
+		return fmt.Errorf("failed to update cluster %s status: %+v", namespace, err)
 	}
 
 	return nil

tests/framework/clients/test_client.go

@@ -59,7 +59,7 @@ func CreateTestClient(k8sHelper *utils.K8sHelper, manifests installer.CephManife
 // Status returns rook status details
 func (c TestClient) Status(namespace string) (client.CephStatus, error) {
 	context := c.k8sh.MakeContext()
-	status, err := client.Status(context, namespace)
+	status, err := client.Status(context, namespace, false)
 	if err != nil {
 		return client.CephStatus{}, fmt.Errorf("failed to get status: %+v", err)
 	}

tests/framework/installer/ceph_installer.go

@@ -31,6 +31,7 @@ import (
 	"github.com/rook/rook/pkg/daemon/ceph/client"
 	opspec "github.com/rook/rook/pkg/operator/ceph/spec"
 	"github.com/rook/rook/tests/framework/utils"
+	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -360,6 +361,9 @@ func (h *CephInstaller) UninstallRookFromMultipleNS(systemNamespace string, name
 			// When the test has failed, it's sometimes useful to have pod descriptions to check
 			// that pods are configured as expected.
 			h.k8shelper.PrintPodDescribeForNamespace(namespace)
+		} else {
+			// if the test passed, check that the ceph status is HEALTH_OK before we tear the cluster down
+			h.checkCephHealthStatus(namespace)
 		}
 
 		roles := h.Manifests.GetClusterRoles(namespace, systemNamespace)
@@ -438,6 +442,33 @@ func (h *CephInstaller) UninstallRookFromMultipleNS(systemNamespace string, name
 	}
 }
 
+func (h *CephInstaller) checkCephHealthStatus(namespace string) {
+	clusterResource, err := h.k8shelper.RookClientset.CephV1().CephClusters(namespace).Get(namespace, metav1.GetOptions{})
+	assert.Nil(h.T(), err)
+	assert.Equal(h.T(), "Created", string(clusterResource.Status.State))
+
+	// Depending on the tests, the health may be fluctuating with different components being started or stopped.
+	// If needed, give it a few seconds to settle and check the status again.
+	if clusterResource.Status.CephStatus.Health != "HEALTH_OK" {
+		time.Sleep(10 * time.Second)
+		clusterResource, err = h.k8shelper.RookClientset.CephV1().CephClusters(namespace).Get(namespace, metav1.GetOptions{})
+		assert.Nil(h.T(), err)
+	}
+
+	// The health status is not stable enough for the integration tests to rely on.
+	// We should enable this check if we can get the ceph status to be stable despite all the changing configurations performed by rook.
+	//assert.Equal(h.T(), "HEALTH_OK", clusterResource.Status.CephStatus.Health)
+	assert.NotEqual(h.T(), "", clusterResource.Status.CephStatus.LastChecked)
+
+	// Print the details if the health is not ok
+	if clusterResource.Status.CephStatus.Health != "HEALTH_OK" {
+		logger.Errorf("Ceph health status: %s", clusterResource.Status.CephStatus.Health)
+		for name, message := range clusterResource.Status.CephStatus.Details {
+			logger.Errorf("Ceph health message: %s. %s: %s", name, message.Severity, message.Message)
+		}
+	}
+}
+
 func (h *CephInstaller) cleanupDir(node, dir string) error {
 	resources := h.Manifests.GetCleanupPod(node, dir)
 	_, err := h.k8shelper.KubectlWithStdin(resources, createFromStdinArgs...)

tests/framework/installer/ceph_manifests.go

@@ -146,6 +146,10 @@ spec:
       type: string
       description: Current State
       JSONPath: .status.state
+    - name: Health
+      type: string
+      description: Ceph Health
+      JSONPath: .status.ceph.health
 ---
 apiVersion: apiextensions.k8s.io/v1beta1
 kind: CustomResourceDefinition
@@ -727,6 +731,8 @@ spec:
         env:
         - name: ROOK_LOG_LEVEL
           value: INFO
+        - name: ROOK_CEPH_STATUS_CHECK_INTERVAL
+          value: "5s"
         - name: ROOK_MON_HEALTHCHECK_INTERVAL
           value: "10s"
         - name: ROOK_MON_OUT_TIMEOUT