forked from rook/rook
Merge pull request #318 from travisn/simplify-inventory
Devices names are treated as ephemeral and not persisted to etcd
This commit is contained in:
@@ -1,79 +0,0 @@
|
||||
/*
|
||||
Copyright 2016 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 api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
|
||||
"github.com/rook/rook/pkg/cephmgr/osd"
|
||||
)
|
||||
|
||||
// Adds the device and configures an OSD on the device.
|
||||
// POST
|
||||
// /device
|
||||
func (h *Handler) AddDevice(w http.ResponseWriter, r *http.Request) {
|
||||
device, ok := handleLoadDeviceFromBody(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
err := osd.AddDesiredDevice(h.context.EtcdClient, device.Name, device.NodeID)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to add device. %v", err)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// Stops the OSD and removes a device from participating in the cluster.
|
||||
// POST
|
||||
// /device/remove
|
||||
func (h *Handler) RemoveDevice(w http.ResponseWriter, r *http.Request) {
|
||||
device, ok := handleLoadDeviceFromBody(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
err := osd.RemoveDesiredDevice(h.context.EtcdClient, device.Name, device.NodeID)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to remove device. %v", err)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func handleLoadDeviceFromBody(w http.ResponseWriter, r *http.Request) (*osd.Device, bool) {
|
||||
body, ok := handleReadBody(w, r, "load device")
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
var device osd.Device
|
||||
if err := json.Unmarshal(body, &device); err != nil {
|
||||
logger.Errorf("failed to unmarshal request body '%s': %+v", string(body), err)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
if device.Name == "" || device.NodeID == "" {
|
||||
logger.Errorf("missing name or nodeId: %+v", device)
|
||||
w.WriteHeader(http.StatusBadRequest)
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return &device, true
|
||||
}
|
||||
+13
-52
@@ -16,6 +16,7 @@ limitations under the License.
|
||||
package api
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -40,50 +41,8 @@ const (
|
||||
SuccessGetPoolECPool1Response = `{"pool":"ecPool1","pool_id":1,"size":3}{"pool":"ecPool1","pool_id":1,"min_size":3}{"pool":"ecPool1","pool_id":1,"crash_replay_interval":0}{"pool":"ecPool1","pool_id":1,"pg_num":100}{"pool":"ecPool1","pool_id":1,"pgp_num":100}{"pool":"ecPool1","pool_id":1,"crush_ruleset":1}{"pool":"ecPool1","pool_id":1,"hashpspool":"true"}{"pool":"ecPool1","pool_id":1,"nodelete":"false"}{"pool":"ecPool1","pool_id":1,"nopgchange":"false"}{"pool":"ecPool1","pool_id":1,"nosizechange":"false"}{"pool":"ecPool1","pool_id":1,"write_fadvise_dontneed":"false"}{"pool":"ecPool1","pool_id":1,"noscrub":"false"}{"pool":"ecPool1","pool_id":1,"nodeep-scrub":"false"}{"pool":"ecPool1","pool_id":1,"use_gmt_hitset":true}{"pool":"ecPool1","pool_id":1,"auid":0}{"pool":"ecPool1","pool_id":1,"erasure_code_profile":"ecPool1_ecprofile"}{"pool":"ecPool1","pool_id":1,"min_write_recency_for_promote":0}{"pool":"ecPool1","pool_id":1,"fast_read":0}{"pool":"ecPool1","pool_id":1}{"pool":"ecPool1","pool_id":1}{"pool":"ecPool1","pool_id":1}{"pool":"ecPool1","pool_id":1}{"pool":"ecPool1","pool_id":1}{"pool":"ecPool1","pool_id":1}`
|
||||
)
|
||||
|
||||
func TestAddRemoveDeviceHandler(t *testing.T) {
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
context := &clusterd.Context{EtcdClient: etcdClient}
|
||||
etcdClient.SetValue("/rook/nodes/config/123/disks/foo/uuid", "12345")
|
||||
|
||||
req, err := http.NewRequest("POST", "http://10.0.0.100/device", strings.NewReader(`{"name":"foo"}`))
|
||||
assert.Nil(t, err)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
h := NewHandler(context, nil, nil)
|
||||
|
||||
// missing the node id
|
||||
h.AddDevice(w, req)
|
||||
assert.Equal(t, http.StatusBadRequest, w.Code)
|
||||
|
||||
// successful request
|
||||
req, err = http.NewRequest("POST", "http://10.0.0.100/device", strings.NewReader(`{"name":"foo","nodeId":"123"}`))
|
||||
assert.Nil(t, err)
|
||||
h = NewHandler(context, nil, nil)
|
||||
w = httptest.NewRecorder()
|
||||
h.AddDevice(w, req)
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
|
||||
devices := etcdClient.GetChildDirs("/rook/services/ceph/osd/desired/123/device")
|
||||
assert.Equal(t, 1, devices.Count())
|
||||
assert.True(t, devices.Contains("foo"))
|
||||
|
||||
// remove the device
|
||||
req, err = http.NewRequest("POST", "http://10.0.0.100/device/remove", strings.NewReader(`{"name":"foo","nodeId":"123"}`))
|
||||
assert.Nil(t, err)
|
||||
h = NewHandler(context, nil, nil)
|
||||
w = httptest.NewRecorder()
|
||||
h.RemoveDevice(w, req)
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
|
||||
// check the desired state
|
||||
devices = etcdClient.GetChildDirs("/rook/services/ceph/osd/desired/123/device")
|
||||
assert.Equal(t, 0, devices.Count())
|
||||
}
|
||||
|
||||
func TestRemoveDeviceHandler(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGetNodesHandler(t *testing.T) {
|
||||
nodeID := "node1"
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
context := &clusterd.Context{EtcdClient: etcdClient}
|
||||
|
||||
@@ -102,16 +61,18 @@ func TestGetNodesHandler(t *testing.T) {
|
||||
assert.Equal(t, http.StatusOK, w.Code)
|
||||
assert.Equal(t, `[]`, w.Body.String())
|
||||
|
||||
// add the disks to etcd
|
||||
disks := []inventory.Disk{
|
||||
inventory.Disk{Type: inventory.DiskType, Size: 50, Rotational: true, Available: true},
|
||||
inventory.Disk{Type: inventory.DiskType, Size: 100, Rotational: false, Available: true},
|
||||
}
|
||||
output, _ := json.Marshal(disks)
|
||||
etcdClient.SetValue(path.Join(inventory.NodesConfigKey, nodeID, "disks"), string(output))
|
||||
|
||||
// set up a discovered node in etcd
|
||||
inventory.SetIPAddress(etcdClient, "node1", "1.2.3.4", "10.0.0.11")
|
||||
inventory.SetLocation(etcdClient, "node1", "root=default,dc=datacenter1")
|
||||
nodeConfigKey := path.Join(inventory.NodesConfigKey, "node1")
|
||||
etcdClient.CreateDir(nodeConfigKey)
|
||||
inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "sda", "123d4869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
100, true, false, "btrfs", "/mnt/abc", inventory.Disk, "", false)
|
||||
inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "sdb", "321d4869-29ee-4bfd-bf21-dfd597bdffff",
|
||||
50, false, false, "ext4", "/mnt/def", inventory.Disk, "", false)
|
||||
appliedOSDKey := "/rook/services/ceph/osd/applied/node1"
|
||||
inventory.SetIPAddress(etcdClient, nodeID, "1.2.3.4", "10.0.0.11")
|
||||
inventory.SetLocation(etcdClient, nodeID, "root=default,dc=datacenter1")
|
||||
appliedOSDKey := path.Join("/rook/services/ceph/osd/applied", nodeID)
|
||||
etcdClient.SetValue(path.Join(appliedOSDKey, "12", "disk-uuid"), "123d4869-29ee-4bfd-bf21-dfd597bd222e")
|
||||
etcdClient.SetValue(path.Join(appliedOSDKey, "13", "disk-uuid"), "321d4869-29ee-4bfd-bf21-dfd597bdffff")
|
||||
|
||||
|
||||
+3
-15
@@ -19,7 +19,6 @@ import (
|
||||
"net/http"
|
||||
|
||||
"github.com/rook/rook/pkg/cephmgr/mon"
|
||||
"github.com/rook/rook/pkg/cephmgr/osd"
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
"github.com/rook/rook/pkg/clusterd/inventory"
|
||||
"github.com/rook/rook/pkg/model"
|
||||
@@ -46,22 +45,11 @@ func (h *Handler) GetNodes(w http.ResponseWriter, r *http.Request) {
|
||||
nodes := make([]model.Node, len(clusterInventory.Nodes))
|
||||
i := 0
|
||||
for nodeID, n := range clusterInventory.Nodes {
|
||||
// look up all the disks that the current node has applied OSDs on
|
||||
appliedIDs, err := osd.GetAppliedOSDs(nodeID, h.context.EtcdClient)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to get applied OSDs for node %s: %+v", nodeID, err)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
storage := uint64(0)
|
||||
for _, d := range n.Disks {
|
||||
for _, uuid := range appliedIDs {
|
||||
if d.UUID == uuid {
|
||||
// current disk is in applied OSD set, add its storage to the running total
|
||||
storage += d.Size
|
||||
}
|
||||
}
|
||||
// Add up the space of all devices.
|
||||
// We should have a separate metric for osd devices, but keep it simple for now.
|
||||
storage += d.Size
|
||||
}
|
||||
|
||||
// determine the node's state/health
|
||||
|
||||
@@ -107,18 +107,6 @@ func (h *Handler) GetRoutes() []Route {
|
||||
"/filesystem/remove",
|
||||
h.RemoveFileSystem,
|
||||
},
|
||||
{
|
||||
"AddDevice",
|
||||
"POST",
|
||||
"/device",
|
||||
h.AddDevice,
|
||||
},
|
||||
{
|
||||
"RemoveDevice",
|
||||
"POST",
|
||||
"/device/remove",
|
||||
h.RemoveDevice,
|
||||
},
|
||||
{
|
||||
"SetLogLevel",
|
||||
"POST",
|
||||
|
||||
+61
-41
@@ -48,14 +48,15 @@ const (
|
||||
)
|
||||
|
||||
type osdAgent struct {
|
||||
cluster *mon.ClusterInfo
|
||||
forceFormat bool
|
||||
location string
|
||||
factory client.ConnectionFactory
|
||||
osdProc map[int]*proc.MonitoredProc
|
||||
devices string
|
||||
configCounter int32
|
||||
osdsCompleted chan struct{}
|
||||
cluster *mon.ClusterInfo
|
||||
forceFormat bool
|
||||
location string
|
||||
factory client.ConnectionFactory
|
||||
osdProc map[int]*proc.MonitoredProc
|
||||
desiredDevices []string
|
||||
devices string
|
||||
configCounter int32
|
||||
osdsCompleted chan struct{}
|
||||
}
|
||||
|
||||
func NewAgent(factory client.ConnectionFactory, devices string, forceFormat bool, location string) *osdAgent {
|
||||
@@ -72,19 +73,13 @@ func (a *osdAgent) Initialize(context *clusterd.Context) error {
|
||||
|
||||
if len(a.devices) > 0 {
|
||||
// add the devices to desired state
|
||||
devices := strings.Split(a.devices, ",")
|
||||
for _, device := range devices {
|
||||
logger.Infof("Adding device %s to desired state", device)
|
||||
err := AddDesiredDevice(context.EtcdClient, device, context.NodeID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to add desired device %s. %v", device, err)
|
||||
}
|
||||
}
|
||||
a.desiredDevices = strings.Split(a.devices, ",")
|
||||
logger.Infof("desired devices for osds: %+v", a.desiredDevices)
|
||||
}
|
||||
|
||||
// if no devices or directories were specified, use the current directory for an osd
|
||||
if len(a.devices) == 0 {
|
||||
logger.Infof("Adding local path to local directory %s", context.ConfigDir)
|
||||
if len(a.desiredDevices) == 0 {
|
||||
logger.Infof("Adding local path %s to desired state", context.ConfigDir)
|
||||
err := AddDesiredDir(context.EtcdClient, context.ConfigDir, context.NodeID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to add current dir %s. %v", context.ConfigDir, err)
|
||||
@@ -180,7 +175,7 @@ func (a *osdAgent) decrementConfigCounter() {
|
||||
}
|
||||
|
||||
func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection client.Connection) error {
|
||||
desiredDevices, err := loadDesiredDevices(context.EtcdClient, context.NodeID)
|
||||
desiredDevices, err := a.loadDesiredDevices(context)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to load desired devices. %v", err)
|
||||
}
|
||||
@@ -277,7 +272,7 @@ func getAppliedKey(nodeID string) string {
|
||||
|
||||
// create and initalize OSDs for all the devices specified in the given config
|
||||
func (a *osdAgent) createDesiredOSDs(adminConn client.Connection, context *clusterd.Context) error {
|
||||
devices, err := loadDesiredDevices(context.EtcdClient, context.NodeID)
|
||||
devices, err := a.loadDesiredDevices(context)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to load desired devices. %v", err)
|
||||
}
|
||||
@@ -355,6 +350,9 @@ func (a *osdAgent) configureDevices(context *clusterd.Context, devices map[strin
|
||||
go func() {
|
||||
defer connection.Shutdown()
|
||||
|
||||
// set the signal that the osd config is completed
|
||||
defer close(a.osdsCompleted)
|
||||
|
||||
a.incrementConfigCounter()
|
||||
defer a.decrementConfigCounter()
|
||||
|
||||
@@ -371,9 +369,6 @@ func (a *osdAgent) configureDevices(context *clusterd.Context, devices map[strin
|
||||
}
|
||||
|
||||
logger.Infof("%d/%d bluestore osds succeeded on this node", succeeded, len(devices))
|
||||
|
||||
// set the signal that the osd config is completed
|
||||
close(a.osdsCompleted)
|
||||
}()
|
||||
|
||||
return nil
|
||||
@@ -387,13 +382,12 @@ func (a *osdAgent) createOrStartOSD(context *clusterd.Context, connection client
|
||||
return err
|
||||
}
|
||||
|
||||
name := config.deviceName
|
||||
// set the desired state of the dir with the osd id. If a device, we will delay setting this until we have the device uuid
|
||||
if dir {
|
||||
name = configRoot
|
||||
}
|
||||
err = setOSDOnDevice(context.EtcdClient, context.NodeID, name, config.id, dir)
|
||||
if err != nil {
|
||||
return err
|
||||
err = setOSDOnDevice(context.EtcdClient, context.NodeID, configRoot, config.id, dir)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to associate osd id %d with the data dir", config.id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -537,10 +531,29 @@ func getPseudoDir(path string) string {
|
||||
return strings.Replace(path, "/", "_", -1)
|
||||
}
|
||||
|
||||
func loadDesiredDevices(etcdClient etcd.KeysAPI, nodeID string) (map[string]int, error) {
|
||||
func (a *osdAgent) loadDesiredDevices(context *clusterd.Context) (map[string]int, error) {
|
||||
// get the device UUID to device name mapping
|
||||
uuidToName := map[string]string{}
|
||||
for _, disk := range context.Inventory.Local.Disks {
|
||||
if disk.UUID != "" {
|
||||
uuidToName[disk.UUID] = disk.Name
|
||||
}
|
||||
}
|
||||
logger.Debugf("uuid to name map: %+v", uuidToName)
|
||||
|
||||
// ensure all the desired devices are in the list
|
||||
devices := map[string]int{}
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID))
|
||||
devKeys, err := etcdClient.Get(ctx.Background(), key, &etcd.GetOptions{Recursive: true})
|
||||
for _, name := range a.desiredDevices {
|
||||
if _, ok := devices[name]; !ok {
|
||||
// add the device to the desired list
|
||||
devices[name] = unassignedOSDID
|
||||
}
|
||||
}
|
||||
logger.Debugf("desired osd id mapping when new: %+v", devices)
|
||||
|
||||
// parse the desired devices from etcd, which are based on the disk uuid.
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, context.NodeID))
|
||||
devKeys, err := context.EtcdClient.Get(ctx.Background(), key, &etcd.GetOptions{Recursive: true})
|
||||
if err != nil {
|
||||
if util.IsEtcdKeyNotFound(err) {
|
||||
return devices, nil
|
||||
@@ -548,23 +561,30 @@ func loadDesiredDevices(etcdClient etcd.KeysAPI, nodeID string) (map[string]int,
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// parse the dirs from etcd
|
||||
for _, dev := range devKeys.Node.Nodes {
|
||||
device := util.GetLeafKeyPath(dev.Key)
|
||||
uuid := util.GetLeafKeyPath(dev.Key)
|
||||
osdID := unassignedOSDID
|
||||
|
||||
for _, setting := range dev.Nodes {
|
||||
if strings.HasSuffix(setting.Key, "/osd-id") {
|
||||
id, err := strconv.Atoi(setting.Value)
|
||||
if err == nil {
|
||||
logger.Debugf("found osd id %d for disk uuid %s", osdID, uuid)
|
||||
osdID = id
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
devices[device] = osdID
|
||||
// translate the disk uuid to the device name
|
||||
if name, ok := uuidToName[uuid]; ok {
|
||||
devices[name] = osdID
|
||||
} else {
|
||||
logger.Warningf("did not find name for disk uuid %s", uuid)
|
||||
}
|
||||
}
|
||||
|
||||
logger.Debugf("final osd id mapping: %+v", devices)
|
||||
|
||||
return devices, nil
|
||||
}
|
||||
|
||||
@@ -585,11 +605,11 @@ func setOSDOnDevice(etcdClient etcd.KeysAPI, nodeID, name string, id int, dir bo
|
||||
}
|
||||
|
||||
// add a device to the desired state
|
||||
func AddDesiredDevice(etcdClient etcd.KeysAPI, device, nodeID string) error {
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), device)
|
||||
func AddDesiredDevice(etcdClient etcd.KeysAPI, nodeID, uuid string, osdID int) error {
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), uuid)
|
||||
err := util.CreateEtcdDir(etcdClient, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to add device %s on node %s to desired. %v", device, nodeID, err)
|
||||
return fmt.Errorf("failed to add device %s on node %s to desired. %v", uuid, nodeID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -641,12 +661,12 @@ func AddDesiredDir(etcdClient etcd.KeysAPI, dir, nodeID string) error {
|
||||
}
|
||||
|
||||
// remove a device from the desired state
|
||||
func RemoveDesiredDevice(etcdClient etcd.KeysAPI, device, nodeID string) error {
|
||||
func RemoveDesiredDevice(etcdClient etcd.KeysAPI, nodeID, uuid string) error {
|
||||
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), device)
|
||||
key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), uuid)
|
||||
_, err := etcdClient.Delete(ctx.Background(), key, &etcd.DeleteOptions{Dir: true})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to remove device %s on node %s from desired. %v", device, nodeID, err)
|
||||
return fmt.Errorf("failed to remove device uuid %s on node %s from desired. %v", uuid, nodeID, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -93,18 +93,16 @@ func TestOSDAgentWithDevices(t *testing.T) {
|
||||
NodeID: nodeID,
|
||||
ConfigDir: "/tmp",
|
||||
ProcMan: proc.New(executor),
|
||||
Inventory: createInventory(),
|
||||
}
|
||||
context.Inventory.Local.Disks = []inventory.LocalDisk{
|
||||
inventory.LocalDisk{Name: "sdx", Size: 1234567890, UUID: "12345"},
|
||||
inventory.LocalDisk{Name: "sdy", Size: 2234567890, UUID: "54321"},
|
||||
}
|
||||
|
||||
// prep the etcd keys that would have been discovered by inventory
|
||||
disksKey := path.Join(inventory.GetNodeConfigKey(context.NodeID), "disks")
|
||||
etcdClient.SetValue(path.Join(disksKey, "sdx", "uuid"), "12345")
|
||||
etcdClient.SetValue(path.Join(disksKey, "sdy", "uuid"), "54321")
|
||||
etcdClient.SetValue(path.Join(disksKey, "sdx", "size"), "1234567890")
|
||||
etcdClient.SetValue(path.Join(disksKey, "sdy", "size"), "2234567890")
|
||||
|
||||
// Set one device as already having an assigned osd id. The other device will go through id selection,
|
||||
// which is mocked in the createTestAgent method to return an id of 3.
|
||||
etcdClient.SetValue(fmt.Sprintf("/rook/services/ceph/osd/desired/%s/device/sdx/osd-id", nodeID), "23")
|
||||
etcdClient.SetValue(fmt.Sprintf("/rook/services/ceph/osd/desired/%s/device/12345/osd-id", nodeID), "23")
|
||||
|
||||
// prep the OSD agent and related orcehstration data
|
||||
prepAgentOrchestrationData(t, agent, etcdClient, context, clusterName)
|
||||
@@ -168,6 +166,7 @@ func TestOSDAgentNoDevices(t *testing.T) {
|
||||
NodeID: nodeID,
|
||||
ProcMan: proc.New(executor),
|
||||
ConfigDir: "/tmp",
|
||||
Inventory: createInventory(),
|
||||
}
|
||||
|
||||
// prep the OSD agent and related orcehstration data
|
||||
@@ -206,12 +205,6 @@ func TestAppliedDevices(t *testing.T) {
|
||||
assert.Equal(t, 0, len(osds))
|
||||
|
||||
// two applied osds
|
||||
nodeConfigKey := path.Join(inventory.NodesConfigKey, nodeID)
|
||||
etcdClient.CreateDir(nodeConfigKey)
|
||||
inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "sda", "ff6d4869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
100, true, false, "btrfs", "/mnt/xyz", inventory.Disk, "", false)
|
||||
inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "sdb", "ff6d4869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
50, false, false, "ext4", "/mnt/zyx", inventory.Disk, "", false)
|
||||
appliedOSDKey := "/rook/services/ceph/osd/applied/abc"
|
||||
etcdClient.SetValue(path.Join(appliedOSDKey, "1", "disk-uuid"), "1234")
|
||||
etcdClient.SetValue(path.Join(appliedOSDKey, "2", "disk-uuid"), "2345")
|
||||
@@ -228,8 +221,9 @@ func TestRemoveDevice(t *testing.T) {
|
||||
etcdClient, agent, conn := createTestAgent(t, nodeID, "")
|
||||
executor := &exectest.MockExecutor{}
|
||||
|
||||
context := &clusterd.Context{EtcdClient: etcdClient, NodeID: nodeID, Executor: executor, ProcMan: proc.New(executor)}
|
||||
etcdClient.SetValue("/rook/services/ceph/osd/desired/a/device/sda/osd-id", "23")
|
||||
context := &clusterd.Context{EtcdClient: etcdClient, NodeID: nodeID, Executor: executor, ProcMan: proc.New(executor), Inventory: createInventory()}
|
||||
context.Inventory.Local.Disks = []inventory.LocalDisk{inventory.LocalDisk{Name: "sda", Size: 1234567890, UUID: "5435435333"}}
|
||||
etcdClient.SetValue("/rook/services/ceph/osd/desired/a/device/5435435333/osd-id", "23")
|
||||
|
||||
// create two applied osds, one of which is desired
|
||||
appliedRoot := "/rook/services/ceph/osd/applied/" + nodeID
|
||||
@@ -298,27 +292,67 @@ func createTestKeyring(t *testing.T, args []string) {
|
||||
}
|
||||
|
||||
func TestDesiredDeviceState(t *testing.T) {
|
||||
nodeID := "a"
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
etcdClient.CreateDir("/rook/nodes/config/a/disks/foo")
|
||||
|
||||
// add a device
|
||||
err := AddDesiredDevice(etcdClient, "foo", "a")
|
||||
err := AddDesiredDevice(etcdClient, nodeID, "myuuid", 23)
|
||||
assert.Nil(t, err)
|
||||
devices := etcdClient.GetChildDirs("/rook/services/ceph/osd/desired/a/device")
|
||||
assert.Equal(t, 1, devices.Count())
|
||||
assert.True(t, devices.Contains("foo"))
|
||||
assert.True(t, devices.Contains("myuuid"))
|
||||
|
||||
// remove the device
|
||||
err = RemoveDesiredDevice(etcdClient, "foo", "a")
|
||||
err = RemoveDesiredDevice(etcdClient, nodeID, "myuuid")
|
||||
assert.Nil(t, err)
|
||||
devices = etcdClient.GetChildDirs("/rook/services/ceph/osd/desired/a/device")
|
||||
assert.Equal(t, 0, devices.Count())
|
||||
|
||||
// removing a non-existent device is a no-op
|
||||
err = RemoveDesiredDevice(etcdClient, "foo", "a")
|
||||
err = RemoveDesiredDevice(etcdClient, nodeID, "foo")
|
||||
assert.Nil(t, err)
|
||||
}
|
||||
|
||||
func TestLoadDesiredDevices(t *testing.T) {
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
a := &osdAgent{desiredDevices: []string{}}
|
||||
|
||||
// no devices are desired
|
||||
context := &clusterd.Context{EtcdClient: etcdClient, Inventory: createInventory(), NodeID: "a"}
|
||||
desired, err := a.loadDesiredDevices(context)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 0, len(desired))
|
||||
|
||||
// two devices are desired and it is a new config
|
||||
context.Inventory.Local.Disks = []inventory.LocalDisk{
|
||||
inventory.LocalDisk{Name: "sda", Size: 1234567890, UUID: "12345"},
|
||||
inventory.LocalDisk{Name: "sdb", Size: 2234567890, UUID: "54321"},
|
||||
}
|
||||
a.desiredDevices = []string{"sda", "sdb"}
|
||||
desired, err = a.loadDesiredDevices(context)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 2, len(desired))
|
||||
assert.Equal(t, -1, desired["sda"])
|
||||
assert.Equal(t, -1, desired["sdb"])
|
||||
|
||||
// two devices are desired and they have previously been configured
|
||||
etcdClient.SetValue("/rook/services/ceph/osd/desired/a/device/12345/osd-id", "23")
|
||||
etcdClient.SetValue("/rook/services/ceph/osd/desired/a/device/54321/osd-id", "24")
|
||||
desired, err = a.loadDesiredDevices(context)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 2, len(desired))
|
||||
assert.Equal(t, 23, desired["sda"])
|
||||
assert.Equal(t, 24, desired["sdb"])
|
||||
|
||||
// no devices are desired and they have previously been configured
|
||||
a.desiredDevices = []string{}
|
||||
desired, err = a.loadDesiredDevices(context)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 2, len(desired))
|
||||
assert.Equal(t, 23, desired["sda"])
|
||||
assert.Equal(t, 24, desired["sdb"])
|
||||
}
|
||||
|
||||
func TestDesiredDirsState(t *testing.T) {
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
|
||||
@@ -336,3 +370,7 @@ func TestDesiredDirsState(t *testing.T) {
|
||||
assert.Equal(t, 1, len(loadedDirs))
|
||||
assert.Equal(t, unassignedOSDID, loadedDirs["/my/dir"])
|
||||
}
|
||||
|
||||
func createInventory() *inventory.Config {
|
||||
return &inventory.Config{Local: &inventory.Hardware{Disks: []inventory.LocalDisk{}}}
|
||||
}
|
||||
|
||||
@@ -99,9 +99,9 @@ func formatDevice(context *clusterd.Context, config *osdConfig, forceFormat bool
|
||||
// Partitions a device for use by a bluestore osd.
|
||||
// If there are any partitions or formatting already on the device, it will be wiped.
|
||||
func partitionBluestoreDevice(context *clusterd.Context, config *osdConfig) error {
|
||||
size, err := inventory.GetDeviceSize(config.deviceName, context.NodeID, context.EtcdClient)
|
||||
size, err := getDiskSize(context, config.deviceName)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get device %s size. %+v", config.deviceName, err)
|
||||
return fmt.Errorf("failed to get device size for osd %d. %+v", config.id, err)
|
||||
}
|
||||
|
||||
cmd := fmt.Sprintf("zap %s", config.deviceName)
|
||||
@@ -110,7 +110,7 @@ func partitionBluestoreDevice(context *clusterd.Context, config *osdConfig) erro
|
||||
return fmt.Errorf("failed to zap partitions on /dev/%s: %+v", config.deviceName, err)
|
||||
}
|
||||
|
||||
scheme, err := partition.GetSimpleScheme(int(size / 1024 / 1024))
|
||||
scheme, err := partition.GetSimpleScheme(config.id, int(size/1024/1024))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get simple scheme. %+v", err)
|
||||
}
|
||||
@@ -129,20 +129,31 @@ func partitionBluestoreDevice(context *clusterd.Context, config *osdConfig) erro
|
||||
return fmt.Errorf("failed to partition /dev/%s. %+v", config.deviceName, err)
|
||||
}
|
||||
|
||||
err = inventory.SetDeviceUUID(context.NodeID, config.deviceName, scheme.DiskUUID, context.EtcdClient)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to set uuid %s. %+v", scheme.DiskUUID, err)
|
||||
}
|
||||
|
||||
// save the scheme
|
||||
err = scheme.Save(config.rootPath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to save partition scheme. %+v", err)
|
||||
}
|
||||
|
||||
// save the desired state of the osd for this device
|
||||
err = setOSDOnDevice(context.EtcdClient, context.NodeID, scheme.DiskUUID, config.id, false)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to associate osd id %d with device %s (%s)", config.id, config.deviceName, scheme.DiskUUID)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getDiskSize(context *clusterd.Context, name string) (uint64, error) {
|
||||
for _, device := range context.Inventory.Local.Disks {
|
||||
if device.Name == name {
|
||||
return device.Size, nil
|
||||
}
|
||||
}
|
||||
|
||||
return 0, fmt.Errorf("device %s not found", name)
|
||||
}
|
||||
|
||||
func registerOSD(bootstrapConn client.Connection, config *osdConfig) error {
|
||||
var err error
|
||||
config.uuid, err = uuid.NewRandom()
|
||||
|
||||
@@ -39,13 +39,14 @@ const (
|
||||
type Scheme struct {
|
||||
Version int `json:"version"`
|
||||
SizeMB int `json:"sizeMb"`
|
||||
ID int `json:"id"`
|
||||
DiskUUID string `json:"diskUuid"`
|
||||
PartitionUUIDs map[string]string `json:"partitionUuids"`
|
||||
}
|
||||
|
||||
// This is a simple scheme to create the wal and db each of size 10% of the disk,
|
||||
// with the remainder of the disk being allocated for the raw data.
|
||||
func GetSimpleScheme(sizeMB int) (*Scheme, error) {
|
||||
func GetSimpleScheme(id, sizeMB int) (*Scheme, error) {
|
||||
diskUUID, err := uuid.NewRandom()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to get disk uuid. %+v", err)
|
||||
@@ -72,6 +73,7 @@ func GetSimpleScheme(sizeMB int) (*Scheme, error) {
|
||||
return &Scheme{
|
||||
Version: SimpleVersion,
|
||||
SizeMB: sizeMB,
|
||||
ID: id,
|
||||
DiskUUID: diskUUID.String(),
|
||||
PartitionUUIDs: uuids,
|
||||
}, nil
|
||||
@@ -87,11 +89,11 @@ func (s *Scheme) GetArgs(name string) []string {
|
||||
tenPercent := s.SizeMB / 10
|
||||
|
||||
// append args for individual partitions
|
||||
args := s.getPartitionArgs(WalPartitionName, walPartition, offset, tenPercent, fmt.Sprintf("osd-wal-%s", s.DiskUUID))
|
||||
args := s.getPartitionArgs(WalPartitionName, walPartition, offset, tenPercent, fmt.Sprintf("ROOK-OSD%d-WAL", s.ID))
|
||||
offset += tenPercent
|
||||
args = append(args, s.getPartitionArgs(DatabasePartitionName, databasePartition, offset, tenPercent, fmt.Sprintf("osd-db-%s", s.DiskUUID))...)
|
||||
args = append(args, s.getPartitionArgs(DatabasePartitionName, databasePartition, offset, tenPercent, fmt.Sprintf("ROOK-OSD%d-DB", s.ID))...)
|
||||
offset += tenPercent
|
||||
args = append(args, s.getPartitionArgs(BlockPartitionName, blockPartition, offset, UseRemainingSpace, fmt.Sprintf("osd-block-%s", s.DiskUUID))...)
|
||||
args = append(args, s.getPartitionArgs(BlockPartitionName, blockPartition, offset, UseRemainingSpace, fmt.Sprintf("ROOK-OSD%d-BLOCK", s.ID))...)
|
||||
|
||||
// append args for the whole device
|
||||
args = append(args, fmt.Sprintf("--disk-guid=%s", s.DiskUUID))
|
||||
|
||||
@@ -24,11 +24,12 @@ import (
|
||||
)
|
||||
|
||||
func TestSimpleScheme(t *testing.T) {
|
||||
scheme, err := GetSimpleScheme(123)
|
||||
scheme, err := GetSimpleScheme(1, 123)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 123, scheme.SizeMB)
|
||||
assert.Equal(t, 36, len(scheme.DiskUUID))
|
||||
assert.Equal(t, 3, len(scheme.PartitionUUIDs))
|
||||
assert.Equal(t, 1, scheme.ID)
|
||||
|
||||
args := scheme.GetArgs("foo")
|
||||
assert.Equal(t, 11, len(args))
|
||||
|
||||
@@ -90,12 +90,6 @@ func (r *ClusterMember) initialize() error {
|
||||
return fmt.Errorf("failed to elect leader. %+v", err)
|
||||
}
|
||||
|
||||
// initialize the hardware inventory
|
||||
err = r.discoverHardware()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to detect initial hardware. %+v", err)
|
||||
}
|
||||
|
||||
// in a goroutine, begin the monitor cluster loop for changes in membership, leadership, etc.
|
||||
go func() {
|
||||
r.refreshLeader()
|
||||
@@ -271,5 +265,11 @@ func (r *ClusterMember) discoverHardware() error {
|
||||
}
|
||||
|
||||
// Discover current state of the disks and other node properties
|
||||
return inventory.DiscoverHardware(r.context.NodeID, r.context.EtcdClient, r.context.Executor)
|
||||
var err error
|
||||
r.context.Inventory.Local, err = inventory.DiscoverHardware(r.context.EtcdClient, r.context.Executor, r.context.NodeID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to perform discovery. %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -589,7 +589,7 @@ func createDefaultDependencies() (*util.MockEtcdClient, *Context, *mockLeaseMana
|
||||
mockLeaseManager := &mockLeaseManager{}
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
machineId := "8e8f532fe96dcae6b1ce335822e5b03c"
|
||||
context := &Context{EtcdClient: etcdClient, NodeID: machineId, Executor: &exectest.MockExecutor{}}
|
||||
context := &Context{EtcdClient: etcdClient, NodeID: machineId, Executor: &exectest.MockExecutor{}, Inventory: &inventory.Config{}}
|
||||
return etcdClient, context, mockLeaseManager, &MockLeader{}
|
||||
}
|
||||
|
||||
|
||||
+72
-207
@@ -16,186 +16,87 @@ limitations under the License.
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
etcd "github.com/coreos/etcd/client"
|
||||
"github.com/rook/rook/pkg/util"
|
||||
ctx "golang.org/x/net/context"
|
||||
|
||||
"github.com/rook/rook/pkg/util/exec"
|
||||
"github.com/rook/rook/pkg/util/sys"
|
||||
ctx "golang.org/x/net/context"
|
||||
)
|
||||
|
||||
const (
|
||||
diskUUIDKey = "uuid"
|
||||
diskSizeKey = "size"
|
||||
diskRotationalKey = "rotational"
|
||||
diskReadonlyKey = "readonly"
|
||||
diskFileSystemKey = "filesystem"
|
||||
diskTypeKey = "type"
|
||||
diskParentKey = "parent"
|
||||
diskHasChildrenKey = "children"
|
||||
diskMountPointKey = "mountpoint"
|
||||
)
|
||||
func GetAvailableDevices(devices []LocalDisk) []string {
|
||||
|
||||
func DiskTypeToStr(diskType DiskType) string {
|
||||
switch diskType {
|
||||
case Disk:
|
||||
return "disk"
|
||||
case Part:
|
||||
return "part"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
}
|
||||
|
||||
func StrToDiskType(diskType string) (DiskType, error) {
|
||||
diskType = strings.ToLower(diskType)
|
||||
switch diskType {
|
||||
case "disk":
|
||||
return Disk, nil
|
||||
case "part":
|
||||
return Part, nil
|
||||
default:
|
||||
return -1, errors.New(fmt.Sprintf("unknown disk type: %s", diskType))
|
||||
}
|
||||
}
|
||||
|
||||
func GetDeviceSize(name, nodeID string, etcdClient etcd.KeysAPI) (uint64, error) {
|
||||
key := path.Join(GetNodeConfigKey(nodeID), disksKey, name, diskSizeKey)
|
||||
resp, err := etcdClient.Get(ctx.Background(), key, nil)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
size, err := strconv.ParseUint(resp.Node.Value, 10, 64)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
return size, nil
|
||||
}
|
||||
|
||||
func GetDeviceFromUUID(uuid, nodeID string, etcdClient etcd.KeysAPI) (string, error) {
|
||||
key := path.Join(GetNodeConfigKey(nodeID), disksKey)
|
||||
devices, err := util.GetDirChildKeys(etcdClient, key)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
for d := range devices.Iter() {
|
||||
resp, err := etcdClient.Get(ctx.Background(), path.Join(key, d, diskUUIDKey), nil)
|
||||
if err != nil || resp == nil || resp.Node == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
if resp.Node.Value == uuid {
|
||||
return d, nil
|
||||
var available []string
|
||||
for _, device := range devices {
|
||||
logger.Debugf("Evaluating device %+v", device)
|
||||
if getDeviceAvailable(device) {
|
||||
logger.Debugf("Available device: %s", device.Name)
|
||||
available = append(available, device.Name)
|
||||
}
|
||||
}
|
||||
|
||||
return "", fmt.Errorf("device for uuid %s not found", uuid)
|
||||
return available
|
||||
}
|
||||
|
||||
func SetDeviceUUID(nodeID, device, uuid string, etcdClient etcd.KeysAPI) error {
|
||||
key := path.Join(GetNodeConfigKey(nodeID), disksKey, device, diskUUIDKey)
|
||||
_, err := etcdClient.Set(ctx.Background(), key, uuid, nil)
|
||||
return err
|
||||
}
|
||||
|
||||
func GetDeviceUUID(device, nodeID string, etcdClient etcd.KeysAPI) (string, error) {
|
||||
key := path.Join(GetNodeConfigKey(nodeID), disksKey, device, diskUUIDKey)
|
||||
resp, err := etcdClient.Get(ctx.Background(), key, nil)
|
||||
func storeDevices(etcdClient etcd.KeysAPI, nodeID string, devices []LocalDisk) error {
|
||||
// store the basic device info in etcd
|
||||
disks := toClusterDisks(devices)
|
||||
output, err := json.Marshal(disks)
|
||||
if err != nil {
|
||||
return "", err
|
||||
return fmt.Errorf("failed to marshal disks. %+v", err)
|
||||
}
|
||||
|
||||
return resp.Node.Value, nil
|
||||
key := path.Join(NodesConfigKey, nodeID, disksKey)
|
||||
_, err = etcdClient.Set(ctx.Background(), key, string(output), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to store disks in etcd. %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getDiskInfo(diskInfo *etcd.Node) (*DiskConfig, error) {
|
||||
disk := &DiskConfig{}
|
||||
disk.Name = util.GetLeafKeyPath(diskInfo.Key)
|
||||
func loadDisksConfig(nodeConfig *NodeConfig, rawDisks string) error {
|
||||
var disks []Disk
|
||||
if err := json.Unmarshal([]byte(rawDisks), &disks); err != nil {
|
||||
return fmt.Errorf("failed to deserialize disks. %+v", err)
|
||||
}
|
||||
nodeConfig.Disks = disks
|
||||
return nil
|
||||
}
|
||||
|
||||
// iterate over all properties of the disk
|
||||
for _, diskProperty := range diskInfo.Nodes {
|
||||
diskPropertyName := util.GetLeafKeyPath(diskProperty.Key)
|
||||
switch diskPropertyName {
|
||||
case diskUUIDKey:
|
||||
disk.UUID = diskProperty.Value
|
||||
case diskSizeKey:
|
||||
size, err := strconv.ParseUint(diskProperty.Value, 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
disk.Size = size
|
||||
}
|
||||
case diskRotationalKey:
|
||||
rotational, err := strconv.ParseInt(diskProperty.Value, 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
disk.Rotational = itob(rotational)
|
||||
}
|
||||
case diskReadonlyKey:
|
||||
readonly, err := strconv.ParseInt(diskProperty.Value, 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
disk.Readonly = itob(readonly)
|
||||
}
|
||||
case diskFileSystemKey:
|
||||
disk.FileSystem = diskProperty.Value
|
||||
case diskMountPointKey:
|
||||
disk.MountPoint = diskProperty.Value
|
||||
case diskTypeKey:
|
||||
diskType, err := StrToDiskType(diskProperty.Value)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
disk.Type = diskType
|
||||
}
|
||||
case diskParentKey:
|
||||
disk.Parent = diskProperty.Value
|
||||
case diskHasChildrenKey:
|
||||
hasChildren, err := strconv.ParseInt(diskProperty.Value, 10, 64)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else {
|
||||
disk.HasChildren = itob(hasChildren)
|
||||
}
|
||||
default:
|
||||
logger.Warningf("unknown disk property key %s, skipping...", diskPropertyName)
|
||||
// Extract the basic disk info that will be used in cluster-wide orchestration decisions.
|
||||
func toClusterDisks(devices []LocalDisk) []Disk {
|
||||
var disks []Disk
|
||||
for _, device := range devices {
|
||||
if device.Type == DiskType || device.Type == SSDType {
|
||||
disk := Disk{
|
||||
Type: device.Type,
|
||||
Size: device.Size,
|
||||
Rotational: device.Rotational,
|
||||
Available: getDeviceAvailable(device)}
|
||||
disks = append(disks, disk)
|
||||
}
|
||||
}
|
||||
|
||||
return disk, nil
|
||||
return disks
|
||||
}
|
||||
|
||||
func btoi(b bool) int {
|
||||
if b {
|
||||
return 1
|
||||
} else {
|
||||
return 0
|
||||
}
|
||||
// check whether a device is available for use by bluestore
|
||||
func getDeviceAvailable(device LocalDisk) bool {
|
||||
return device.Parent == "" && device.Type == DiskType && device.FileSystem == ""
|
||||
}
|
||||
|
||||
func itob(i int64) bool {
|
||||
if i == 0 {
|
||||
return false
|
||||
} else {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func discoverDisks(nodeConfigKey string, etcdClient etcd.KeysAPI, executor exec.Executor) error {
|
||||
// Discover all the details of devices available on the local node
|
||||
func discoverDevices(executor exec.Executor) ([]LocalDisk, error) {
|
||||
|
||||
var disks []LocalDisk
|
||||
devices, err := sys.ListDevices(executor)
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, d := range devices {
|
||||
@@ -207,7 +108,7 @@ func discoverDisks(nodeConfigKey string, etcdClient etcd.KeysAPI, executor exec.
|
||||
}
|
||||
|
||||
diskType, ok := diskProps["TYPE"]
|
||||
if !ok || (diskType != "ssd" && diskType != "disk" && diskType != "part") {
|
||||
if !ok || (diskType != SSDType && diskType != DiskType && diskType != PartType) {
|
||||
// unsupported disk type, just continue
|
||||
continue
|
||||
}
|
||||
@@ -224,71 +125,35 @@ func discoverDisks(nodeConfigKey string, etcdClient etcd.KeysAPI, executor exec.
|
||||
|
||||
fs, err := sys.GetDeviceFilesystems(d, executor)
|
||||
if err != nil {
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
dkey := path.Join(nodeConfigKey, disksKey, d)
|
||||
disk := LocalDisk{Name: d, UUID: diskUUID, FileSystem: fs}
|
||||
|
||||
if _, err := etcdClient.Set(ctx.Background(), path.Join(dkey, diskUUIDKey), diskUUID, nil); err != nil {
|
||||
return err
|
||||
if val, ok := diskProps["TYPE"]; ok {
|
||||
disk.Type = val
|
||||
}
|
||||
if err := setSimpleDiskProperty("SIZE", diskSizeKey, dkey, diskProps, etcdClient); err != nil {
|
||||
return err
|
||||
if val, ok := diskProps["SIZE"]; ok {
|
||||
if size, err := strconv.ParseUint(val, 10, 64); err == nil {
|
||||
disk.Size = size
|
||||
}
|
||||
}
|
||||
if err := setSimpleDiskProperty("ROTA", diskRotationalKey, dkey, diskProps, etcdClient); err != nil {
|
||||
return err
|
||||
if val, ok := diskProps["ROTA"]; ok {
|
||||
if rotates, err := strconv.ParseBool(val); err == nil {
|
||||
disk.Rotational = rotates
|
||||
}
|
||||
}
|
||||
if err := setSimpleDiskProperty("RO", diskReadonlyKey, dkey, diskProps, etcdClient); err != nil {
|
||||
return err
|
||||
if val, ok := diskProps["RO"]; ok {
|
||||
if ro, err := strconv.ParseBool(val); err == nil {
|
||||
disk.Readonly = ro
|
||||
}
|
||||
}
|
||||
if err := setSimpleDiskProperty("PKNAME", diskParentKey, dkey, diskProps, etcdClient); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := etcdClient.Set(ctx.Background(), path.Join(dkey, diskFileSystemKey), fs, nil); err != nil {
|
||||
return err
|
||||
if val, ok := diskProps["PKNAME"]; ok {
|
||||
disk.Parent = val
|
||||
}
|
||||
|
||||
disks = append(disks, disk)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func setSimpleDiskProperty(propName, keyName, diskKey string, diskPropMap map[string]string, etcdClient etcd.KeysAPI) error {
|
||||
val, ok := diskPropMap[propName]
|
||||
if !ok {
|
||||
return fmt.Errorf("disk property %s not found in map: %+v", propName, diskPropMap)
|
||||
}
|
||||
if _, err := etcdClient.Set(ctx.Background(), path.Join(diskKey, keyName), val, nil); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// test usage only
|
||||
func TestSetDiskInfo(etcdClient *util.MockEtcdClient, hardwareKey string, name string, uuid string, size uint64, rotational bool, readonly bool,
|
||||
filesystem string, mountPoint string, diskType DiskType, parent string, hasChildren bool) DiskConfig {
|
||||
|
||||
diskKey := path.Join(hardwareKey, disksKey, name)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskUUIDKey), uuid, nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskSizeKey), strconv.FormatUint(size, 10), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskRotationalKey), strconv.Itoa(btoi(rotational)), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskReadonlyKey), strconv.Itoa(btoi(readonly)), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskFileSystemKey), filesystem, nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskMountPointKey), mountPoint, nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskTypeKey), DiskTypeToStr(diskType), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskParentKey), parent, nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(diskKey, diskHasChildrenKey), strconv.Itoa(btoi(hasChildren)), nil)
|
||||
|
||||
return DiskConfig{
|
||||
Name: name,
|
||||
UUID: uuid,
|
||||
Size: size,
|
||||
Rotational: rotational,
|
||||
Readonly: readonly,
|
||||
FileSystem: filesystem,
|
||||
MountPoint: mountPoint,
|
||||
Type: diskType,
|
||||
Parent: parent,
|
||||
HasChildren: hasChildren,
|
||||
}
|
||||
return disks, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
/*
|
||||
Copyright 2016 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 inventory
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"path"
|
||||
"testing"
|
||||
|
||||
"github.com/rook/rook/pkg/util"
|
||||
exectest "github.com/rook/rook/pkg/util/exec/test"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestSerializeClusterDisks(t *testing.T) {
|
||||
nodeID := "df1c87e8266843f2ab822c0d72f584d3"
|
||||
etcdClient := &util.MockEtcdClient{}
|
||||
d1 := LocalDisk{Name: "sda", UUID: "u1", Size: 23, Rotational: true, Readonly: false,
|
||||
FileSystem: "btrfs", MountPoint: "/mnt/abc", Type: DiskType, HasChildren: true}
|
||||
d2 := LocalDisk{Name: "sdb", UUID: "u2", Size: 24, Rotational: true, Readonly: false,
|
||||
Type: DiskType, HasChildren: true}
|
||||
|
||||
err := storeDevices(etcdClient, nodeID, []LocalDisk{d1, d2})
|
||||
assert.Nil(t, err)
|
||||
|
||||
key := path.Join(NodesConfigKey, nodeID, disksKey)
|
||||
rawDisk := etcdClient.GetValue(key)
|
||||
var disks []Disk
|
||||
err = json.Unmarshal([]byte(rawDisk), &disks)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 2, len(disks))
|
||||
assert.False(t, disks[0].Available)
|
||||
assert.True(t, disks[0].Rotational)
|
||||
assert.Equal(t, d1.Size, disks[0].Size)
|
||||
assert.Equal(t, d1.Type, disks[0].Type)
|
||||
|
||||
assert.True(t, disks[1].Available)
|
||||
assert.True(t, disks[1].Rotational)
|
||||
assert.Equal(t, d2.Size, disks[1].Size)
|
||||
assert.Equal(t, d2.Type, disks[1].Type)
|
||||
}
|
||||
|
||||
func TestAvailableDisks(t *testing.T) {
|
||||
|
||||
// no disks discovered for a node is an error
|
||||
disks := GetAvailableDevices([]LocalDisk{})
|
||||
assert.Equal(t, 0, len(disks))
|
||||
|
||||
// no available disks because of the formatting
|
||||
d1 := LocalDisk{Name: "sda", UUID: "myuuid1", Size: 123, Rotational: true, Readonly: false, FileSystem: "btrfs", MountPoint: "/mnt/abc", Type: DiskType, HasChildren: true}
|
||||
disks = GetAvailableDevices([]LocalDisk{d1})
|
||||
assert.Equal(t, 0, len(disks))
|
||||
|
||||
// multiple available disks
|
||||
d2 := LocalDisk{Name: "sdb", UUID: "myuuid2", Size: 123, Rotational: true, Readonly: false, Type: DiskType, HasChildren: true}
|
||||
d3 := LocalDisk{Name: "sdc", UUID: "myuuid3", Size: 123, Rotational: true, Readonly: false, Type: DiskType, HasChildren: true}
|
||||
disks = GetAvailableDevices([]LocalDisk{d1, d2, d3})
|
||||
assert.Equal(t, 2, len(disks))
|
||||
assert.Equal(t, "sdb", disks[0])
|
||||
assert.Equal(t, "sdc", disks[1])
|
||||
|
||||
// partitions don't result in more available devices
|
||||
d4 := LocalDisk{Name: "sdb1", UUID: "myuuid4", Size: 123, Rotational: true, Readonly: false, Type: PartType, HasChildren: true}
|
||||
d5 := LocalDisk{Name: "sdb2", UUID: "myuuid5", Size: 123, Rotational: true, Readonly: false, Type: PartType, HasChildren: true}
|
||||
disks = GetAvailableDevices([]LocalDisk{d1, d2, d3, d4, d5})
|
||||
assert.Equal(t, 2, len(disks))
|
||||
assert.Equal(t, "sdb", disks[0])
|
||||
assert.Equal(t, "sdc", disks[1])
|
||||
}
|
||||
|
||||
func TestDiscoverDevices(t *testing.T) {
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommand: func(actionName string, command string, arg ...string) error {
|
||||
logger.Infof("mock execute. %s. %s", actionName, command)
|
||||
return nil
|
||||
},
|
||||
MockExecuteCommandWithOutput: func(actionName string, command string, arg ...string) (string, error) {
|
||||
logger.Infof("mock execute with output. %s. %s", actionName, command)
|
||||
return "", nil
|
||||
},
|
||||
}
|
||||
devices, err := discoverDevices(executor)
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, 0, len(devices))
|
||||
}
|
||||
@@ -17,53 +17,56 @@ package inventory
|
||||
|
||||
import "time"
|
||||
|
||||
type DiskType int
|
||||
|
||||
const (
|
||||
MemoryTotalSizeKey = "total"
|
||||
NetworkIPv4AddressKey = "ipv4"
|
||||
NetworkIPv6AddressKey = "ipv6"
|
||||
NetworkSpeedKey = "speed"
|
||||
ProcPhysicalIDKey = "physical-id"
|
||||
ProcSiblingsKey = "siblings"
|
||||
ProcCoreIDKey = "core-id"
|
||||
ProcNumCoresKey = "cores"
|
||||
ProcSpeedKey = "speed"
|
||||
ProcBitsKey = "arch"
|
||||
Disk DiskType = iota
|
||||
Part
|
||||
DiskType = "disk"
|
||||
SSDType = "ssd"
|
||||
PartType = "part"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Nodes map[string]*NodeConfig `json:"nodes"`
|
||||
Local *Hardware `json:"local"`
|
||||
}
|
||||
|
||||
// Basic config info for a node in the cluster
|
||||
type NodeConfig struct {
|
||||
Disks []DiskConfig `json:"disks"`
|
||||
Disks []Disk `json:"disks"`
|
||||
Processors []ProcessorConfig `json:"processors"`
|
||||
Memory MemoryConfig `json:"memory"`
|
||||
NetworkAdapters []NetworkConfig `json:"networkAdapters"`
|
||||
Memory uint64 `json:"memory"`
|
||||
PublicIP string `json:"publicIp"`
|
||||
PrivateIP string `json:"privateIp"`
|
||||
HeartbeatAge time.Duration `json:"heartbeatAge"`
|
||||
Location string `json:"location"`
|
||||
}
|
||||
|
||||
type DiskConfig struct {
|
||||
Name string `json:"name"`
|
||||
UUID string `json:"uuid"`
|
||||
Size uint64 `json:"size"`
|
||||
Rotational bool `json:"rotational"`
|
||||
Readonly bool `json:"readonly"`
|
||||
FileSystem string `json:"fileSystem"`
|
||||
MountPoint string `json:"mountPoint"`
|
||||
Type DiskType `json:"type"`
|
||||
Parent string `json:"parent"`
|
||||
HasChildren bool `json:"hasChildren"`
|
||||
type Disk struct {
|
||||
Available bool `json:"available"`
|
||||
Type string `json:"type"`
|
||||
Size uint64 `json:"size"`
|
||||
Rotational bool `json:"rotational"`
|
||||
}
|
||||
|
||||
type MemoryConfig struct {
|
||||
TotalSize uint64 `json:"totalSize"`
|
||||
// Local hardware info
|
||||
type Hardware struct {
|
||||
Disks []LocalDisk `json:"disks"`
|
||||
Processors []ProcessorConfig `json:"processors"`
|
||||
NetworkAdapters []NetworkConfig `json:"networkAdapters"`
|
||||
Memory uint64 `json:"memory"`
|
||||
}
|
||||
|
||||
type LocalDisk struct {
|
||||
Name string `json:"name"`
|
||||
ID string `json:"id"`
|
||||
UUID string `json:"uuid"`
|
||||
Size uint64 `json:"size"`
|
||||
Rotational bool `json:"rotational"`
|
||||
Readonly bool `json:"readonly"`
|
||||
FileSystem string `json:"fileSystem"`
|
||||
MountPoint string `json:"mountPoint"`
|
||||
Type string `json:"type"`
|
||||
Parent string `json:"parent"`
|
||||
HasChildren bool `json:"hasChildren"`
|
||||
}
|
||||
|
||||
type ProcessorConfig struct {
|
||||
|
||||
@@ -23,6 +23,7 @@ import (
|
||||
ctx "golang.org/x/net/context"
|
||||
|
||||
"github.com/rook/rook/pkg/util"
|
||||
"github.com/rook/rook/pkg/util/exec"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -31,6 +32,28 @@ const (
|
||||
TriggerHardwareDetectionKey = "trigger-hardware-detection"
|
||||
)
|
||||
|
||||
// Discover all the hardware properties for this node.
|
||||
// Store the important properties in etcd and more detailed info for the local node in the context.
|
||||
func DiscoverHardware(etcdClient etcd.KeysAPI, executor exec.Executor, nodeID string) (*Hardware, error) {
|
||||
devices, err := discoverDevices(executor)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to discover devices. %+v", err)
|
||||
}
|
||||
|
||||
if err = storeDevices(etcdClient, nodeID, devices); err != nil {
|
||||
return nil, fmt.Errorf("failed to store disks in etcd. %+v", err)
|
||||
}
|
||||
|
||||
mem := getSystemMemory()
|
||||
if err = storeMemory(etcdClient, nodeID, mem); err != nil {
|
||||
return nil, fmt.Errorf("failed to store system memory in etcd. %+v", err)
|
||||
}
|
||||
|
||||
// TODO: Discover network adapters, processors, ...
|
||||
|
||||
return &Hardware{Disks: devices, Memory: mem}, nil
|
||||
}
|
||||
|
||||
func LoadDiscoveredNodes(etcdClient etcd.KeysAPI) (*Config, error) {
|
||||
|
||||
// Get the discovered state of the infrastructure
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
/*
|
||||
Copyright 2016 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 inventory
|
||||
|
||||
// #include <unistd.h>
|
||||
import "C"
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"path"
|
||||
"strconv"
|
||||
|
||||
etcd "github.com/coreos/etcd/client"
|
||||
ctx "golang.org/x/net/context"
|
||||
)
|
||||
|
||||
func getSystemMemory() uint64 {
|
||||
return uint64(C.sysconf(C._SC_PHYS_PAGES) * C.sysconf(C._SC_PAGE_SIZE))
|
||||
}
|
||||
|
||||
func storeMemory(etcdClient etcd.KeysAPI, nodeID string, size uint64) error {
|
||||
key := path.Join(NodesConfigKey, nodeID, memoryKey)
|
||||
_, err := etcdClient.Set(ctx.Background(), key, strconv.FormatUint(size, 10), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to store memory in etcd. %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadMemoryConfig(nodeConfig *NodeConfig, rawMemory string) error {
|
||||
size, err := strconv.ParseUint(rawMemory, 10, 64)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to parse memory. %+v", err)
|
||||
}
|
||||
nodeConfig.Memory = size
|
||||
return nil
|
||||
}
|
||||
+47
-171
@@ -19,13 +19,13 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
ctx "golang.org/x/net/context"
|
||||
|
||||
"github.com/rook/rook/pkg/util"
|
||||
"github.com/rook/rook/pkg/util/exec"
|
||||
|
||||
"encoding/json"
|
||||
|
||||
etcd "github.com/coreos/etcd/client"
|
||||
)
|
||||
@@ -34,8 +34,8 @@ const (
|
||||
HeartbeatKey = "heartbeat"
|
||||
disksKey = "disks"
|
||||
processorsKey = "cpu"
|
||||
networkKey = "net"
|
||||
memoryKey = "mem"
|
||||
networkKey = "network"
|
||||
memoryKey = "memory"
|
||||
locationKey = "location"
|
||||
heartbeatTtlSeconds = 60 * 60
|
||||
publicIpAddressKey = "publicIp"
|
||||
@@ -46,17 +46,6 @@ var (
|
||||
HeartbeatTtlDuration = time.Duration(heartbeatTtlSeconds) * time.Second
|
||||
)
|
||||
|
||||
func DiscoverHardware(nodeID string, etcdClient etcd.KeysAPI, executor exec.Executor) error {
|
||||
nodeConfigKey := GetNodeConfigKey(nodeID)
|
||||
if err := discoverDisks(nodeConfigKey, etcdClient, executor); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// TODO: discover more hardware properties
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// gets the key under which all node hardware/config will be stored
|
||||
func GetNodeConfigKey(nodeID string) string {
|
||||
return path.Join(NodesConfigKey, nodeID)
|
||||
@@ -131,7 +120,6 @@ func loadNodeHealth(etcdClient etcd.KeysAPI, nodes map[string]*NodeConfig) error
|
||||
}
|
||||
|
||||
func loadSingleNodeHealth(node *NodeConfig, health *etcd.Node) error {
|
||||
|
||||
for _, prop := range health.Nodes {
|
||||
switch util.GetLeafKeyPath(prop.Key) {
|
||||
case HeartbeatKey:
|
||||
@@ -151,7 +139,6 @@ func SetIPAddress(etcdClient etcd.KeysAPI, nodeId, publicIP, privateIP string) e
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return setConfigProperty(etcdClient, nodeId, privateIpAddressKey, privateIP)
|
||||
}
|
||||
|
||||
@@ -176,25 +163,25 @@ func loadHardwareConfig(nodeId string, nodeConfig *NodeConfig, nodeInfo *etcd.No
|
||||
var err error
|
||||
switch key {
|
||||
case disksKey:
|
||||
err = loadDisksConfig(nodeConfig, nodeConfigRoot)
|
||||
err = loadDisksConfig(nodeConfig, nodeConfigRoot.Value)
|
||||
|
||||
case processorsKey:
|
||||
err = loadProcessorsConfig(nodeConfig, nodeConfigRoot)
|
||||
|
||||
case memoryKey:
|
||||
err = loadMemoryConfig(nodeConfig, nodeConfigRoot)
|
||||
err = loadProcessorsConfig(nodeConfig, nodeConfigRoot.Value)
|
||||
|
||||
case networkKey:
|
||||
err = loadNetworkConfig(nodeConfig, nodeConfigRoot)
|
||||
err = loadNetworkConfig(nodeConfig, nodeConfigRoot.Value)
|
||||
|
||||
case memoryKey:
|
||||
err = loadMemoryConfig(nodeConfig, nodeConfigRoot.Value)
|
||||
|
||||
case publicIpAddressKey:
|
||||
err = loadSimpleConfigStringProperty(&(nodeConfig.PublicIP), nodeConfigRoot, "Public IP")
|
||||
nodeConfig.PublicIP = nodeConfigRoot.Value
|
||||
|
||||
case privateIpAddressKey:
|
||||
err = loadSimpleConfigStringProperty(&(nodeConfig.PrivateIP), nodeConfigRoot, "Private IP")
|
||||
nodeConfig.PrivateIP = nodeConfigRoot.Value
|
||||
|
||||
case locationKey:
|
||||
err = loadSimpleConfigStringProperty(&(nodeConfig.Location), nodeConfigRoot, "Location")
|
||||
nodeConfig.Location = nodeConfigRoot.Value
|
||||
|
||||
default:
|
||||
logger.Warningf("unexpected hardware component: %s, skipping...", nodeConfigRoot)
|
||||
@@ -209,161 +196,50 @@ func loadHardwareConfig(nodeId string, nodeConfig *NodeConfig, nodeInfo *etcd.No
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadDisksConfig(nodeConfig *NodeConfig, disksRootNode *etcd.Node) error {
|
||||
numDisks := 0
|
||||
if disksRootNode.Nodes != nil {
|
||||
numDisks = len(disksRootNode.Nodes)
|
||||
func loadProcessorsConfig(nodeConfig *NodeConfig, rawProcs string) error {
|
||||
var processors []ProcessorConfig
|
||||
if err := json.Unmarshal([]byte(rawProcs), &processors); err != nil {
|
||||
return fmt.Errorf("failed to deserialize processors. %+v", err)
|
||||
}
|
||||
nodeConfig.Processors = processors
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadNetworkConfig(nodeConfig *NodeConfig, rawAdapters string) error {
|
||||
var adapters []NetworkConfig
|
||||
if err := json.Unmarshal([]byte(rawAdapters), &adapters); err != nil {
|
||||
return fmt.Errorf("failed to deserialize network adapters. %+v", err)
|
||||
}
|
||||
nodeConfig.NetworkAdapters = adapters
|
||||
return nil
|
||||
}
|
||||
|
||||
func storeNetworkConfig(etcdClient etcd.KeysAPI, nodeID string, config []NetworkConfig) error {
|
||||
output, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal network config. %+v", err)
|
||||
}
|
||||
|
||||
nodeConfig.Disks = make([]DiskConfig, numDisks)
|
||||
|
||||
// iterate over all disks from etcd
|
||||
for i, diskInfo := range disksRootNode.Nodes {
|
||||
disk, err := getDiskInfo(diskInfo)
|
||||
if err != nil {
|
||||
logger.Errorf("Failed to get disk. err=%v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
nodeConfig.Disks[i] = *disk
|
||||
|
||||
key := path.Join(NodesConfigKey, nodeID, networkKey)
|
||||
_, err = etcdClient.Set(ctx.Background(), key, string(output), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to store network config in etcd. %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadProcessorsConfig(nodeConfig *NodeConfig, procsRootNode *etcd.Node) error {
|
||||
numProcs := 0
|
||||
if procsRootNode.Nodes != nil {
|
||||
numProcs = len(procsRootNode.Nodes)
|
||||
func storeProcessorConfig(etcdClient etcd.KeysAPI, nodeID string, config []ProcessorConfig) error {
|
||||
output, err := json.Marshal(config)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to marshal processor config. %+v", err)
|
||||
}
|
||||
|
||||
nodeConfig.Processors = make([]ProcessorConfig, numProcs)
|
||||
|
||||
// iterate over all processors from etcd
|
||||
for i, procInfo := range procsRootNode.Nodes {
|
||||
proc := ProcessorConfig{}
|
||||
if procID, err := strconv.ParseUint(util.GetLeafKeyPath(procInfo.Key), 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.ID = uint(procID)
|
||||
}
|
||||
|
||||
// iterate over all properties of the processor
|
||||
for _, procProperty := range procInfo.Nodes {
|
||||
procPropertyName := util.GetLeafKeyPath(procProperty.Key)
|
||||
switch procPropertyName {
|
||||
case ProcPhysicalIDKey:
|
||||
if phsyicalId, err := strconv.ParseUint(procProperty.Value, 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.PhysicalID = uint(phsyicalId)
|
||||
}
|
||||
case ProcSiblingsKey:
|
||||
if siblings, err := strconv.ParseUint(procProperty.Value, 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.Siblings = uint(siblings)
|
||||
}
|
||||
case ProcCoreIDKey:
|
||||
if coreId, err := strconv.ParseUint(procProperty.Value, 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.CoreID = uint(coreId)
|
||||
}
|
||||
case ProcNumCoresKey:
|
||||
if numCores, err := strconv.ParseUint(procProperty.Value, 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.NumCores = uint(numCores)
|
||||
}
|
||||
case ProcSpeedKey:
|
||||
if speed, err := strconv.ParseFloat(procProperty.Value, 64); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.Speed = speed
|
||||
}
|
||||
case ProcBitsKey:
|
||||
if numBits, err := strconv.ParseUint(procProperty.Value, 10, 32); err != nil {
|
||||
return err
|
||||
} else {
|
||||
proc.Bits = uint(numBits)
|
||||
}
|
||||
default:
|
||||
logger.Warningf("unknown processor property key %s, skipping", procPropertyName)
|
||||
}
|
||||
}
|
||||
|
||||
nodeConfig.Processors[i] = proc
|
||||
key := path.Join(NodesConfigKey, nodeID, processorsKey)
|
||||
_, err = etcdClient.Set(ctx.Background(), key, string(output), nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to store processor config in etcd. %+v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadMemoryConfig(nodeConfig *NodeConfig, memoryRootNode *etcd.Node) error {
|
||||
mem := MemoryConfig{}
|
||||
for _, memProperty := range memoryRootNode.Nodes {
|
||||
memPropertyName := util.GetLeafKeyPath(memProperty.Key)
|
||||
switch memPropertyName {
|
||||
case MemoryTotalSizeKey:
|
||||
if size, err := strconv.ParseUint(memProperty.Value, 10, 64); err != nil {
|
||||
return err
|
||||
} else {
|
||||
mem.TotalSize = size
|
||||
}
|
||||
default:
|
||||
logger.Warningf("unknown memory property key %s, skipping", memPropertyName)
|
||||
}
|
||||
}
|
||||
|
||||
nodeConfig.Memory = mem
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadNetworkConfig(nodeConfig *NodeConfig, networkRootNode *etcd.Node) error {
|
||||
numNics := 0
|
||||
if networkRootNode.Nodes != nil {
|
||||
numNics = len(networkRootNode.Nodes)
|
||||
}
|
||||
|
||||
nodeConfig.NetworkAdapters = make([]NetworkConfig, numNics)
|
||||
|
||||
// iterate over all network adapters from etcd
|
||||
for i, netInfo := range networkRootNode.Nodes {
|
||||
net := NetworkConfig{}
|
||||
net.Name = util.GetLeafKeyPath(netInfo.Key)
|
||||
|
||||
// iterate over all properties of the network adapter
|
||||
for _, netProperty := range netInfo.Nodes {
|
||||
netPropertyName := util.GetLeafKeyPath(netProperty.Key)
|
||||
switch netPropertyName {
|
||||
case NetworkIPv4AddressKey:
|
||||
net.IPv4Address = netProperty.Value
|
||||
case NetworkIPv6AddressKey:
|
||||
net.IPv6Address = netProperty.Value
|
||||
case NetworkSpeedKey:
|
||||
if netProperty.Value == "" {
|
||||
net.Speed = 0
|
||||
} else if speed, err := strconv.ParseUint(netProperty.Value, 10, 64); err != nil {
|
||||
return err
|
||||
} else {
|
||||
net.Speed = speed
|
||||
}
|
||||
default:
|
||||
logger.Warningf("unknown network adapter property key %s, skipping", netPropertyName)
|
||||
}
|
||||
}
|
||||
|
||||
nodeConfig.NetworkAdapters[i] = net
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func loadSimpleConfigStringProperty(cfgField *string, cfgNode *etcd.Node, propName string) error {
|
||||
if cfgNode.Dir {
|
||||
return fmt.Errorf("%s node '%s' is a directory, but it's expected to be a key", propName, cfgNode.Key)
|
||||
}
|
||||
*cfgField = cfgNode.Value
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -17,13 +17,9 @@ package inventory
|
||||
|
||||
import (
|
||||
"path"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
etcd "github.com/coreos/etcd/client"
|
||||
ctx "golang.org/x/net/context"
|
||||
|
||||
"github.com/rook/rook/pkg/util"
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
@@ -61,45 +57,40 @@ func TestLoadDiscoveredNodes(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestLoadHardwareConfig(t *testing.T) {
|
||||
machineId := "df1c87e8266843f2ab822c0d72f584d3"
|
||||
nodeID := "12345"
|
||||
etcdClient := util.NewMockEtcdClient()
|
||||
|
||||
// set up the hardware date in etcd
|
||||
hardwareKey := path.Join(NodesConfigKey, machineId)
|
||||
etcdClient.CreateDir(hardwareKey)
|
||||
|
||||
// setup disk info in etcd
|
||||
disksConfig := make([]DiskConfig, 2)
|
||||
disksConfig[0] = TestSetDiskInfo(etcdClient, hardwareKey, "sda", "12344869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
10737418240, true, false, "btrfs", "/mnt/abc", Disk, "", true)
|
||||
disksConfig[1] = TestSetDiskInfo(etcdClient, hardwareKey, "sda2", "23454869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
2097152, false, true, "", "", Part, "sda", false)
|
||||
d1 := LocalDisk{Name: "sda", UUID: "uuid1", Size: 10737418240, Rotational: true, Readonly: false, Type: DiskType, HasChildren: true}
|
||||
d2 := LocalDisk{Name: "sda2", UUID: "uuid2", Size: 2097152, Rotational: false, Readonly: true, Type: PartType, HasChildren: false}
|
||||
err := storeDevices(etcdClient, nodeID, []LocalDisk{d1, d2})
|
||||
assert.Nil(t, err)
|
||||
|
||||
// setup processor info in etcd
|
||||
procsKey := path.Join(hardwareKey, processorsKey)
|
||||
etcdClient.Set(ctx.Background(), procsKey, "", &etcd.SetOptions{Dir: true})
|
||||
procsConfig := make([]ProcessorConfig, 3)
|
||||
procsConfig[0] = setProcInfo(etcdClient, procsKey, 0, 0, 1, 0, 1, 1234.56, 64)
|
||||
procsConfig[1] = setProcInfo(etcdClient, procsKey, 1, 1, 2, 0, 2, 8000.00, 32)
|
||||
procsConfig[2] = setProcInfo(etcdClient, procsKey, 2, 1, 2, 1, 2, 4000.01, 32)
|
||||
p1 := ProcessorConfig{ID: 0, PhysicalID: 3, Siblings: 1, CoreID: 6, NumCores: 1, Speed: 1234.56, Bits: 64}
|
||||
p2 := ProcessorConfig{ID: 1, PhysicalID: 4, Siblings: 2, CoreID: 7, NumCores: 2, Speed: 8000.00, Bits: 32}
|
||||
p3 := ProcessorConfig{ID: 2, PhysicalID: 5, Siblings: 0, CoreID: 8, NumCores: 4, Speed: 4000.01, Bits: 32}
|
||||
err = storeProcessorConfig(etcdClient, nodeID, []ProcessorConfig{p1, p2, p3})
|
||||
assert.Nil(t, err)
|
||||
|
||||
// setup memory info in etcd
|
||||
memKey := path.Join(hardwareKey, memoryKey)
|
||||
etcdClient.Set(ctx.Background(), memKey, "", &etcd.SetOptions{Dir: true})
|
||||
memConfig := setMemoryInfo(etcdClient, memKey, 4149252096)
|
||||
mem := getSystemMemory()
|
||||
err = storeMemory(etcdClient, nodeID, mem)
|
||||
assert.Nil(t, err)
|
||||
|
||||
// set up network info in etcd
|
||||
netKey := path.Join(hardwareKey, networkKey)
|
||||
etcdClient.Set(ctx.Background(), netKey, "", &etcd.SetOptions{Dir: true})
|
||||
netsConfig := make([]NetworkConfig, 2)
|
||||
netsConfig[0] = setNetInfo(etcdClient, netKey, "eth0", "172.17.42.1/16", "fe80::42:4aff:fefe:13d7/64", 0)
|
||||
netsConfig[1] = setNetInfo(etcdClient, netKey, "veth2b6453a", "", "fe80::7c0f:acff:feff:478d/64", 10000)
|
||||
n1 := NetworkConfig{Name: "eth0", IPv4Address: "172.17.42.1/16", IPv6Address: "fe80::42:4aff:fefe:13d7/64", Speed: 0}
|
||||
n2 := NetworkConfig{Name: "veth2b6453a", IPv6Address: "fe80::7c0f:acff:feff:478d/64", Speed: 10000}
|
||||
err = storeNetworkConfig(etcdClient, nodeID, []NetworkConfig{n1, n2})
|
||||
assert.Nil(t, err)
|
||||
|
||||
// set IP address in etcd
|
||||
SetIPAddress(etcdClient, machineId, "10.0.0.43", "10.2.3.4")
|
||||
publicIP := "10.0.0.43"
|
||||
privateIP := "10.2.3.4"
|
||||
SetIPAddress(etcdClient, nodeID, publicIP, privateIP)
|
||||
|
||||
// set location in etcd
|
||||
SetLocation(etcdClient, machineId, "root=default,dc=datacenter1")
|
||||
SetLocation(etcdClient, nodeID, "root=default,dc=datacenter1")
|
||||
|
||||
// load the discovered node config
|
||||
nodeConfig, err := loadNodeConfig(etcdClient)
|
||||
@@ -107,132 +98,29 @@ func TestLoadHardwareConfig(t *testing.T) {
|
||||
assert.NotNil(t, nodeConfig, "loaded node config should not be nil")
|
||||
assert.Equal(t, 1, len(nodeConfig))
|
||||
|
||||
// verify all the hardware configuration retrieved from etcd
|
||||
verifyDiskConfig(t, nodeConfig[machineId], disksConfig)
|
||||
verifyProcConfig(t, nodeConfig[machineId], procsConfig)
|
||||
verifyMemoryConfig(t, nodeConfig[machineId], memConfig)
|
||||
verifyNetworkConfig(t, nodeConfig[machineId], netsConfig)
|
||||
assert.Equal(t, "10.0.0.43", nodeConfig[machineId].PublicIP)
|
||||
assert.Equal(t, "root=default,dc=datacenter1", nodeConfig[machineId].Location)
|
||||
}
|
||||
cfg := nodeConfig[nodeID]
|
||||
|
||||
func setProcInfo(etcdClient *util.MockEtcdClient, procsKey string, procId uint, physicalId uint, siblings uint,
|
||||
coreID uint, numCores uint, speed float64, bits uint) ProcessorConfig {
|
||||
// verify the single disk (the partition is not saved)
|
||||
assert.Equal(t, 1, len(cfg.Disks))
|
||||
assert.True(t, cfg.Disks[0].Available)
|
||||
assert.Equal(t, d1.Rotational, cfg.Disks[0].Rotational)
|
||||
assert.Equal(t, d1.Size, cfg.Disks[0].Size)
|
||||
assert.Equal(t, d1.Type, cfg.Disks[0].Type)
|
||||
|
||||
procKey := path.Join(procsKey, strconv.FormatUint(uint64(procId), 10))
|
||||
etcdClient.Set(ctx.Background(), procKey, "", &etcd.SetOptions{Dir: true})
|
||||
// verify the processors
|
||||
assert.Equal(t, 3, len(cfg.Processors))
|
||||
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcPhysicalIDKey), strconv.FormatUint(uint64(physicalId), 10), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcSiblingsKey), strconv.FormatUint(uint64(siblings), 10), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcCoreIDKey), strconv.FormatUint(uint64(coreID), 10), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcNumCoresKey), strconv.FormatUint(uint64(numCores), 10), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcSpeedKey), strconv.FormatFloat(speed, 'f', 2, 64), nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(procKey, ProcBitsKey), strconv.FormatUint(uint64(bits), 10), nil)
|
||||
// verify the network adapters
|
||||
assert.Equal(t, 2, len(cfg.NetworkAdapters))
|
||||
|
||||
return ProcessorConfig{
|
||||
ID: procId,
|
||||
PhysicalID: physicalId,
|
||||
Siblings: siblings,
|
||||
CoreID: coreID,
|
||||
NumCores: numCores,
|
||||
Speed: speed,
|
||||
Bits: bits,
|
||||
}
|
||||
}
|
||||
// verify the simple values
|
||||
assert.Equal(t, mem, cfg.Memory)
|
||||
assert.Equal(t, "10.0.0.43", cfg.PublicIP)
|
||||
assert.Equal(t, "root=default,dc=datacenter1", cfg.Location)
|
||||
|
||||
func setMemoryInfo(etcdClient *util.MockEtcdClient, memKey string, totalMem uint64) MemoryConfig {
|
||||
etcdClient.Set(ctx.Background(), path.Join(memKey, MemoryTotalSizeKey), strconv.FormatUint(totalMem, 10), nil)
|
||||
return MemoryConfig{TotalSize: totalMem}
|
||||
}
|
||||
// verify that some values are in etcd
|
||||
key := path.Join(NodesConfigKey, nodeID)
|
||||
assert.Equal(t, publicIP, etcdClient.GetValue(path.Join(key, publicIpAddressKey)))
|
||||
assert.Equal(t, privateIP, etcdClient.GetValue(path.Join(key, privateIpAddressKey)))
|
||||
|
||||
func setNetInfo(etcdClient *util.MockEtcdClient, netsKey string, name string, ipv4Addr string, ipv6Addr string, speed uint64) NetworkConfig {
|
||||
netKey := path.Join(netsKey, name)
|
||||
etcdClient.Set(ctx.Background(), netKey, "", &etcd.SetOptions{Dir: true})
|
||||
|
||||
etcdClient.Set(ctx.Background(), path.Join(netKey, NetworkIPv4AddressKey), ipv4Addr, nil)
|
||||
etcdClient.Set(ctx.Background(), path.Join(netKey, NetworkIPv6AddressKey), ipv6Addr, nil)
|
||||
speedStr := ""
|
||||
if speed > 0 {
|
||||
speedStr = strconv.FormatUint(speed, 10)
|
||||
}
|
||||
etcdClient.Set(ctx.Background(), path.Join(netKey, NetworkSpeedKey), speedStr, nil)
|
||||
|
||||
return NetworkConfig{
|
||||
Name: name,
|
||||
IPv4Address: ipv4Addr,
|
||||
IPv6Address: ipv6Addr,
|
||||
Speed: speed,
|
||||
}
|
||||
}
|
||||
|
||||
func verifyDiskConfig(t *testing.T, nodeConfig *NodeConfig, expectedDisksConfig []DiskConfig) {
|
||||
assert.Equal(t, len(expectedDisksConfig), len(nodeConfig.Disks))
|
||||
|
||||
for _, expectedDisk := range expectedDisksConfig {
|
||||
var matchingActual DiskConfig
|
||||
for _, actualDisk := range nodeConfig.Disks {
|
||||
if actualDisk.UUID == expectedDisk.UUID {
|
||||
matchingActual = actualDisk
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
assert.NotNil(t, matchingActual, "missing actual disk %s", expectedDisk.UUID)
|
||||
assert.Equal(t, expectedDisk, matchingActual)
|
||||
}
|
||||
}
|
||||
|
||||
func verifyProcConfig(t *testing.T, nodeConfig *NodeConfig, expectedProcsConfig []ProcessorConfig) {
|
||||
assert.Equal(t, len(expectedProcsConfig), len(nodeConfig.Processors))
|
||||
|
||||
for _, expectedProc := range expectedProcsConfig {
|
||||
var matchingActual ProcessorConfig
|
||||
for _, actualProc := range nodeConfig.Processors {
|
||||
if actualProc.ID == expectedProc.ID {
|
||||
matchingActual = actualProc
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
assert.NotNil(t, matchingActual, "missing actual proc %d", expectedProc.ID)
|
||||
assert.Equal(t, expectedProc, matchingActual)
|
||||
}
|
||||
}
|
||||
|
||||
func verifyMemoryConfig(t *testing.T, nodeConfig *NodeConfig, expectedMemConfig MemoryConfig) {
|
||||
assert.NotNil(t, nodeConfig.Memory)
|
||||
assert.Equal(t, expectedMemConfig, nodeConfig.Memory)
|
||||
}
|
||||
|
||||
func verifyNetworkConfig(t *testing.T, nodeConfig *NodeConfig, expectedNetsConfig []NetworkConfig) {
|
||||
assert.Equal(t, len(expectedNetsConfig), len(nodeConfig.NetworkAdapters))
|
||||
|
||||
for _, expectedNet := range expectedNetsConfig {
|
||||
var matchingActual NetworkConfig
|
||||
for _, actualNet := range nodeConfig.NetworkAdapters {
|
||||
if actualNet.Name == expectedNet.Name {
|
||||
matchingActual = actualNet
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
assert.NotNil(t, matchingActual, "missing actual network adapter %s", expectedNet.Name)
|
||||
assert.Equal(t, expectedNet, matchingActual)
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetSimpleDiskPropertiesFromSerial(t *testing.T) {
|
||||
nodeID := "df1c87e8266843f2ab822c0d72f584d3"
|
||||
etcdClient := &util.MockEtcdClient{}
|
||||
hardwareKey := path.Join(NodesConfigKey, nodeID)
|
||||
etcdClient.Set(ctx.Background(), hardwareKey, "", &etcd.SetOptions{Dir: true})
|
||||
TestSetDiskInfo(etcdClient, hardwareKey, "sda", "abcd4869-29ee-4bfd-bf21-dfd597bd222e",
|
||||
10737418240, true, false, "btrfs", "/mnt/abc", Disk, "", true)
|
||||
|
||||
diskNode, _ := etcdClient.Get(ctx.Background(), path.Join(hardwareKey, "disks", "sda"), nil)
|
||||
disk, err := getDiskInfo(diskNode.Node)
|
||||
assert.Nil(t, err)
|
||||
|
||||
assert.Equal(t, "sda", disk.Name)
|
||||
assert.Equal(t, "abcd4869-29ee-4bfd-bf21-dfd597bd222e", disk.UUID)
|
||||
}
|
||||
|
||||
@@ -77,7 +77,15 @@ func StartJoinCluster(services []*ClusterService, configDir, nodeID, discoveryUR
|
||||
ProcMan: proc.New(executor),
|
||||
ConfigDir: configDir,
|
||||
LogLevel: logLevel,
|
||||
Inventory: &inventory.Config{},
|
||||
}
|
||||
|
||||
// initialize the device inventory
|
||||
context.Inventory.Local, err = inventory.DiscoverHardware(context.EtcdClient, context.Executor, context.NodeID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed initial hardware discovery. %+v", err)
|
||||
}
|
||||
|
||||
clusterLeader := newServicesLeader(context)
|
||||
clusterMember := newClusterMember(context, leaseManager, clusterLeader)
|
||||
clusterLeader.parent = clusterMember
|
||||
|
||||
Reference in New Issue
Block a user