feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

Merge pull request #5805 from kesavanvt/add_device_class

Browse Source 
ceph: Adding deviceClasses info to CephCluster CR
This commit is contained in:
Travis Nielsen
2020-07-22 16:00:18 -06:00
committed by GitHub
parent 3e30b6fbdf 453200045e
commit 77e787b00e
4 changed files with 154 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/ceph.rook.io/v1/types.go
+9
View File
@@ -188,6 +188,7 @@ type ClusterStatus struct {
Message string `json:"message,omitempty"`
Conditions []Condition `json:"conditions,omitempty"`
CephStatus *CephStatus `json:"ceph,omitempty"`
CephStorage *CephStorage `json:"storage,omitempty"`
CephVersion *ClusterVersion `json:"version,omitempty"`
}
@@ -199,6 +200,14 @@ type CephStatus struct {
PreviousHealth string `json:"previousHealth,omitempty"`
}
type CephStorage struct {
DeviceClasses []DeviceClasses `json:"deviceClasses,omitempty"`
}
type DeviceClasses struct {
Name string `json:"name,omitempty"`
}
type ClusterVersion struct {
Image string `json:"image,omitempty"`
Version string `json:"version,omitempty"`
pkg/daemon/ceph/client/deviceclass.go
+41
View File
@@ -0,0 +1,41 @@
/*
Copyright 2020 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 client
import (
"encoding/json"
"github.com/pkg/errors"
"github.com/rook/rook/pkg/clusterd"
)
// GetDeviceClasses gets the available device classes.
func GetDeviceClasses(context *clusterd.Context, clusterInfo *ClusterInfo) ([]string, error) {
args := []string{"osd", "crush", "class", "ls"}
cmd := NewCephCommand(context, clusterInfo, args)
buf, err := cmd.Run()
if err != nil {
return nil, errors.Wrapf(err, "failed to get deviceclasses. %s", string(buf))
}
var deviceclass []string
if err := json.Unmarshal(buf, &deviceclass); err != nil {
return nil, errors.Wrap(err, "failed to unmarshal osd crush class list response")
}
return deviceclass, nil
}
pkg/operator/ceph/cluster/osd/health.go
+56 -6
View File
@@ -17,13 +17,16 @@ limitations under the License.
package osd
import (
"context"
"fmt"
"reflect"
"time"
"github.com/pkg/errors"
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"
opcontroller "github.com/rook/rook/pkg/operator/ceph/controller"
"github.com/rook/rook/pkg/operator/k8sutil"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -75,10 +78,7 @@ func (m *OSDHealthMonitor) Start(stopCh chan struct{}) {
select {
case <-time.After(m.interval):
logger.Debug("checking osd processes status.")
err := m.checkOSDHealth()
if err != nil {
logger.Debugf("failed OSD status check. %v", err)
}
m.checkOSDHealth()
case <-stopCh:
logger.Infof("Stopping monitoring of OSDs in namespace %s", m.clusterInfo.Namespace)
@@ -93,10 +93,34 @@ func (m *OSDHealthMonitor) Update(removeOSDsIfOUTAndSafeToRemove bool) {
}
// checkOSDHealth takes action when needed if the OSDs are not healthy
func (m *OSDHealthMonitor) checkOSDHealth() error {
func (m *OSDHealthMonitor) checkOSDHealth() {
err := m.checkOSDDump()
if err != nil {
logger.Debugf("failed to check OSD Dump. %v", err)
}
err = m.checkDeviceClasses()
if err != nil {
logger.Debugf("failed to check device classes. %v", err)
}
}
func (m *OSDHealthMonitor) checkDeviceClasses() error {
devices, err := client.GetDeviceClasses(m.context, m.clusterInfo)
if err != nil {
return errors.Wrap(err, "failed to get osd device classes")
}
if len(devices) > 0 {
m.updateCephStatus(devices)
}
return nil
}
func (m *OSDHealthMonitor) checkOSDDump() error {
osdDump, err := client.GetOSDDump(m.context, m.clusterInfo)
if err != nil {
return err
return errors.Wrap(err, "failed to get osd dump")
}
for _, osdStatus := range osdDump.OSDs {
@@ -184,3 +208,29 @@ func (m *OSDHealthMonitor) restartOSDIfStuck(osdID int) error {
}
return nil
}
// updateCephStorage updates the CR with deviceclass details
func (m *OSDHealthMonitor) updateCephStatus(devices []string) {
cephCluster := &cephv1.CephCluster{}
cephClusterStorage := cephv1.CephStorage{}
for _, device := range devices {
cephClusterStorage.DeviceClasses = append(cephClusterStorage.DeviceClasses, cephv1.DeviceClasses{Name: device})
}
err := m.context.Client.Get(context.TODO(), m.clusterInfo.NamespacedName(), cephCluster)
if err != nil {
if kerrors.IsNotFound(err) {
logger.Debug("CephCluster resource not found. Ignoring since object must be deleted.")
return
}
logger.Errorf("failed to retrieve ceph cluster %q to update ceph Storage. %v", m.clusterInfo.NamespacedName().Name, err)
return
}
if !reflect.DeepEqual(cephCluster.Status.CephStorage, &cephClusterStorage) {
cephCluster.Status.CephStorage = &cephClusterStorage
if err := opcontroller.UpdateStatus(m.context.Client, cephCluster); err != nil {
logger.Errorf("failed to update cluster %q Storage. %v", m.clusterInfo.NamespacedName().Name, err)
return
}
}
}
pkg/operator/ceph/cluster/osd/health_test.go
+48 -1
View File
@@ -22,6 +22,7 @@ import (
"time"
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
"github.com/rook/rook/pkg/client/clientset/versioned/scheme"
"github.com/rook/rook/pkg/clusterd"
"github.com/rook/rook/pkg/daemon/ceph/client"
"github.com/rook/rook/pkg/operator/k8sutil"
@@ -32,6 +33,8 @@ import (
v1 "k8s.io/api/core/v1"
kerrors "k8s.io/apimachinery/pkg/api/errors"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
)
func TestOSDHealthCheck(t *testing.T) {
@@ -93,7 +96,7 @@ func TestOSDHealthCheck(t *testing.T) {
osdMon := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{})
// Run OSD monitoring routine
err := osdMon.checkOSDHealth()
err := osdMon.checkOSDDump()
assert.Nil(t, err)
// After creating an OSD, the dump has 1 mocked cmd and safe to destroy has 1 mocked cmd
assert.Equal(t, 2, execCount)
@@ -196,3 +199,47 @@ func TestNewOSDHealthMonitor(t *testing.T) {
})
}
}
func TestDeviceClasses(t *testing.T) {
clusterInfo := client.AdminClusterInfo("fake")
clusterInfo.SetName("rook-ceph")
var execCount = 0
executor := &exectest.MockExecutor{
MockExecuteCommandWithOutputFile: func(command string, outFileArg string, args ...string) (string, error) {
return "{\"key\":\"mysecurekey\", \"osdid\":3.0}", nil
},
}
executor.MockExecuteCommandWithOutputFile = func(command string, outFileArg string, args ...string) (string, error) {
logger.Infof("ExecuteCommandWithOutputFile: %s %v", command, args)
execCount++
if args[1] == "crush" && args[2] == "class" && args[3] == "ls" {
// Mock executor for OSD crush class list command, returning ssd as available device class
return `["ssd"]`, nil
}
return "", nil
}
cephCluster := &cephv1.CephCluster{}
// Objects to track in the fake client.
object := []runtime.Object{
cephCluster,
}
s := scheme.Scheme
// Create a fake client to mock API calls.
cl := fake.NewFakeClientWithScheme(s, object...)
context := &clusterd.Context{
Executor: executor,
Client: cl,
}
// Initializing an OSD monitoring
osdMon := NewOSDHealthMonitor(context, clusterInfo, true, cephv1.CephClusterHealthCheckSpec{})
// Run OSD monitoring routine
err := osdMon.checkDeviceClasses()
assert.Nil(t, err)
// checkDeviceClasses has 1 mocked cmd for fetching the device classes
assert.Equal(t, 1, execCount)
}