From 7c942604f63579e878e4387e75c8c8caedea7bb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Juan=20Miguel=20Olmo=20Mart=C3=ADnez?= Date: Mon, 7 Oct 2019 11:14:43 +0200 Subject: [PATCH] ceph: Get data in dev. configmaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit **Description of your changes:** This modification adds the information extracted from 'ceph-volume inventory': command to the device configmaps generated by the discovery daemon when "rook discover" starts with the new boolean "--use-ceph-volume" parameter. Resolves # https://github.com/rook/rook/issues/2606 Now the field contains all the information returned from command. Signed-off-by: Juan Miguel Olmo Martínez --- cmd/rook/discover.go | 6 +- pkg/clusterd/disk.go | 1 - pkg/daemon/discover/discover.go | 83 +++++++++++++++++++++++++- pkg/daemon/discover/discover_test.go | 68 +++++++++++++++++++++ pkg/operator/ceph/operator.go | 2 +- pkg/operator/discover/discover.go | 26 ++++---- pkg/operator/discover/discover_test.go | 2 +- pkg/operator/edgefs/operator.go | 4 +- pkg/util/sys/device.go | 2 + 9 files changed, 176 insertions(+), 18 deletions(-) diff --git a/cmd/rook/discover.go b/cmd/rook/discover.go index 79d5abf24..468515a46 100644 --- a/cmd/rook/discover.go +++ b/cmd/rook/discover.go @@ -32,10 +32,14 @@ var ( // interval between discovering devices discoverDevicesInterval time.Duration + + // Uses ceph-volume inventory to extend devices information + usesCVInventory bool ) func init() { discoverCmd.Flags().DurationVar(&discoverDevicesInterval, "discover-interval", 60*time.Minute, "interval between discovering devices (default 60m)") + discoverCmd.Flags().BoolVar(&usesCVInventory, "use-ceph-volume", false, "Use ceph-volume inventory to extend storage devices information (default false)") flags.SetFlagsFromEnv(discoverCmd.Flags(), rook.RookEnvVarPrefix) discoverCmd.RunE = startDiscover @@ -48,7 +52,7 @@ func startDiscover(cmd *cobra.Command, args []string) error { context := rook.NewContext() - err := discover.Run(context, discoverDevicesInterval) + err := discover.Run(context, discoverDevicesInterval, usesCVInventory) if err != nil { rook.TerminateFatal(err) } diff --git a/pkg/clusterd/disk.go b/pkg/clusterd/disk.go index 09fb64411..d95b4c83b 100644 --- a/pkg/clusterd/disk.go +++ b/pkg/clusterd/disk.go @@ -43,7 +43,6 @@ func ignoreDevice(d string) bool { // Discover all the details of devices available on the local node func DiscoverDevices(executor exec.Executor) ([]*sys.LocalDisk, error) { - var disks []*sys.LocalDisk devices, err := sys.ListDevices(executor) if err != nil { diff --git a/pkg/daemon/discover/discover.go b/pkg/daemon/discover/discover.go index 019ad978f..fab0b5da5 100644 --- a/pkg/daemon/discover/discover.go +++ b/pkg/daemon/discover/discover.go @@ -24,6 +24,7 @@ import ( "os" "os/exec" "os/signal" + "path" "regexp" "strings" "syscall" @@ -51,18 +52,30 @@ var ( cmName string cm *v1.ConfigMap udevEventPeriod = time.Duration(5) * time.Second + useCVInventory bool ) -func Run(context *clusterd.Context, probeInterval time.Duration) error { +type CephVolumeInventory struct { + Path string `json:"path"` + Available bool `json:"available"` + RejectedReasons json.RawMessage `json:"rejected_reasons"` + SysAPI json.RawMessage `json:"sys_api"` + LVS json.RawMessage `json:"lvs"` +} + +func Run(context *clusterd.Context, probeInterval time.Duration, useCV bool) error { if context == nil { return fmt.Errorf("nil context") } - logger.Infof("device discovery interval is %s", probeInterval.String()) + logger.Debugf("device discovery interval is %q", probeInterval.String()) + logger.Debugf("use ceph-volume inventory is %t", useCV) nodeName = os.Getenv(k8sutil.NodeNameEnvVar) namespace = os.Getenv(k8sutil.PodNamespaceEnvVar) cmName = k8sutil.TruncateNodeName(LocalDiskCMName, nodeName) + useCVInventory = useCV sigc := make(chan os.Signal, 1) signal.Notify(sigc, syscall.SIGTERM) + err := updateDeviceCM(context) if err != nil { logger.Infof("failed to update device configmap: %v", err) @@ -244,6 +257,10 @@ func checkDeviceListsEqual(oldDevs, newDevs []sys.LocalDisk) bool { if oldDev.Partitions != nil && match.Partitions == nil { return false } + if string(oldDev.CephVolumeData) == "" && string(match.CephVolumeData) != "" { + // return ceph volume inventory data was not enabled before + return false + } } for _, newDev := range newDevs { @@ -290,6 +307,7 @@ func updateDeviceCM(context *clusterd.Context) error { logger.Infof("failed to marshal: %v", err) return err } + deviceStr := string(deviceJson) if cm == nil { cm, err = context.Clientset.CoreV1().ConfigMaps(namespace).Get(cmName, metav1.GetOptions{}) @@ -305,6 +323,7 @@ func updateDeviceCM(context *clusterd.Context) error { data := make(map[string]string, 1) data[LocalDiskCMData] = deviceStr + // the map doesn't exist yet, create it now cm = &v1.ConfigMap{ ObjectMeta: metav1.ObjectMeta{ @@ -348,6 +367,18 @@ func probeDevices(context *clusterd.Context) ([]sys.LocalDisk, error) { if err != nil { return devices, fmt.Errorf("failed initial hardware discovery. %+v", err) } + + // ceph-volume inventory command takes a little time to complete. + // Get this data only if it is needed and once by function execution + var cvInventory *map[string]string = nil + if useCVInventory { + logger.Infof("Getting ceph-volume inventory information") + cvInventory, err = getCephVolumeInventory(context) + if err != nil { + logger.Errorf("error getting ceph-volume inventory: %v", err) + } + } + for _, device := range localDevices { if device == nil { continue @@ -372,9 +403,57 @@ func probeDevices(context *clusterd.Context) ([]sys.LocalDisk, error) { device.Filesystem = fs device.Empty = clusterd.GetDeviceEmpty(device) + // Add the information provided by ceph-volume inventory + if cvInventory != nil { + CVData, deviceExists := (*cvInventory)[path.Join("/dev/", device.Name)] + if deviceExists { + device.CephVolumeData = CVData + } else { + logger.Errorf("ceph-volume information for device %q not found", device.Name) + } + } else { + device.CephVolumeData = "" + } + devices = append(devices, *device) } logger.Infof("available devices: %+v", devices) return devices, nil } + +// getCephVolumeInventory: Return a map of strings indexed by device with the +// information about the device returned by the command +func getCephVolumeInventory(context *clusterd.Context) (*map[string]string, error) { + inventory, err := context.Executor.ExecuteCommandWithOutput(false, "", "ceph-volume", "inventory", "--format", "json") + if err != nil { + return nil, fmt.Errorf("failed to execute ceph-volume inventory. %+v", err) + } + + // Return a map with the information of each device indexed by path + CVDevices := make(map[string]string) + + // No data retrieved from ceph-volume + if inventory == "" { + return &CVDevices, nil + } + + // Get a slice to store the json data + bInventory := []byte(inventory) + var CVInventory []CephVolumeInventory + err = json.Unmarshal(bInventory, &CVInventory) + if err != nil { + return &CVDevices, fmt.Errorf("error unmarshalling json data coming from ceph-volume inventory. %v", err) + } + + for _, device := range CVInventory { + jsonData, err := json.Marshal(device) + if err != nil { + logger.Errorf("error marshaling json data for device: %v", device.Path) + } else { + CVDevices[device.Path] = string(jsonData) + } + } + + return &CVDevices, nil +} diff --git a/pkg/daemon/discover/discover_test.go b/pkg/daemon/discover/discover_test.go index b1ca5fb60..55f636994 100644 --- a/pkg/daemon/discover/discover_test.go +++ b/pkg/daemon/discover/discover_test.go @@ -18,6 +18,7 @@ limitations under the License. package discover import ( + "fmt" "testing" "github.com/rook/rook/pkg/clusterd" @@ -349,3 +350,70 @@ func TestDeviceListsEqual(t *testing.T) { }, )) } + +func TestGetCephVolumeInventory(t *testing.T) { + run := 0 + executor := &exectest.MockExecutor{ + MockExecuteCommandWithOutput: func(debug bool, actionName string, command string, arg ...string) (string, error) { + run++ + logger.Infof("run %d action %s command %s", run, actionName, command) + switch { + case run == 1: + return `[{"available": true, "rejected_reasons": [], "sys_api": {"scheduler_mode": "noop", +"rotational": "0", "vendor": "ATA", "human_readable_size": "25.00 GB", "sectors": 0, "sas_device_handle": "", +"partitions": {}, "rev": "1.0", "sas_address": "", "locked": 0, "sectorsize": "512", "removable": "0", "path": "/dev/sdb", +"support_discard": "", "model": "VBOX SMART HARDDISK", "ro": "0", "nr_requests": "128", "size": 26843545600.0}, +"lvs": [], "path": "/dev/sdb"}, {"available": false, "rejected_reasons": ["locked"], "sys_api": {"scheduler_mode": "noop", + "rotational": "1", "vendor": "ATA", "human_readable_size": "32.00 GB", "sectors": 0, "sas_device_handle": "", + "partitions": {"sda2": {"start": "2099200", "holders": ["dm-0", "dm-1"], "sectorsize": 512, "sectors": "65009664", + "size": "31.00 GB"}, "sda1": {"start": "2048", "holders": [], "sectorsize": 512, "sectors": "2097152", "size": "1024.00 MB"}}, + "rev": "1.0", "sas_address": "", "locked": 1, "sectorsize": "512", "removable": "0", "path": "/dev/sda", "support_discard": "", + "model": "VBOX HARDDISK", "ro": "0", "nr_requests": "128", "size": 34359738368.0}, "lvs": [{"comment": "not used by ceph", "name": "swap"}, + {"comment": "not used by ceph", "name": "root"}], "path": "/dev/sda"}] + `, nil + case run == 2: // No data returned from Ceph Volume + return ``, nil + case run == 3: // No devices returned from Ceph Volume + return `[]`, nil + case run == 4: // Error executing Ceph Volume + return ``, fmt.Errorf("unexplainable error") + case run == 5: // A device without sys_api data + return `[{"available": true }]`, nil + } + return "", nil + }, + } + + context := &clusterd.Context{Executor: executor} + + dev_sda := `{"path":"/dev/sda","available":false,"rejected_reasons":["locked"],"sys_api":{"scheduler_mode":"noop","rotational":"1","vendor":"ATA","human_readable_size":"32.00 GB","sectors":0,"sas_device_handle":"","partitions":{"sda2":{"start":"2099200","holders":["dm-0","dm-1"],"sectorsize":512,"sectors":"65009664","size":"31.00 GB"},"sda1":{"start":"2048","holders":[],"sectorsize":512,"sectors":"2097152","size":"1024.00 MB"}},"rev":"1.0","sas_address":"","locked":1,"sectorsize":"512","removable":"0","path":"/dev/sda","support_discard":"","model":"VBOX HARDDISK","ro":"0","nr_requests":"128","size":34359738368.0},"lvs":[{"comment":"not used by ceph","name":"swap"},{"comment":"not used by ceph","name":"root"}]}` + dev_sdb := `{"path":"/dev/sdb","available":true,"rejected_reasons":[],"sys_api":{"scheduler_mode":"noop","rotational":"0","vendor":"ATA","human_readable_size":"25.00 GB","sectors":0,"sas_device_handle":"","partitions":{},"rev":"1.0","sas_address":"","locked":0,"sectorsize":"512","removable":"0","path":"/dev/sdb","support_discard":"","model":"VBOX SMART HARDDISK","ro":"0","nr_requests":"128","size":26843545600.0},"lvs":[]}` + + // Normal execution + cvdata, err := getCephVolumeInventory(context) + + assert.Nil(t, err) + assert.Equal(t, len(*cvdata), 2) + assert.Equal(t, (*cvdata)["/dev/sda"], dev_sda) + assert.Equal(t, (*cvdata)["/dev/sdb"], dev_sdb) + + // No data returned from Ceph Volume + cvdata, err = getCephVolumeInventory(context) + assert.Nil(t, err) + assert.Equal(t, len(*cvdata), 0) + + // No devices returned from Ceph Volume + cvdata, err = getCephVolumeInventory(context) + assert.Nil(t, err) + assert.Equal(t, len(*cvdata), 0) + + // Error executing Ceph Volume + cvdata, err = getCephVolumeInventory(context) + assert.Error(t, err, "unexplainable error") + assert.Nil(t, cvdata, 0) + + // // A device without sys_api data + cvdata, err = getCephVolumeInventory(context) + assert.Nil(t, err) + assert.Equal(t, len(*cvdata), 1) +} diff --git a/pkg/operator/ceph/operator.go b/pkg/operator/ceph/operator.go index 9f85f479d..fd9891ba3 100644 --- a/pkg/operator/ceph/operator.go +++ b/pkg/operator/ceph/operator.go @@ -106,7 +106,7 @@ func (o *Operator) Run() error { if EnableDiscoveryDaemon { rookDiscover := discover.New(o.context.Clientset) - if err := rookDiscover.Start(o.operatorNamespace, o.rookImage, o.securityAccount); err != nil { + if err := rookDiscover.Start(o.operatorNamespace, o.rookImage, o.securityAccount, true); err != nil { return fmt.Errorf("error starting device discovery daemonset. %+v", err) } } diff --git a/pkg/operator/discover/discover.go b/pkg/operator/discover/discover.go index 23839f85d..109a5a3c4 100644 --- a/pkg/operator/discover/discover.go +++ b/pkg/operator/discover/discover.go @@ -49,6 +49,7 @@ const ( deviceInUseAppName = "rook-claimed-devices" deviceInUseClusterAttr = "rook.io/cluster" discoverIntervalEnv = "ROOK_DISCOVER_DEVICES_INTERVAL" + defaultDiscoverInterval = "60m" ) var logger = capnslog.NewPackageLogger("github.com/rook/rook", "op-discover") @@ -66,17 +67,23 @@ func New(clientset kubernetes.Interface) *Discover { } // Start the discover -func (d *Discover) Start(namespace, discoverImage, securityAccount string) error { +func (d *Discover) Start(namespace, discoverImage, securityAccount string, useCephVolume bool) error { - err := d.createDiscoverDaemonSet(namespace, discoverImage, securityAccount) + err := d.createDiscoverDaemonSet(namespace, discoverImage, securityAccount, useCephVolume) if err != nil { return fmt.Errorf("Error starting discover daemonset: %v", err) } return nil } -func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAccount string) error { +func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAccount string, useCephVolume bool) error { privileged := true + discovery_parameters := []string{"discover", + "--discover-interval", getEnvVar(discoverIntervalEnv, defaultDiscoverInterval)} + if useCephVolume { + discovery_parameters = append(discovery_parameters, "--use-ceph-volume") + } + ds := &apps.DaemonSet{ ObjectMeta: metav1.ObjectMeta{ Name: discoverDaemonsetName, @@ -102,7 +109,7 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc { Name: discoverDaemonsetName, Image: discoverImage, - Args: []string{"discover", "--discover-interval", getDiscoverInterval()}, + Args: discovery_parameters, SecurityContext: &v1.SecurityContext{ Privileged: &privileged, }, @@ -211,13 +218,12 @@ func (d *Discover) createDiscoverDaemonSet(namespace, discoverImage, securityAcc } -func getDiscoverInterval() string { - discoverValue := os.Getenv(discoverIntervalEnv) - if discoverValue != "" { - return discoverValue +func getEnvVar(varName string, defaultValue string) string { + envValue := os.Getenv(varName) + if envValue != "" { + return envValue } - // Default is 60 minutes - return "60m" + return defaultValue } // ListDevices lists all devices discovered on all nodes or specific node if node name is provided. diff --git a/pkg/operator/discover/discover_test.go b/pkg/operator/discover/discover_test.go index 9c7df596a..2d9f6e278 100644 --- a/pkg/operator/discover/discover_test.go +++ b/pkg/operator/discover/discover_test.go @@ -46,7 +46,7 @@ func TestStartDiscoveryDaemonset(t *testing.T) { a := New(clientset) // start a basic cluster - err := a.Start(namespace, "rook/rook:myversion", "mysa") + err := a.Start(namespace, "rook/rook:myversion", "mysa", false) assert.Nil(t, err) // check daemonset parameters diff --git a/pkg/operator/edgefs/operator.go b/pkg/operator/edgefs/operator.go index 2a3675d93..3b197508d 100644 --- a/pkg/operator/edgefs/operator.go +++ b/pkg/operator/edgefs/operator.go @@ -29,7 +29,7 @@ import ( "github.com/rook/rook/pkg/operator/discover" "github.com/rook/rook/pkg/operator/edgefs/cluster" "github.com/rook/rook/pkg/operator/k8sutil" - "k8s.io/api/core/v1" + v1 "k8s.io/api/core/v1" ) var logger = capnslog.NewPackageLogger("github.com/rook/rook", "edgefs-operator") @@ -65,7 +65,7 @@ func (o *Operator) Run() error { } rookDiscover := discover.New(o.context.Clientset) - if err := rookDiscover.Start(namespace, o.rookImage, o.securityAccount); err != nil { + if err := rookDiscover.Start(namespace, o.rookImage, o.securityAccount, false); err != nil { return fmt.Errorf("Error starting device discovery daemonset: %v", err) } diff --git a/pkg/util/sys/device.go b/pkg/util/sys/device.go index 4c7857dca..638c421df 100644 --- a/pkg/util/sys/device.go +++ b/pkg/util/sys/device.go @@ -79,6 +79,8 @@ type LocalDisk struct { WWNVendorExtension string `json:"wwnVendorExtension"` // Empty checks whether the device is completely empty Empty bool `json:"empty"` + // Information provided by Ceph Volume Inventory + CephVolumeData string `json:"cephVolumeData,omitempty"` } func ListDevices(executor exec.Executor) ([]string, error) {