ceph: Get <ceph-volume inventory> data in dev. configmaps

**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 <cephVolumeData> field contains all the information returned
from <ceph-volume inventory> command.

Signed-off-by: Juan Miguel Olmo Martínez <jolmomar@redhat.com>

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)
 	}

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 {

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 <ceph-volume inventory>
+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
+}

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)
+}

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)
 		}
 	}

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.

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

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)
 	}
 

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) {