From 5ebb651e37774aae6f3a0becd139d5a89a9bedec Mon Sep 17 00:00:00 2001 From: travisn Date: Sat, 6 Apr 2019 07:53:52 -0600 Subject: [PATCH] ceph: update the cephcluster custom resource with ceph health The CephCluster resource will now expose the health of the ceph cluster so admins can query the status with k8s api or kubectl instead of needing to run the rook toolbox. The operator will query the ceph status periodically and update the status in the custom resource. Signed-off-by: travisn --- PendingReleaseNotes.md | 1 + .../rook-ceph/templates/deployment.yaml | 4 + cluster/charts/rook-ceph/values.yaml.tmpl | 3 + cluster/examples/kubernetes/ceph/common.yaml | 4 + .../examples/kubernetes/ceph/operator.yaml | 3 + pkg/apis/ceph.rook.io/v1/types.go | 18 ++- .../ceph.rook.io/v1/zz_generated.deepcopy.go | 46 +++++- pkg/daemon/ceph/client/command.go | 5 + pkg/daemon/ceph/client/mon.go | 27 ---- pkg/daemon/ceph/client/status.go | 6 +- pkg/operator/ceph/cluster/cephstatus.go | 140 ++++++++++++++++++ pkg/operator/ceph/cluster/cephstatus_test.go | 90 +++++++++++ pkg/operator/ceph/cluster/controller.go | 19 ++- tests/framework/clients/test_client.go | 2 +- tests/framework/installer/ceph_installer.go | 31 ++++ tests/framework/installer/ceph_manifests.go | 6 + 16 files changed, 366 insertions(+), 39 deletions(-) create mode 100644 pkg/operator/ceph/cluster/cephstatus.go create mode 100644 pkg/operator/ceph/cluster/cephstatus_test.go diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index b0c54dc52..30b1791d6 100644 --- a/PendingReleaseNotes.md +++ b/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. diff --git a/cluster/charts/rook-ceph/templates/deployment.yaml b/cluster/charts/rook-ceph/templates/deployment.yaml index 008082884..b602a184d 100644 --- a/cluster/charts/rook-ceph/templates/deployment.yaml +++ b/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 diff --git a/cluster/charts/rook-ceph/values.yaml.tmpl b/cluster/charts/rook-ceph/values.yaml.tmpl index 690265502..c982b0930 100644 --- a/cluster/charts/rook-ceph/values.yaml.tmpl +++ b/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" diff --git a/cluster/examples/kubernetes/ceph/common.yaml b/cluster/examples/kubernetes/ceph/common.yaml index 6b0f082b7..215071d34 100644 --- a/cluster/examples/kubernetes/ceph/common.yaml +++ b/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 diff --git a/cluster/examples/kubernetes/ceph/operator.yaml b/cluster/examples/kubernetes/ceph/operator.yaml index 830308bc4..ff63bd489 100644 --- a/cluster/examples/kubernetes/ceph/operator.yaml +++ b/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" diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go index be0c57f1c..87ba8d4dd 100755 --- a/pkg/apis/ceph.rook.io/v1/types.go +++ b/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 diff --git a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go b/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go index b8fa345ec..35b364995 100644 --- a/pkg/apis/ceph.rook.io/v1/zz_generated.deepcopy.go +++ b/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 } diff --git a/pkg/daemon/ceph/client/command.go b/pkg/daemon/ceph/client/command.go index d16ed8ba6..a6be9aa03 100644 --- a/pkg/daemon/ceph/client/command.go +++ b/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) diff --git a/pkg/daemon/ceph/client/mon.go b/pkg/daemon/ceph/client/mon.go index 8d6d11353..5f892b307 100644 --- a/pkg/daemon/ceph/client/mon.go +++ b/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) diff --git a/pkg/daemon/ceph/client/status.go b/pkg/daemon/ceph/client/status.go index c58250ea9..d49178c2a 100644 --- a/pkg/daemon/ceph/client/status.go +++ b/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 } diff --git a/pkg/operator/ceph/cluster/cephstatus.go b/pkg/operator/ceph/cluster/cephstatus.go new file mode 100644 index 000000000..abd133852 --- /dev/null +++ b/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) +} diff --git a/pkg/operator/ceph/cluster/cephstatus_test.go b/pkg/operator/ceph/cluster/cephstatus_test.go new file mode 100644 index 000000000..e79cb296c --- /dev/null +++ b/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) +} diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 2fbfdca22..28dbd0bc8 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/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 diff --git a/tests/framework/clients/test_client.go b/tests/framework/clients/test_client.go index cd7f53355..3f5f0d9b4 100644 --- a/tests/framework/clients/test_client.go +++ b/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) } diff --git a/tests/framework/installer/ceph_installer.go b/tests/framework/installer/ceph_installer.go index 6e91d644a..60a5075e4 100644 --- a/tests/framework/installer/ceph_installer.go +++ b/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...) diff --git a/tests/framework/installer/ceph_manifests.go b/tests/framework/installer/ceph_manifests.go index da589f3b2..bb90db99b 100644 --- a/tests/framework/installer/ceph_manifests.go +++ b/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