Merge branch 'master' into etcd-crash

.gitignore

@@ -14,3 +14,5 @@ debug.test
 
 .vagrant/
 demo/vagrant/cloud-config.yml
+
+.virtualbox/

cmd/castle/castle.go

@@ -16,11 +16,11 @@ var (
 )
 
 const (
+	systemLogDir   = "/var/log/castle"
 	outputPadding  = 3
 	outputMinWidth = 10
 	outputTabWidth = 0
 	outputPadChar  = ' '
-	logFileDir     = "/tmp/castle/log"
 	logFileName    = "castle.log"
 )
 
@@ -32,10 +32,10 @@ var rootCmd = &cobra.Command{
 
 func Main() {
 	// set up logging to a log file instead of stdout (only command output and errors should go to stdout/stderr)
-	if err := os.MkdirAll(logFileDir, 0744); err != nil {
-		log.Fatalf("failed to create logging dir '%s': %+v", logFileDir, err)
+	if err := os.MkdirAll(systemLogDir, 0744); err != nil {
+		log.Fatalf("failed to create logging dir '%s': %+v", systemLogDir, err)
 	}
-	logFilePath := filepath.Join(logFileDir, logFileName)
+	logFilePath := filepath.Join(systemLogDir, logFileName)
 	logFile, err := os.OpenFile(logFilePath, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644)
 	if err != nil {
 		log.Fatalf("failed to open log file '%s': %v", logFilePath, err)

cmd/castled/main.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"path"
 	"strings"
 	"time"
 
@@ -17,6 +18,7 @@ import (
 	"github.com/quantum/castle/pkg/cephmgr"
 	"github.com/quantum/castle/pkg/cephmgr/cephd"
 	"github.com/quantum/castle/pkg/clusterd"
+
 	"github.com/quantum/castle/pkg/util"
 	"github.com/quantum/castle/pkg/util/flags"
 	"github.com/quantum/castle/pkg/util/proc"
@@ -36,6 +38,7 @@ type config struct {
 	publicIPv4   string
 	privateIPv4  string
 	devices      string
+	dataDir      string
 	forceFormat  bool
 	location     string
 }
@@ -62,7 +65,8 @@ func init() {
 	rootCmd.Flags().StringVar(&cfg.etcdMembers, "etcd-members", "", "etcd members to connect to. Overrides the discovery URL. Example: http://10.23.45.56:2379")
 	rootCmd.Flags().StringVar(&cfg.publicIPv4, "public-ipv4", "", "public IPv4 address for this machine")
 	rootCmd.Flags().StringVar(&cfg.privateIPv4, "private-ipv4", "127.0.0.1", "private IPv4 address for this machine")
-	rootCmd.Flags().StringVar(&cfg.devices, "devices", "", "comma separated list of devices to use")
+	rootCmd.Flags().StringVar(&cfg.devices, "data-devices", "", "comma separated list of devices to use for storage")
+	rootCmd.Flags().StringVar(&cfg.dataDir, "data-dir", "/var/lib/castle", "directory for storing configuration")
 	rootCmd.Flags().BoolVar(&cfg.forceFormat, "force-format", false,
 		"true to force the format of any specified devices, even if they already have a filesystem.  BE CAREFUL!")
 	rootCmd.Flags().StringVar(&cfg.location, "location", "", "location of this node for CRUSH placement")
@@ -79,10 +83,18 @@ func addCommands() {
 }
 
 func joinCluster(cmd *cobra.Command, args []string) error {
-
+	// verify required flags
 	if err := flags.VerifyRequiredFlags(cmd, []string{}); err != nil {
 		return err
 	}
+
+	// ensure the data root exists
+	if err := os.MkdirAll(cfg.dataDir, 0744); err != nil {
+		log.Printf("failed to create data directory at %s: %+v", cfg.dataDir, err)
+		return nil
+	}
+
+	// load the etcd discovery url
 	if cfg.discoveryURL == "" && cfg.etcdMembers == "" {
 		// if discovery isn't specified and etcd members aren't specified, try to request a default discovery URL
 		discURL, err := loadDefaultDiscoveryURL()
@@ -112,14 +124,14 @@ func joinCluster(cmd *cobra.Command, args []string) error {
 	}
 
 	// start the cluster orchestration services
-	context, err := clusterd.StartJoinCluster(services, procMan, cfg.nodeID, cfg.discoveryURL, cfg.etcdMembers, cfg.publicIPv4, cfg.privateIPv4)
+	context, err := clusterd.StartJoinCluster(services, procMan, cfg.dataDir, cfg.nodeID, cfg.discoveryURL, cfg.etcdMembers, cfg.publicIPv4, cfg.privateIPv4)
 	if err != nil {
 		return err
 	}
 
 	go func() {
 		// set up routes and start HTTP server for REST API
-		h := api.NewHandler(context.EtcdClient, cephmgr.NewConnectionFactory(), cephd.New())
+		h := api.NewHandler(context, cephmgr.NewConnectionFactory(), cephd.New())
 		r := api.NewRouter(h.GetRoutes())
 		if err := http.ListenAndServe(":8124", r); err != nil {
 			log.Printf("API server error: %+v", err)
@@ -151,7 +163,7 @@ func setFlagsFromEnv(flags *pflag.FlagSet) error {
 
 func loadDefaultDiscoveryURL() (string, error) {
 	// try to load the cached discovery URL if it exists
-	cachedPath := "/tmp/castle-discovery-url"
+	cachedPath := path.Join(cfg.dataDir, "castle-discovery-url")
 	fileContent, err := ioutil.ReadFile(cachedPath)
 	if err == nil {
 		return strings.TrimSpace(string(fileContent)), nil

demo/vagrant/Vagrantfile

@@ -3,6 +3,8 @@ VAGRANTFILE_API_VERSION = "2"
 $channel = "alpha"
 $nodes = 3
 $new_discovery_url="https://discovery.etcd.io/new?size=1"
+$node_disks = 3
+$disk_size = 5
 
 require_relative 'util.rb'
 require 'open-uri'
@@ -15,6 +17,8 @@ Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
       node.vm.hostname = name
       node.vm.network "private_network", ip: "172.20.20.#{10+i}"
 
+      attach_volumes(node, $node_disks, $disk_size)
+
       # enable if you want to run for locally build ACI files
       #node.vm.synced_folder "../../release", "/release", id: "release", :nfs => true, :mount_options => ['nolock,vers=3,tcp']
 

demo/vagrant/cloud-config.yml.in

@@ -28,9 +28,10 @@ coreos:
       Environment=CASTLED_DISCOVERY_URL=%%token%%
       Environment=CASTLED_PUBLIC_IPV4=$public_ipv4
       Environment=CASTLED_PRIVATE_IPV4=$private_ipv4
-      Environment=CASTLED_DATA_DIRS=/var/lib/castled/data
+      Environment=CASTLED_DATA_DIR=/var/lib/castled
+      #Environment=CASTLED_DATA_DEVICES=sdb,sdc,sdd
 
-      ExecStartPre=/usr/bin/mkdir -p ${CASTLED_DATA_DIRS}
+      ExecStartPre=/usr/bin/mkdir -p ${CASTLED_DATA_DIR}
 
       ExecStart=/usr/bin/rkt run \
         --trust-keys-from-https \

demo/vagrant/util.rb

@@ -1,6 +1,6 @@
 def set_coreos_box(config, channel)
   config.vm.box = "coreos-%s" % channel
-  config.vm.box_url = "http://%s.release.core-os.net/amd64-usr/current/coreos_production_vagrant.json" % $coreos_channel
+  config.vm.box_url = "http://%s.release.core-os.net/amd64-usr/current/coreos_production_vagrant.json" % $channel
 
   ["vmware_fusion", "vmware_workstation"].each do |vmware|
     config.vm.provider vmware do |v, override|
@@ -54,3 +54,48 @@ class CloudConfigPlugin < Vagrant.plugin('2')
     Provisioner
   end
 end
+
+class VagrantPlugins::ProviderVirtualBox::Action::SetName
+  alias_method :original_call, :call
+  def call(env)
+    machine = env[:machine]
+    driver = machine.provider.driver
+    uuid = driver.instance_eval { @uuid }
+    ui = env[:ui]
+
+    controller_name="SATA Controller"
+
+    vm_info = driver.execute("showvminfo", uuid)
+    controller_already_exists = vm_info.match("Storage Controller Name.*#{controller_name}")
+
+    if controller_already_exists
+      ui.info "already has the #{controller_name} hdd controller, skipping creation/add"
+    else
+      ui.info "creating #{controller_name} hdd controller"
+      driver.execute(
+        'storagectl',
+        uuid,
+        '--name', "#{controller_name}",
+        '--add', 'sata',
+        '--controller', 'IntelAHCI')
+    end
+
+    original_call(env)
+  end
+end
+
+# Add persistent storage volumes
+def attach_volumes(node, num_volumes, volume_size)
+
+    node.vm.provider :virtualbox do |v, override|
+      (1..num_volumes).each do |disk|
+        diskname = File.join(File.dirname(File.expand_path(__FILE__)), ".virtualbox", "#{node.vm.hostname}-#{disk}.vdi")
+        unless File.exist?(diskname)
+          v.customize ['createhd', '--filename', diskname, '--size', volume_size * 1024]
+        end
+        v.customize ['storageattach', :id, '--storagectl', 'SATA Controller', '--port', disk, '--device', 0, '--type', 'hdd', '--medium', diskname]
+      end
+    end
+
+end
+

pkg/api/device.go

@@ -17,7 +17,7 @@ func (h *Handler) AddDevice(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err := cephmgr.AddDesiredDevice(h.EtcdClient, device)
+	err := cephmgr.AddDesiredDevice(h.context.EtcdClient, device.Name, device.NodeID)
 	if err != nil {
 		log.Printf("failed to add device. %v", err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -34,7 +34,7 @@ func (h *Handler) RemoveDevice(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err := cephmgr.RemoveDesiredDevice(h.EtcdClient, device)
+	err := cephmgr.RemoveDesiredDevice(h.context.EtcdClient, device.Name, device.NodeID)
 	if err != nil {
 		log.Printf("failed to remove device. %v", err)
 		w.WriteHeader(http.StatusInternalServerError)

pkg/api/handlers.go

@@ -7,20 +7,20 @@ import (
 	"log"
 	"net/http"
 
-	etcd "github.com/coreos/etcd/client"
 	"github.com/quantum/castle/pkg/cephmgr"
 	ceph "github.com/quantum/castle/pkg/cephmgr/client"
+	"github.com/quantum/castle/pkg/clusterd"
 )
 
 type Handler struct {
-	EtcdClient        etcd.KeysAPI
+	context           *clusterd.Context
 	ConnectionFactory cephmgr.ConnectionFactory
 	CephFactory       ceph.ConnectionFactory
 }
 
-func NewHandler(etcdClient etcd.KeysAPI, connFactory cephmgr.ConnectionFactory, cephFactory ceph.ConnectionFactory) *Handler {
+func NewHandler(context *clusterd.Context, connFactory cephmgr.ConnectionFactory, cephFactory ceph.ConnectionFactory) *Handler {
 	return &Handler{
-		EtcdClient:        etcdClient,
+		context:           context,
 		ConnectionFactory: connFactory,
 		CephFactory:       cephFactory,
 	}
@@ -73,7 +73,7 @@ func (h *Handler) GetCrushMap(w http.ResponseWriter, r *http.Request) {
 // /mon
 func (h *Handler) GetMonitors(w http.ResponseWriter, r *http.Request) {
 
-	desiredMons, err := cephmgr.GetDesiredMonitors(h.EtcdClient)
+	desiredMons, err := cephmgr.GetDesiredMonitors(h.context.EtcdClient)
 	if err != nil {
 		log.Printf("failed to load monitors: %+v", err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -130,7 +130,7 @@ func handleReadBody(w http.ResponseWriter, r *http.Request, opName string) ([]by
 }
 
 func (h *Handler) handleConnectToCeph(w http.ResponseWriter) (ceph.Connection, bool) {
-	adminConn, err := h.ConnectionFactory.ConnectAsAdmin(h.CephFactory, h.EtcdClient)
+	adminConn, err := h.ConnectionFactory.ConnectAsAdmin(h.context, h.CephFactory)
 	if err != nil {
 		log.Printf("failed to connect to cluster as admin: %+v", err)
 		w.WriteHeader(http.StatusInternalServerError)

pkg/api/handlers_test.go

@@ -15,6 +15,7 @@ import (
 	ceph "github.com/quantum/castle/pkg/cephmgr/client"
 	testceph "github.com/quantum/castle/pkg/cephmgr/client/test"
 	"github.com/quantum/castle/pkg/cephmgr/test"
+	"github.com/quantum/castle/pkg/clusterd"
 	"github.com/quantum/castle/pkg/clusterd/inventory"
 	"github.com/quantum/castle/pkg/util"
 	"github.com/stretchr/testify/assert"
@@ -26,12 +27,15 @@ const (
 )
 
 func TestAddRemoveDeviceHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+	etcdClient.SetValue("/castle/nodes/config/123/disks/myserial/name", "foo")
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/device", strings.NewReader(`{"name":"foo"}`))
 	assert.Nil(t, err)
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
-	h := NewHandler(etcdClient, nil, nil)
+	h := NewHandler(context, nil, nil)
 
 	// missing the node id
 	h.AddDevice(w, req)
@@ -40,19 +44,19 @@ func TestAddRemoveDeviceHandler(t *testing.T) {
 	// 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(etcdClient, nil, nil)
+	h = NewHandler(context, nil, nil)
 	w = httptest.NewRecorder()
 	h.AddDevice(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 
 	devices := etcdClient.GetChildDirs("/castle/services/ceph/osd/desired/123/device")
 	assert.Equal(t, 1, devices.Count())
-	assert.True(t, devices.Contains("foo"))
+	assert.True(t, devices.Contains("myserial"))
 
 	// 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(etcdClient, nil, nil)
+	h = NewHandler(context, nil, nil)
 	w = httptest.NewRecorder()
 	h.RemoveDevice(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
@@ -66,16 +70,18 @@ func TestRemoveDeviceHandler(t *testing.T) {
 }
 
 func TestGetNodesHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("GET", "http://10.0.0.100/node", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	etcdClient.SetValue("/castle/services/ceph/name", "cluster5")
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
-	h := NewHandler(etcdClient, &test.MockConnectionFactory{}, cephFactory)
+	h := NewHandler(context, &test.MockConnectionFactory{}, cephFactory)
 
 	// no nodes discovered, should return empty set
 	h.GetNodes(w, req)
@@ -87,13 +93,13 @@ func TestGetNodesHandler(t *testing.T) {
 	inventory.SetLocation(etcdClient, "node1", "root=default,dc=datacenter1")
 	nodeConfigKey := path.Join(inventory.NodesConfigKey, "node1")
 	etcdClient.CreateDir(nodeConfigKey)
-	inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "MB2CK3F6S5041EPCPJ4T", "sda", "506d4869-29ee-4bfd-bf21-dfd597bd222e",
+	inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "serial1", "sda", "506d4869-29ee-4bfd-bf21-dfd597bd222e",
 		100, true, false, "btrfs", "/mnt/abc", inventory.Disk, "", false)
-	inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "2B9C7KZN3VBM77PSA63P", "sdb", "506d4869-29ee-4bfd-bf21-dfd597bd222e",
+	inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "serial2", "sdb", "506d4869-29ee-4bfd-bf21-dfd597bd222e",
 		50, false, false, "ext4", "/mnt/def", inventory.Disk, "", false)
-	appliedOSDKey := "/castle/services/ceph/osd/applied/node1/devices"
-	etcdClient.SetValue(path.Join(appliedOSDKey, "sda", "serial"), "MB2CK3F6S5041EPCPJ4T")
-	etcdClient.SetValue(path.Join(appliedOSDKey, "sdb", "serial"), "2B9C7KZN3VBM77PSA63P")
+	appliedOSDKey := "/castle/services/ceph/osd/applied/node1/device"
+	etcdClient.CreateDir(path.Join(appliedOSDKey, "serial1"))
+	etcdClient.CreateDir(path.Join(appliedOSDKey, "serial2"))
 
 	// since a node exists (with storage), it should be returned now
 	w = httptest.NewRecorder()
@@ -105,20 +111,21 @@ func TestGetNodesHandler(t *testing.T) {
 }
 
 func TestGetNodesHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
 	req, err := http.NewRequest("GET", "http://10.0.0.100/node", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 
 	// failure during node lookup, should return an error status code
 	etcdClient.MockGet = func(context ctx.Context, key string, opts *etcd.GetOptions) (*etcd.Response, error) {
 		return nil, fmt.Errorf("mock etcd GET error")
 	}
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
-	h := NewHandler(etcdClient, &test.MockConnectionFactory{}, cephFactory)
+	h := NewHandler(context, &test.MockConnectionFactory{}, cephFactory)
 	h.GetNodes(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
 	assert.Equal(t, ``, w.Body.String())
@@ -126,6 +133,7 @@ func TestGetNodesHandlerFailure(t *testing.T) {
 
 func TestGetMonsHandler(t *testing.T) {
 	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 
 	req, err := http.NewRequest("GET", "http://10.0.0.100/mon", nil)
@@ -135,7 +143,7 @@ func TestGetMonsHandler(t *testing.T) {
 	w := httptest.NewRecorder()
 
 	// no mons will be returned, should be empty output
-	h := NewHandler(etcdClient, &test.MockConnectionFactory{}, cephFactory)
+	h := NewHandler(context, &test.MockConnectionFactory{}, cephFactory)
 	h.GetMonitors(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, `[]`, w.Body.String())
@@ -148,13 +156,16 @@ func TestGetMonsHandler(t *testing.T) {
 
 	// monitors should be returned now, verify the output
 	w = httptest.NewRecorder()
-	h = NewHandler(etcdClient, &test.MockConnectionFactory{}, cephFactory)
+	h = NewHandler(context, &test.MockConnectionFactory{}, cephFactory)
 	h.GetMonitors(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, "{\"status\":{\"quorum\":[0],\"monmap\":{\"mons\":[{\"name\":\"mon0\",\"rank\":0,\"addr\":\"10.37.129.87:6790\"}]}},\"desired\":[{\"name\":\"mon0\",\"endpoint\":\"1.2.3.4:8765\"}]}", w.Body.String())
 }
 
 func TestGetPoolsHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("GET", "http://10.0.0.100/pool", nil)
 	if err != nil {
 		log.Fatal(err)
@@ -162,7 +173,6 @@ func TestGetPoolsHandler(t *testing.T) {
 
 	// first return no storage pools
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
 	cephFactory.Conn = &testceph.MockConnection{
@@ -170,12 +180,12 @@ func TestGetPoolsHandler(t *testing.T) {
 			return []byte(`[]`), "", nil
 		},
 	}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
 	// no storage pools will be returned, should be empty output
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetPools(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, `[]`, w.Body.String())
@@ -207,13 +217,16 @@ func TestGetPoolsHandler(t *testing.T) {
 	}
 
 	// storage pools should be returned now, verify the output
-	h = NewHandler(etcdClient, connFactory, cephFactory)
+	h = NewHandler(context, connFactory, cephFactory)
 	h.GetPools(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, "[{\"poolName\":\"rbd\",\"poolNum\":0,\"type\":0,\"replicationConfig\":{\"size\":1},\"erasureCodedConfig\":{\"dataChunkCount\":0,\"codingChunkCount\":0,\"algorithm\":\"\"}},{\"poolName\":\"ecPool1\",\"poolNum\":1,\"type\":1,\"replicationConfig\":{\"size\":0},\"erasureCodedConfig\":{\"dataChunkCount\":2,\"codingChunkCount\":1,\"algorithm\":\"jerasure::reed_sol_van\"}}]", w.Body.String())
 }
 
 func TestGetPoolsHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("GET", "http://10.0.0.100/pool", nil)
 	if err != nil {
 		log.Fatal(err)
@@ -221,20 +234,22 @@ func TestGetPoolsHandlerFailure(t *testing.T) {
 
 	// encounter an error during GetPools
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return nil, fmt.Errorf("mock error for connect as admin")
 	}
 
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetPools(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
 	assert.Equal(t, "", w.Body.String())
 }
 
 func TestCreatePoolHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/pool",
 		strings.NewReader(`{"poolName":"ecPool1","poolNum":0,"type":1,"replicationConfig":{"size":0},"erasureCodedConfig":{"dataChunkCount":2,"codingChunkCount":1,"algorithm":""}}`))
 	if err != nil {
@@ -242,7 +257,6 @@ func TestCreatePoolHandler(t *testing.T) {
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
 	cephFactory.Conn = &testceph.MockConnection{
@@ -261,11 +275,11 @@ func TestCreatePoolHandler(t *testing.T) {
 			return nil, "", fmt.Errorf("unexpected mon_command '%s'", string(args))
 		},
 	}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 
 	h.CreatePool(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
@@ -273,13 +287,15 @@ func TestCreatePoolHandler(t *testing.T) {
 }
 
 func TestCreatePoolHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/pool", strings.NewReader(`{"poolname":"pool1"}`))
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
 	cephFactory.Conn = &testceph.MockConnection{
@@ -288,11 +304,11 @@ func TestCreatePoolHandlerFailure(t *testing.T) {
 			return []byte(""), errMsg, fmt.Errorf(errMsg)
 		},
 	}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 
 	h.CreatePool(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
@@ -300,6 +316,9 @@ func TestCreatePoolHandlerFailure(t *testing.T) {
 }
 
 func TestGetImagesHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("GET", "http://10.0.0.100/image", nil)
 	if err != nil {
 		log.Fatal(err)
@@ -307,7 +326,6 @@ func TestGetImagesHandler(t *testing.T) {
 
 	// first return no storage pools, which means no images will be returned either
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	cephFactory.Conn = &testceph.MockConnection{
 		MockMonCommand: func(args []byte) (buffer []byte, info string, err error) {
@@ -315,12 +333,12 @@ func TestGetImagesHandler(t *testing.T) {
 		},
 	}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
 	// no images will be returned, should be empty output
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetImages(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, `[]`, w.Body.String())
@@ -355,19 +373,21 @@ func TestGetImagesHandler(t *testing.T) {
 	}
 
 	// verify that the expected images are returned
-	h = NewHandler(etcdClient, connFactory, cephFactory)
+	h = NewHandler(context, connFactory, cephFactory)
 	h.GetImages(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, "[{\"imageName\":\"image1 - pool0\",\"poolName\":\"pool0\",\"size\":100,\"device\":\"\",\"mountPoint\":\"\"},{\"imageName\":\"image1 - pool1\",\"poolName\":\"pool1\",\"size\":100,\"device\":\"\",\"mountPoint\":\"\"}]", w.Body.String())
 }
 
 func TestGetImagesHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("GET", "http://10.0.0.100/image", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{
 		Fsid:      "myfsid",
 		SecretKey: "mykey",
@@ -378,34 +398,36 @@ func TestGetImagesHandlerFailure(t *testing.T) {
 		},
 	}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
 	// GetImages should fail due to the mocked error for listing pools
 	w := httptest.NewRecorder()
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetImages(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
 	assert.Equal(t, ``, w.Body.String())
 }
 
 func TestCreateImageHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/image", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 
 	// image is missing from request body, should be bad request
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.CreateImage(w, req)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assert.Equal(t, ``, w.Body.String())
@@ -416,7 +438,7 @@ func TestCreateImageHandler(t *testing.T) {
 		log.Fatal(err)
 	}
 	w = httptest.NewRecorder()
-	h = NewHandler(etcdClient, connFactory, cephFactory)
+	h = NewHandler(context, connFactory, cephFactory)
 	h.CreateImage(w, req)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assert.Equal(t, ``, w.Body.String())
@@ -427,7 +449,7 @@ func TestCreateImageHandler(t *testing.T) {
 		log.Fatal(err)
 	}
 	w = httptest.NewRecorder()
-	h = NewHandler(etcdClient, connFactory, cephFactory)
+	h = NewHandler(context, connFactory, cephFactory)
 	h.CreateImage(w, req)
 	assert.Equal(t, http.StatusBadRequest, w.Code)
 	assert.Equal(t, ``, w.Body.String())
@@ -447,23 +469,25 @@ func TestCreateImageHandler(t *testing.T) {
 		},
 	}
 	w = httptest.NewRecorder()
-	h = NewHandler(etcdClient, connFactory, cephFactory)
+	h = NewHandler(context, connFactory, cephFactory)
 	h.CreateImage(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, `succeeded created image myImage1`, w.Body.String())
 }
 
 func TestCreateImageHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/image", strings.NewReader(`{"imageName":"myImage1","poolName":"myPool1","size":1024}`))
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 	cephFactory.Conn = &testceph.MockConnection{
@@ -478,23 +502,25 @@ func TestCreateImageHandlerFailure(t *testing.T) {
 	}
 
 	// create image request should fail while creating the image
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.CreateImage(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
 	assert.Equal(t, ``, w.Body.String())
 }
 
 func TestGetImageMapInfoHandler(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/image/mapinfo", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 	cephFactory.Conn = &testceph.MockConnection{
@@ -510,29 +536,31 @@ func TestGetImageMapInfoHandler(t *testing.T) {
 	}
 
 	// get image map info and verify the response
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetImageMapInfo(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
 	assert.Equal(t, "{\"monAddresses\":[\"10.37.129.87:6790\"],\"userName\":\"admin\",\"secretKey\":\"AQBsCv1X5oD9GhAARHVU9N+kFRWDjyLA1dqzIg==\"}", w.Body.String())
 }
 
 func TestGetImageMapInfoHandlerFailure(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+	context := &clusterd.Context{EtcdClient: etcdClient}
+
 	req, err := http.NewRequest("POST", "http://10.0.0.100/image/mapinfo", nil)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	w := httptest.NewRecorder()
-	etcdClient := util.NewMockEtcdClient()
 	cephFactory := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
 	connFactory := &test.MockConnectionFactory{}
-	connFactory.MockConnectAsAdmin = func(cephFactory ceph.ConnectionFactory, etcdClient etcd.KeysAPI) (ceph.Connection, error) {
+	connFactory.MockConnectAsAdmin = func(context *clusterd.Context, cephFactory ceph.ConnectionFactory) (ceph.Connection, error) {
 		return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")
 	}
 	cephFactory.Conn = &testceph.MockConnection{}
 
 	// get image map info should fail becuase there's no mock response set up for auth get-key
-	h := NewHandler(etcdClient, connFactory, cephFactory)
+	h := NewHandler(context, connFactory, cephFactory)
 	h.GetImageMapInfo(w, req)
 	assert.Equal(t, http.StatusInternalServerError, w.Code)
 	assert.Equal(t, "", w.Body.String())

pkg/api/node.go

@@ -14,14 +14,14 @@ import (
 // GET
 // /node
 func (h *Handler) GetNodes(w http.ResponseWriter, r *http.Request) {
-	clusterInventory, err := inventory.LoadDiscoveredNodes(h.EtcdClient)
+	clusterInventory, err := inventory.LoadDiscoveredNodes(h.context.EtcdClient)
 	if err != nil {
 		log.Printf("failed to load discovered nodes: %+v", err)
 		w.WriteHeader(http.StatusInternalServerError)
 		return
 	}
 
-	clusterName, err := cephmgr.GetClusterName(h.EtcdClient)
+	clusterName, err := cephmgr.GetClusterName(h.context.EtcdClient)
 	if err != nil {
 		log.Printf("failed to get cluster name: %+v", err)
 		w.WriteHeader(http.StatusInternalServerError)
@@ -32,7 +32,7 @@ func (h *Handler) GetNodes(w http.ResponseWriter, r *http.Request) {
 	i := 0
 	for nodeID, n := range clusterInventory.Nodes {
 		// look up all the disks that the current node has applied OSDs on
-		appliedSerials, err := cephmgr.GetAppliedOSDs(nodeID, h.EtcdClient)
+		appliedSerials, err := cephmgr.GetAppliedOSDs(nodeID, h.context.EtcdClient)
 		if err != nil {
 			log.Printf("failed to get applied OSDs for node %s: %+v", nodeID, err)
 			w.WriteHeader(http.StatusInternalServerError)
@@ -41,7 +41,7 @@ func (h *Handler) GetNodes(w http.ResponseWriter, r *http.Request) {
 
 		storage := uint64(0)
 		for _, d := range n.Disks {
-			for _, s := range appliedSerials {
+			for s := range appliedSerials.Iter() {
 				if s == d.Serial {
 					// current disk is in applied OSD set, add its storage to the running total
 					storage += d.Size

pkg/cephmgr/cephleader.go

@@ -89,7 +89,7 @@ func (c *cephLeader) handleOrchestratorEvents() {
 			// Perform a full refresh of the cluster to ensure the monitors are running with quorum
 			err := c.configureCephMons(e.Context())
 			if err != nil {
-				log.Printf("failed to configure ceph mons. %v", err)
+				log.Printf("FAILED TO CONFIGURE CEPH MONS. %v", err)
 				continue
 			}
 		}
@@ -98,7 +98,7 @@ func (c *cephLeader) handleOrchestratorEvents() {
 			// Configure the OSDs
 			err := configureOSDs(e.Context(), osdsToRefresh)
 			if err != nil {
-				log.Printf("failed to configure ceph osds. %v", err)
+				log.Printf("FAILED TO CONFIGURE CEPH OSDs. %v", err)
 				continue
 			}
 		}

pkg/cephmgr/cephleader_test.go

@@ -39,6 +39,7 @@ func TestCephLeaders(t *testing.T) {
 	context := &clusterd.Context{
 		EtcdClient: etcdClient,
 		Inventory:  inv,
+		ConfigDir:  "/tmp",
 	}
 
 	// mock the agent responses that the deployments were successful to start mons and osds
@@ -89,6 +90,7 @@ func TestMoveUnhealthyMonitor(t *testing.T) {
 	context := &clusterd.Context{
 		EtcdClient: etcdClient,
 		Inventory:  inv,
+		ConfigDir:  "/tmp",
 	}
 
 	// mock the agent responses that the deployments were successful to start mons and osds

pkg/cephmgr/config.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-ini/ini"
 	"github.com/quantum/castle/pkg/cephmgr/client"
+	"github.com/quantum/castle/pkg/clusterd"
 )
 
 type CephMonitorConfig struct {
@@ -22,6 +23,7 @@ type cephConfig struct {
 }
 
 type cephGlobalConfig struct {
+	EnableExperimental       string `ini:"enable experimental unrecoverable data corrupting features"`
 	FSID                     string `ini:"fsid,omitempty"`
 	RunDir                   string `ini:"run dir,omitempty"`
 	MonMembers               string `ini:"mon initial members,omitempty"`
@@ -30,6 +32,7 @@ type cephGlobalConfig struct {
 	DebugLogDefaultLevel     int    `ini:"debug default"`
 	DebugLogRadosLevel       int    `ini:"debug rados"`
 	DebugLogMonLevel         int    `ini:"debug mon"`
+	DebugLogBluestoreLevel   int    `ini:"debug bluestore"`
 	OsdPgBits                int    `ini:"osd pg bits,omitempty"`
 	OsdPgpBits               int    `ini:"osd pgp bits,omitempty"`
 	OsdPoolDefaultSize       int    `ini:"osd pool default size,omitempty"`
@@ -38,20 +41,21 @@ type cephGlobalConfig struct {
 	OsdPoolDefaultPgpNum     int    `ini:"osd pool default pgp num,omitempty"`
 	OsdMaxObjectNameLen      int    `ini:"osd max object name len,omitempty"`
 	OsdMaxObjectNamespaceLen int    `ini:"osd max object namespace len,omitempty"`
+	OsdObjectStore           string `ini:"osd objectstore"`
 	RbdDefaultFeatures       int    `ini:"rbd_default_features,omitempty"`
 }
 
-func ConnectToClusterAsAdmin(factory client.ConnectionFactory, cluster *ClusterInfo) (client.Connection, error) {
+func ConnectToClusterAsAdmin(context *clusterd.Context, factory client.ConnectionFactory, cluster *ClusterInfo) (client.Connection, error) {
 	if len(cluster.Monitors) == 0 {
 		return nil, errors.New("no monitors")
 	}
 	// write the monitor keyring to disk
 	monName := getFirstMonitor(cluster)
-	if err := writeMonitorKeyring(monName, cluster); err != nil {
+	if err := writeMonitorKeyring(context.ConfigDir, monName, cluster); err != nil {
 		return nil, err
 	}
 
-	return connectToCluster(factory, cluster, getMonRunDirPath(monName), "admin", getMonKeyringPath(monName))
+	return connectToCluster(context, factory, cluster, getMonRunDirPath(context.ConfigDir, monName), "admin", getMonKeyringPath(context.ConfigDir, monName))
 }
 
 // get the path of a given monitor's config file
@@ -60,9 +64,14 @@ func getConfFilePath(root, clusterName string) string {
 }
 
 // generates and writes the monitor config file to disk
-func generateConfigFile(cluster *ClusterInfo, pathRoot, user, keyringPath string, userConfig *cephConfig) (string, error) {
+func generateConnectionConfigFile(context *clusterd.Context, cluster *ClusterInfo, pathRoot, user, keyringPath string) (string, error) {
+	return generateConfigFile(context, cluster, pathRoot, user, keyringPath, false, nil, nil)
+}
+
+// generates and writes the monitor config file to disk
+func generateConfigFile(context *clusterd.Context, cluster *ClusterInfo, pathRoot, user, keyringPath string, bluestore bool, userConfig *cephConfig, clientSettings map[string]string) (string, error) {
 	if pathRoot == "" {
-		pathRoot = getMonRunDirPath(getFirstMonitor(cluster))
+		pathRoot = getMonRunDirPath(context.ConfigDir, getFirstMonitor(cluster))
 	}
 
 	// create the config directory
@@ -70,12 +79,12 @@ func generateConfigFile(cluster *ClusterInfo, pathRoot, user, keyringPath string
 		fmt.Printf("failed to create config directory at %s: %+v", pathRoot, err)
 	}
 
-	configFile, err := createGlobalConfigFileSection(cluster, pathRoot, userConfig)
+	configFile, err := createGlobalConfigFileSection(cluster, pathRoot, bluestore, userConfig)
 	if err != nil {
 		return "", fmt.Errorf("failed to create global config section, %+v", err)
 	}
 
-	if err := addClientConfigFileSection(configFile, getQualifiedUser(user), keyringPath); err != nil {
+	if err := addClientConfigFileSection(configFile, getQualifiedUser(user), keyringPath, clientSettings); err != nil {
 		return "", fmt.Errorf("failed to add admin client config section, %+v", err)
 	}
 
@@ -111,10 +120,10 @@ func getFirstMonitor(cluster *ClusterInfo) string {
 }
 
 // opens a connection to the cluster that can be used for management operations
-func connectToCluster(factory client.ConnectionFactory, cluster *ClusterInfo, basePath, user, keyringPath string) (client.Connection, error) {
+func connectToCluster(context *clusterd.Context, factory client.ConnectionFactory, cluster *ClusterInfo, basePath, user, keyringPath string) (client.Connection, error) {
 	log.Printf("connecting to ceph cluster %s with user %s", cluster.Name, user)
 
-	confFilePath, err := generateConfigFile(cluster, basePath, user, keyringPath, nil)
+	confFilePath, err := generateConnectionConfigFile(context, cluster, basePath, user, keyringPath)
 	if err != nil {
 		return nil, fmt.Errorf("failed to generate config file: %v", err)
 	}
@@ -135,7 +144,7 @@ func connectToCluster(factory client.ConnectionFactory, cluster *ClusterInfo, ba
 	return conn, nil
 }
 
-func createDefaultCephConfig(cluster *ClusterInfo, runDir string) *cephConfig {
+func createDefaultCephConfig(cluster *ClusterInfo, runDir string, bluestore bool) *cephConfig {
 	// extract a list of just the monitor names, which will populate the "mon initial members"
 	// global config field
 	monMembers := make([]string, len(cluster.Monitors))
@@ -145,35 +154,45 @@ func createDefaultCephConfig(cluster *ClusterInfo, runDir string) *cephConfig {
 		i++
 	}
 
+	experimental := ""
+	store := "filestore"
+	if bluestore {
+		experimental = "bluestore rocksdb"
+		store = "bluestore"
+	}
+
 	return &cephConfig{
 		&cephGlobalConfig{
-			FSID:                  cluster.FSID,
-			RunDir:                runDir,
-			MonMembers:            strings.Join(monMembers, " "),
-			LogFile:               "/dev/stdout",
-			MonClusterLogFile:     "/dev/stdout",
-			DebugLogDefaultLevel:  0,
-			DebugLogRadosLevel:    0,
-			DebugLogMonLevel:      0,
-			OsdPgBits:             11,
-			OsdPgpBits:            11,
-			OsdPoolDefaultSize:    1,
-			OsdPoolDefaultMinSize: 1,
-			OsdPoolDefaultPgNum:   100,
-			OsdPoolDefaultPgpNum:  100,
-			RbdDefaultFeatures:    3,
+			EnableExperimental:     experimental,
+			FSID:                   cluster.FSID,
+			RunDir:                 runDir,
+			MonMembers:             strings.Join(monMembers, " "),
+			LogFile:                "/dev/stdout",
+			MonClusterLogFile:      "/dev/stdout",
+			DebugLogDefaultLevel:   0,
+			DebugLogRadosLevel:     0,
+			DebugLogMonLevel:       0,
+			DebugLogBluestoreLevel: 0,
+			OsdPgBits:              11,
+			OsdPgpBits:             11,
+			OsdPoolDefaultSize:     1,
+			OsdPoolDefaultMinSize:  1,
+			OsdPoolDefaultPgNum:    100,
+			OsdPoolDefaultPgpNum:   100,
+			OsdObjectStore:         store,
+			RbdDefaultFeatures:     3,
 		},
 	}
 }
 
-func createGlobalConfigFileSection(cluster *ClusterInfo, runDir string, userConfig *cephConfig) (*ini.File, error) {
+func createGlobalConfigFileSection(cluster *ClusterInfo, runDir string, bluestore bool, userConfig *cephConfig) (*ini.File, error) {
 	var ceph *cephConfig
 
 	if userConfig != nil {
 		// use the user config since it was provided
 		ceph = userConfig
 	} else {
-		ceph = createDefaultCephConfig(cluster, runDir)
+		ceph = createDefaultCephConfig(cluster, runDir, bluestore)
 	}
 
 	configFile := ini.Empty()
@@ -181,7 +200,7 @@ func createGlobalConfigFileSection(cluster *ClusterInfo, runDir string, userConf
 	return configFile, err
 }
 
-func addClientConfigFileSection(configFile *ini.File, clientName, keyringPath string) error {
+func addClientConfigFileSection(configFile *ini.File, clientName, keyringPath string, settings map[string]string) error {
 	s, err := configFile.NewSection(clientName)
 	if err != nil {
 		return err
@@ -191,6 +210,12 @@ func addClientConfigFileSection(configFile *ini.File, clientName, keyringPath st
 		return err
 	}
 
+	for key, val := range settings {
+		if _, err := s.NewKey(key, val); err != nil {
+			return fmt.Errorf("failed to add key %s. %v", key, err)
+		}
+	}
+
 	return nil
 }
 

pkg/cephmgr/conn.go

@@ -1,12 +1,12 @@
 package cephmgr
 
 import (
-	etcd "github.com/coreos/etcd/client"
 	"github.com/quantum/castle/pkg/cephmgr/client"
+	"github.com/quantum/castle/pkg/clusterd"
 )
 
 type ConnectionFactory interface {
-	ConnectAsAdmin(cephFactory client.ConnectionFactory, etcdClient etcd.KeysAPI) (client.Connection, error)
+	ConnectAsAdmin(context *clusterd.Context, cephFactory client.ConnectionFactory) (client.Connection, error)
 }
 
 type castleConnFactory struct {
@@ -15,14 +15,14 @@ type castleConnFactory struct {
 func NewConnectionFactory() ConnectionFactory { return &castleConnFactory{} }
 
 func (c *castleConnFactory) ConnectAsAdmin(
-	cephFactory client.ConnectionFactory, etcdClient etcd.KeysAPI) (client.Connection, error) {
+	context *clusterd.Context, cephFactory client.ConnectionFactory) (client.Connection, error) {
 
 	// load information about the cluster
-	cluster, err := LoadClusterInfo(etcdClient)
+	cluster, err := LoadClusterInfo(context.EtcdClient)
 	if err != nil {
 		return nil, err
 	}
 
 	// open an admin connection to the cluster
-	return ConnectToClusterAsAdmin(cephFactory, cluster)
+	return ConnectToClusterAsAdmin(context, cephFactory, cluster)
 }

pkg/cephmgr/conn_test.go

@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	testceph "github.com/quantum/castle/pkg/cephmgr/client/test"
+	"github.com/quantum/castle/pkg/clusterd"
 	"github.com/quantum/castle/pkg/util"
 	"github.com/stretchr/testify/assert"
 )
@@ -13,8 +14,9 @@ func TestBasicConn(t *testing.T) {
 	createTestClusterInfo(etcdClient, []string{"a"})
 	factory := NewConnectionFactory()
 	fact := &testceph.MockConnectionFactory{Fsid: "myfsid", SecretKey: "mykey"}
+	context := &clusterd.Context{EtcdClient: etcdClient, ConfigDir: "/tmp"}
 
-	conn, err := factory.ConnectAsAdmin(fact, etcdClient)
+	conn, err := factory.ConnectAsAdmin(context, fact)
 	assert.Nil(t, err)
 	assert.NotNil(t, conn)
 }

pkg/cephmgr/mon.go

@@ -27,21 +27,16 @@ func getMonitorEndpointKey(name string) string {
 }
 
 // get the path of a given monitor's run dir
-func getMonRunDirPath(monName string) string {
-	return fmt.Sprintf("/tmp/%s", monName)
-}
-
-// get the path of a given monitor's config file
-func getMonConfFilePath(monName string) string {
-	return filepath.Join(getMonRunDirPath(monName), "config")
+func getMonRunDirPath(configDir, monName string) string {
+	return path.Join(configDir, monName)
 }
 
 // get the path of a given monitor's keyring
-func getMonKeyringPath(monName string) string {
-	return filepath.Join(getMonRunDirPath(monName), "keyring")
+func getMonKeyringPath(configDir, monName string) string {
+	return filepath.Join(getMonRunDirPath(configDir, monName), "keyring")
 }
 
 // get the path of a given monitor's data dir
-func getMonDataDirPath(monName string) string {
-	return filepath.Join(getMonRunDirPath(monName), fmt.Sprintf("mon.%s", monName))
+func getMonDataDirPath(configDir, monName string) string {
+	return filepath.Join(getMonRunDirPath(configDir, monName), fmt.Sprintf("mon.%s", monName))
 }

pkg/cephmgr/monagent.go

@@ -93,18 +93,18 @@ func (a *monAgent) DestroyLocalService(context *clusterd.Context) error {
 // creates and initializes the given monitors file systems
 func (a *monAgent) makeMonitorFileSystem(context *clusterd.Context, cluster *ClusterInfo, monName string) error {
 	// write the keyring to disk
-	if err := writeMonitorKeyring(monName, cluster); err != nil {
+	if err := writeMonitorKeyring(context.ConfigDir, monName, cluster); err != nil {
 		return err
 	}
 
 	// write the config file to disk
-	confFilePath, err := generateConfigFile(cluster, getMonRunDirPath(monName), "admin", getMonKeyringPath(monName), nil)
+	confFilePath, err := generateConnectionConfigFile(context, cluster, getMonRunDirPath(context.ConfigDir, monName), "admin", getMonKeyringPath(context.ConfigDir, monName))
 	if err != nil {
 		return err
 	}
 
 	// create monitor data dir
-	monDataDir := getMonDataDirPath(monName)
+	monDataDir := getMonDataDirPath(context.ConfigDir, monName)
 	if err := os.MkdirAll(filepath.Dir(monDataDir), 0744); err != nil {
 		fmt.Printf("failed to create monitor data directory at %s: %+v", monDataDir, err)
 	}
@@ -117,7 +117,7 @@ func (a *monAgent) makeMonitorFileSystem(context *clusterd.Context, cluster *Clu
 		fmt.Sprintf("--name=mon.%s", monName),
 		fmt.Sprintf("--mon-data=%s", monDataDir),
 		fmt.Sprintf("--conf=%s", confFilePath),
-		fmt.Sprintf("--keyring=%s", getMonKeyringPath(monName)))
+		fmt.Sprintf("--keyring=%s", getMonKeyringPath(context.ConfigDir, monName)))
 	if err != nil {
 		return fmt.Errorf("failed mon %s --mkfs: %+v", monName, err)
 	}
@@ -141,8 +141,8 @@ func (a *monAgent) runMonitor(context *clusterd.Context, cluster *ClusterInfo, m
 		"--foreground",
 		fmt.Sprintf("--cluster=%s", cluster.Name),
 		monNameArg,
-		fmt.Sprintf("--mon-data=%s", getMonDataDirPath(monitor.Name)),
-		fmt.Sprintf("--conf=%s", getConfFilePath(getMonRunDirPath(monitor.Name), cluster.Name)),
+		fmt.Sprintf("--mon-data=%s", getMonDataDirPath(context.ConfigDir, monitor.Name)),
+		fmt.Sprintf("--conf=%s", getConfFilePath(getMonRunDirPath(context.ConfigDir, monitor.Name), cluster.Name)),
 		fmt.Sprintf("--public-addr=%s", monitor.Endpoint))
 	if err != nil {
 		return fmt.Errorf("failed to start monitor %s: %v", monitor.Name, err)
@@ -157,9 +157,9 @@ func (a *monAgent) runMonitor(context *clusterd.Context, cluster *ClusterInfo, m
 }
 
 // writes the monitor keyring to disk
-func writeMonitorKeyring(monName string, c *ClusterInfo) error {
+func writeMonitorKeyring(configDir, monName string, c *ClusterInfo) error {
 	keyring := fmt.Sprintf(monitorKeyringTemplate, c.MonitorSecret, c.AdminSecret)
-	keyringPath := getMonKeyringPath(monName)
+	keyringPath := getMonKeyringPath(configDir, monName)
 	if err := os.MkdirAll(filepath.Dir(keyringPath), 0744); err != nil {
 		return fmt.Errorf("failed to create keyring directory for %s: %+v", keyringPath, err)
 	}

pkg/cephmgr/monagent_test.go

@@ -48,6 +48,7 @@ func TestMonAgent(t *testing.T) {
 		EtcdClient: etcdClient,
 		NodeID:     "a",
 		ProcMan:    &proc.ProcManager{Trap: procTrap},
+		ConfigDir:  "/tmp",
 	}
 
 	factory := &testceph.MockConnectionFactory{Fsid: "f", SecretKey: "k"}

pkg/cephmgr/monleader.go

@@ -80,7 +80,7 @@ func removeMonitorsFromQuorum(factory client.ConnectionFactory, context *cluster
 	}
 
 	// open an admin connection to the cluster
-	conn, err := ConnectToClusterAsAdmin(factory, cluster)
+	conn, err := ConnectToClusterAsAdmin(context, factory, cluster)
 	if err != nil {
 		return err
 	}
@@ -337,7 +337,7 @@ func calculateMonitorCount(nodeCount int) int {
 func waitForQuorum(factory client.ConnectionFactory, context *clusterd.Context, cluster *ClusterInfo) error {
 
 	// open an admin connection to the cluster
-	adminConn, err := ConnectToClusterAsAdmin(factory, cluster)
+	adminConn, err := ConnectToClusterAsAdmin(context, factory, cluster)
 	if err != nil {
 		return err
 	}

pkg/cephmgr/monleader_test.go

@@ -20,6 +20,7 @@ func TestMonSelection(t *testing.T) {
 	context := &clusterd.Context{
 		EtcdClient: etcdClient,
 		Inventory:  inv,
+		ConfigDir:  "/tmp",
 	}
 
 	// choose 1 mon from the set of 1 nodes
@@ -79,7 +80,7 @@ func TestMonOnUnhealthyNode(t *testing.T) {
 
 	// the monitor is on the bad node
 	badNode := &clusterd.UnhealthyNode{NodeID: "a"}
-	context := &clusterd.Context{EtcdClient: etcdClient}
+	context := &clusterd.Context{EtcdClient: etcdClient, ConfigDir: "/tmp"}
 	response, err := monsOnUnhealthyNode(context, []*clusterd.UnhealthyNode{badNode})
 	assert.True(t, response)
 	assert.Nil(t, err)
@@ -138,6 +139,7 @@ func TestUnhealthyMon(t *testing.T) {
 	context := &clusterd.Context{
 		EtcdClient: etcdClient,
 		Inventory:  inv,
+		ConfigDir:  "/tmp",
 	}
 
 	// choose 3 mons from the set of 4 nodes, but don't choose the unhealthy node 'a'

pkg/cephmgr/osd.go

@@ -12,13 +12,13 @@ import (
 	"strconv"
 	"strings"
 	"syscall"
-	"time"
 
 	etcd "github.com/coreos/etcd/client"
 	ctx "golang.org/x/net/context"
 
 	"github.com/google/uuid"
 	"github.com/quantum/castle/pkg/cephmgr/client"
+	"github.com/quantum/castle/pkg/cephmgr/partition"
 	"github.com/quantum/castle/pkg/clusterd"
 	"github.com/quantum/castle/pkg/clusterd/inventory"
 	"github.com/quantum/castle/pkg/util"
@@ -29,7 +29,11 @@ import (
 const (
 	DevicesValue                = "devices"
 	ForceFormatValue            = "forceFormat"
-	deviceDesiredKey            = "/castle/services/ceph/osd/desired/%s/device"
+	sgdisk                      = "sgdisk"
+	cephOsdKey                  = "/castle/services/ceph/osd"
+	desiredOsdRootKey           = cephOsdKey + "/" + desiredKey + "/%s"
+	deviceDesiredKey            = desiredOsdRootKey + "/device"
+	dirDesiredKey               = desiredOsdRootKey + "/dir"
 	bootstrapOSDKeyringTemplate = `
 [client.bootstrap-osd]
 	key = %s
@@ -40,52 +44,111 @@ const (
 type Device struct {
 	Name   string `json:"name"`
 	NodeID string `json:"nodeId"`
+	Dir    bool   `json:"bool"`
 }
 
-func loadDesiredDevices(etcdClient etcd.KeysAPI, nodeID string) (*util.Set, error) {
+func loadDesiredDevices(etcdClient etcd.KeysAPI, nodeID string) (map[string]string, error) {
 	key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID))
 	children, err := util.GetDirChildKeys(etcdClient, key)
 	if err != nil {
 		return nil, fmt.Errorf("could not get desired devices. %v", err)
 	}
 
-	return children, nil
+	devices := map[string]string{}
+	for serial := range children.Iter() {
+		device, err := inventory.GetDeviceFromSerial(serial, nodeID, etcdClient)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get device name for serial %s. %v", serial, err)
+		}
+
+		devices[device] = serial
+	}
+
+	return devices, nil
 }
 
 // add a device to the desired state
-func AddDesiredDevice(etcdClient etcd.KeysAPI, device *Device) error {
-	key := path.Join(fmt.Sprintf(deviceDesiredKey, device.NodeID), device.Name)
-	err := util.CreateEtcdDir(etcdClient, key)
+func AddDesiredDevice(etcdClient etcd.KeysAPI, device, nodeID string) error {
+	serial, err := inventory.GetSerialFromDevice(device, nodeID, etcdClient)
 	if err != nil {
-		return fmt.Errorf("failed to add device %s on node %s to desired. %v", device.Name, device.NodeID, err)
+		return fmt.Errorf("failed to get serial for device %s. %v", device, err)
+	}
+
+	key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), serial)
+	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 nil
+}
+
+func loadDesiredDirs(etcdClient etcd.KeysAPI, nodeID string) (*util.Set, error) {
+	dirs := util.NewSet()
+	key := path.Join(fmt.Sprintf(dirDesiredKey, nodeID))
+	dirKeys, err := etcdClient.Get(ctx.Background(), key, &etcd.GetOptions{Recursive: true})
+	if err != nil {
+		if util.IsEtcdKeyNotFound(err) {
+			return dirs, nil
+		}
+		return nil, err
+	}
+
+	// parse the dirs from etcd
+	for _, dev := range dirKeys.Node.Nodes {
+		for _, setting := range dev.Nodes {
+			if strings.HasSuffix(setting.Key, "/path") {
+				dirs.Add(setting.Value)
+			}
+		}
+	}
+
+	return dirs, nil
+}
+
+// add a device to the desired state
+func AddDesiredDir(etcdClient etcd.KeysAPI, dir, nodeID string) error {
+	key := path.Join(fmt.Sprintf(dirDesiredKey, nodeID), getPseudoDir(dir), "path")
+	_, err := etcdClient.Set(ctx.Background(), key, dir, nil)
+	if err != nil {
+		return fmt.Errorf("failed to add desired dir %s on node %s. %v", dir, nodeID, err)
 	}
 
 	return nil
 }
 
 // remove a device from the desired state
-func RemoveDesiredDevice(etcdClient etcd.KeysAPI, device *Device) error {
-	key := path.Join(fmt.Sprintf(deviceDesiredKey, device.NodeID), device.Name)
-	_, err := etcdClient.Delete(ctx.Background(), key, &etcd.DeleteOptions{Dir: true})
+func RemoveDesiredDevice(etcdClient etcd.KeysAPI, device, nodeID string) error {
+	serial, err := inventory.GetSerialFromDevice(device, nodeID, etcdClient)
 	if err != nil {
-		return fmt.Errorf("failed to remove device %s on node %s from desired. %v", device.Name, device.NodeID, err)
+		return fmt.Errorf("failed to get serial for device %s. %v", device, err)
+	}
+
+	key := path.Join(fmt.Sprintf(deviceDesiredKey, nodeID), serial)
+	_, 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 nil
 }
 
 // get the bootstrap OSD root dir
-func getBootstrapOSDDir() string {
-	return "/tmp/bootstrap-osd"
+func getBootstrapOSDDir(configDir string) string {
+	return path.Join(configDir, "bootstrap-osd")
 }
 
-func getOSDRootDir(osdID int) string {
-	return fmt.Sprintf("/tmp/osd%d", osdID)
+func getOSDConfigDir(configDir, serial string) string {
+	return fmt.Sprintf(path.Join(configDir, "device", serial))
+}
+
+func getOSDRootDir(root string, osdID int) string {
+	return fmt.Sprintf("%s/osd%d", root, osdID)
 }
 
 // get the full path to the bootstrap OSD keyring
-func getBootstrapOSDKeyringPath(clusterName string) string {
-	return fmt.Sprintf("%s/%s.keyring", getBootstrapOSDDir(), clusterName)
+func getBootstrapOSDKeyringPath(configDir, clusterName string) string {
+	return fmt.Sprintf("%s/%s.keyring", getBootstrapOSDDir(configDir), clusterName)
 }
 
 // get the full path to the given OSD's config file
@@ -109,8 +172,8 @@ func getOSDTempMonMapPath(osdDataPath string) string {
 }
 
 // create a keyring for the bootstrap-osd client, it gets a limited set of privileges
-func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error {
-	bootstrapOSDKeyringPath := getBootstrapOSDKeyringPath(clusterName)
+func createOSDBootstrapKeyring(conn client.Connection, configDir, clusterName string) error {
+	bootstrapOSDKeyringPath := getBootstrapOSDKeyringPath(configDir, clusterName)
 	_, err := os.Stat(bootstrapOSDKeyringPath)
 	if err == nil {
 		// no error, the file exists, bail out with no error
@@ -124,7 +187,7 @@ func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error
 	// get-or-create-key for client.bootstrap-osd
 	bootstrapOSDKey, err := client.AuthGetOrCreateKey(conn, "client.bootstrap-osd", []string{"mon", "allow profile bootstrap-osd"})
 	if err != nil {
-		return err
+		return fmt.Errorf("failed to get or create osd auth key %s. %+v", bootstrapOSDKeyringPath, err)
 	}
 
 	log.Printf("succeeded bootstrap OSD get/create key, bootstrapOSDKey: %s", bootstrapOSDKey)
@@ -132,8 +195,9 @@ func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error
 	// write the bootstrap-osd keyring to disk
 	bootstrapOSDKeyringDir := filepath.Dir(bootstrapOSDKeyringPath)
 	if err := os.MkdirAll(bootstrapOSDKeyringDir, 0744); err != nil {
-		fmt.Printf("failed to create bootstrap OSD keyring dir at %s: %+v", bootstrapOSDKeyringDir, err)
+		return fmt.Errorf("failed to create bootstrap OSD keyring dir at %s: %+v", bootstrapOSDKeyringDir, err)
 	}
+
 	bootstrapOSDKeyring := fmt.Sprintf(bootstrapOSDKeyringTemplate, bootstrapOSDKey)
 	log.Printf("Writing osd keyring to: %s", bootstrapOSDKeyring)
 	if err := ioutil.WriteFile(bootstrapOSDKeyringPath, []byte(bootstrapOSDKeyring), 0644); err != nil {
@@ -144,9 +208,9 @@ func createOSDBootstrapKeyring(conn client.Connection, clusterName string) error
 }
 
 // format the given device for usage by an OSD
-func formatOSD(device string, forceFormat bool, executor exec.Executor) error {
+func formatDevice(context *clusterd.Context, device, serial, configDir string, forceFormat bool) error {
 	// format the current volume
-	devFS, err := sys.GetDeviceFilesystem(device, executor)
+	devFS, err := sys.GetDeviceFilesystem(device, context.Executor)
 	if err != nil {
 		return fmt.Errorf("failed to get device %s filesystem: %+v", device, err)
 	}
@@ -157,12 +221,13 @@ func formatOSD(device string, forceFormat bool, executor exec.Executor) error {
 	}
 
 	if devFS == "" || forceFormat {
-		// execute the format operation
-		cmd := fmt.Sprintf("format %s", device)
-		err = executor.ExecuteCommand(cmd, "sudo", "/usr/sbin/mkfs.btrfs", "-f", "-m", "single", "-n", "32768", fmt.Sprintf("/dev/%s", device))
+		log.Printf("Partitioning device %s for bluestore", device)
+
+		err := partitionBluestoreDevice(context, device, serial, configDir)
 		if err != nil {
-			return fmt.Errorf("command %s failed: %+v", cmd, err)
+			return fmt.Errorf("failed to partion device %s. %v", device, err)
 		}
+
 	} else {
 		// disk is already formatted and the user doesn't want to force it, return no error, but also specify that no format was done
 		log.Printf("device %s already formatted with %s", device, devFS)
@@ -172,67 +237,47 @@ func formatOSD(device string, forceFormat bool, executor exec.Executor) error {
 	return nil
 }
 
-// mount the OSD data directory onto the given device
-func mountOSD(device string, mountPath string, executor exec.Executor) error {
-	cmd := fmt.Sprintf("lsblk %s", device)
-	var diskUUID string
-
-	retryCount := 0
-	retryMax := 10
-	sleepTime := 2
-	for {
-		// there is lag in between when a filesytem is created and its UUID is available.  retry as needed
-		// until we have a usable UUID for the newly formatted filesystem.
-		var err error
-		diskUUID, err = executor.ExecuteCommandWithOutput(cmd, "lsblk", fmt.Sprintf("/dev/%s", device), "-d", "-n", "-r", "-o", "UUID")
-		if err != nil {
-			return fmt.Errorf("command %s failed: %+v", cmd, err)
-		}
-
-		if diskUUID == "skip-UUID-verification" {
-			// skip verifying the uuid during tests
-			break
-
-		} else if diskUUID != "" {
-			// we got the UUID from the disk.  Verify this UUID is up to date in the /dev/disk/by-uuid dir by
-			// checking for it multiple times in a row.  For an existing device, the device UUID and the
-			// by-uuid link can take a bit to get updated after getting formatted.  Increase our confidence
-			// that we have the updated UUID by performing this check multiple times in a row.
-			log.Printf("verifying UUID %s", diskUUID)
-			uuidCheckOK := true
-			uuidCheckCount := 0
-			for uuidCheckCount < 3 {
-				uuidCheckCount++
-				if _, err := os.Stat(fmt.Sprintf("/dev/disk/by-uuid/%s", diskUUID)); os.IsNotExist(err) {
-					// the UUID we got for the disk does not exist under /dev/disk/by-uuid.  Retry.
-					uuidCheckOK = false
-					break
-				}
-				<-time.After(time.Duration(500) * time.Millisecond)
-			}
-
-			if uuidCheckOK {
-				log.Printf("device %s UUID created: %s", device, diskUUID)
-				break
-			}
-		}
-
-		retryCount++
-		if retryCount > retryMax {
-			return fmt.Errorf("exceeded max retry count waiting for device %s UUID to be created", device)
-		}
-
-		<-time.After(time.Duration(sleepTime) * time.Second)
+// 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, device, serial, configDir string) error {
+	size, err := inventory.GetSizeFromSerial(serial, context.NodeID, context.EtcdClient)
+	if err != nil {
+		return fmt.Errorf("failed to get device %s size. %+v", serial, err)
 	}
 
-	// mount the volume
-	if err := sys.MountDeviceWithOptions(
-		fmt.Sprintf("/dev/disk/by-uuid/%s", diskUUID), mountPath, "user_subvol_rm_allowed", executor); err != nil {
-		return err
+	cmd := fmt.Sprintf("zap partitions on %s", device)
+	err = context.Executor.ExecuteCommand(cmd, sgdisk, "--zap-all", "/dev/"+device)
+	if err != nil {
+		return fmt.Errorf("failed to zap partitions on /dev/%s: %+v", device, err)
 	}
 
-	// chown for the current user since we had to format and mount with sudo
-	sys.ChownForCurrentUser(mountPath, executor)
+	scheme, err := partition.GetSimpleScheme(int(size / 1024 / 1024))
+	if err != nil {
+		return fmt.Errorf("failed to get simple scheme. %+v", err)
+	}
+
+	// get args for creating partitions
+	args := scheme.GetArgs(device)
+	if err != nil {
+		return fmt.Errorf("failed to get partition args. %+v", err)
+	}
+
+	// execute the partition command
+	err = context.Executor.ExecuteCommand(cmd, sgdisk, args...)
+	if err != nil {
+		return fmt.Errorf("failed to partition /dev/%s. %+v", device, err)
+	}
+
+	// ensure the folder exists
+	if err := os.MkdirAll(configDir, 0744); err != nil {
+		log.Fatalf("failed to create config dir %s. %+v", configDir, err)
+	}
+
+	// save the scheme
+	err = scheme.Save(configDir)
+	if err != nil {
+		return fmt.Errorf("failed to save partition scheme. %+v", err)
+	}
 
 	return nil
 }
@@ -286,86 +331,119 @@ func findSubdirByPattern(pattern, dir string) (string, error) {
 	return "", nil
 }
 
-func getOSDInfo(osdDataPath string) (int, uuid.UUID, error) {
-	idFile := filepath.Join(osdDataPath, "whoami")
+func loadOSDInfo(config *osdConfig) error {
+	idFile := filepath.Join(config.rootPath, "whoami")
 	idContent, err := ioutil.ReadFile(idFile)
 	if err != nil {
-		return -1, uuid.UUID{}, fmt.Errorf("failed to read OSD ID from %s: %+v", idFile, err)
+		return fmt.Errorf("failed to read OSD ID from %s: %+v", idFile, err)
 	}
 
 	osdID, err := strconv.Atoi(strings.TrimSpace(string(idContent[:])))
 	if err != nil {
-		return -1, uuid.UUID{}, fmt.Errorf("failed to parse OSD ID from %s with content %s: %+v", idFile, idContent, err)
+		return fmt.Errorf("failed to parse OSD ID from %s with content %s: %+v", idFile, idContent, err)
 	}
 
-	uuidFile := filepath.Join(osdDataPath, "fsid")
+	uuidFile := filepath.Join(config.rootPath, "fsid")
 	fsidContent, err := ioutil.ReadFile(uuidFile)
 	if err != nil {
-		return -1, uuid.UUID{}, fmt.Errorf("failed to read UUID from %s: %+v", uuidFile, err)
+		return fmt.Errorf("failed to read UUID from %s: %+v", uuidFile, err)
 	}
 
 	osdUUID, err := uuid.Parse(strings.TrimSpace(string(fsidContent[:])))
 	if err != nil {
-		return -1, uuid.UUID{},
-			fmt.Errorf("failed to parse UUID from %s with content %s: %+v", uuidFile, string(fsidContent[:]), err)
+		return fmt.Errorf("failed to parse UUID from %s with content %s: %+v", uuidFile, string(fsidContent[:]), err)
 	}
 
-	return osdID, osdUUID, nil
+	config.id = osdID
+	config.uuid = osdUUID
+	return nil
 }
 
-func initializeOSD(factory client.ConnectionFactory, context *clusterd.Context, osdDataRoot string, osdID int, osdUUID uuid.UUID,
-	device string, bootstrapConn client.Connection, cluster *ClusterInfo, location string) (string, error) {
+func initializeOSD(config *osdConfig, factory client.ConnectionFactory, context *clusterd.Context,
+	bootstrapConn client.Connection, cluster *ClusterInfo, location string, executor exec.Executor) error {
 
-	// ensure that the OSD data directory is created
-	osdDataPath := filepath.Join(osdDataRoot, fmt.Sprintf("%s-%d", cluster.Name, osdID))
-	if err := os.MkdirAll(osdDataPath, 0777); err != nil {
-		return "", fmt.Errorf("failed to create OSD data dir at %s, %+v", osdDataPath, err)
-	}
-
-	cephConfig := createDefaultCephConfig(cluster, osdDataPath)
-	if device == localDeviceName {
+	cephConfig := createDefaultCephConfig(cluster, config.rootPath, config.bluestore)
+	if !config.bluestore {
 		// using the local file system requires some config overrides
 		// http://docs.ceph.com/docs/jewel/rados/configuration/filesystem-recommendations/#not-recommended
 		cephConfig.cephGlobalConfig.OsdMaxObjectNameLen = 256
 		cephConfig.cephGlobalConfig.OsdMaxObjectNamespaceLen = 64
 	}
 
-	// write the OSD config file to disk
-	keyringPath := getOSDKeyringPath(osdDataPath)
-	_, err := generateConfigFile(cluster, osdDataPath, fmt.Sprintf("osd.%d", osdID), keyringPath, cephConfig)
+	// bluestore has some extra settings
+	settings, err := getStoreSettings(context, config)
 	if err != nil {
-		return "", fmt.Errorf("failed to write OSD %d config file: %+v", osdID, err)
+		return fmt.Errorf("failed to read store settings. %+v", err)
+	}
+
+	// write the OSD config file to disk
+	keyringPath := getOSDKeyringPath(config.rootPath)
+	_, err = generateConfigFile(context, cluster, config.rootPath, fmt.Sprintf("osd.%d", config.id), keyringPath, config.bluestore, cephConfig, settings)
+	if err != nil {
+		return fmt.Errorf("failed to write OSD %d config file: %+v", config.id, err)
 	}
 
 	// get the current monmap, it will be needed for creating the OSD file system
 	monMapRaw, err := getMonMap(bootstrapConn)
 	if err != nil {
-		return "", fmt.Errorf("failed to get mon map: %+v", err)
+		return fmt.Errorf("failed to get mon map: %+v", err)
 	}
 
 	// create/initalize the OSD file system and journal
-	if err := createOSDFileSystem(context, cluster.Name, osdID, osdUUID, osdDataPath, monMapRaw); err != nil {
-		return "", err
+	if err := createOSDFileSystem(context, cluster.Name, config, monMapRaw); err != nil {
+		return err
 	}
 
 	// add auth privileges for the OSD, the bootstrap-osd privileges were very limited
-	if err := addOSDAuth(bootstrapConn, osdID, osdDataPath); err != nil {
-		return "", err
+	if err := addOSDAuth(bootstrapConn, config.id, config.rootPath); err != nil {
+		return err
 	}
 
 	// open a connection to the cluster using the OSDs creds
-	osdConn, err := connectToCluster(factory, cluster, osdDataRoot, fmt.Sprintf("osd.%d", osdID), keyringPath)
+	osdConn, err := connectToCluster(context, factory, cluster, path.Join(config.rootPath, "tmp"), fmt.Sprintf("osd.%d", config.id), keyringPath)
 	if err != nil {
-		return "", err
+		return err
 	}
 	defer osdConn.Shutdown()
 
 	// add the new OSD to the cluster crush map
-	if err := addOSDToCrushMap(osdConn, context, osdID, osdDataRoot, location); err != nil {
-		return "", err
+	if err := addOSDToCrushMap(osdConn, context, config.id, config.rootPath, location); err != nil {
+		return err
 	}
 
-	return osdDataPath, nil
+	return nil
+}
+
+func getStoreSettings(context *clusterd.Context, config *osdConfig) (map[string]string, error) {
+	settings := map[string]string{}
+	if !config.bluestore {
+		return settings, nil
+	}
+
+	scheme, err := partition.LoadScheme(config.rootPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load partition config. %+v", err)
+	}
+
+	walPartition, ok := scheme.PartitionUUIDs[partition.WalPartitionName]
+	if !ok {
+		return nil, fmt.Errorf("failed to find wal partition for osd %d", config.id)
+	}
+	dbPartition, ok := scheme.PartitionUUIDs[partition.DatabasePartitionName]
+	if !ok {
+		return nil, fmt.Errorf("failed to find db partition for osd %d", config.id)
+	}
+	blockPartition, ok := scheme.PartitionUUIDs[partition.BlockPartitionName]
+	if !ok {
+		return nil, fmt.Errorf("failed to find block partition for osd %d", config.id)
+	}
+
+	prefix := "/dev/disk/by-partuuid"
+	settings["bluestore block wal path"] = path.Join(prefix, walPartition)
+	settings["bluestore block db path"] = path.Join(prefix, dbPartition)
+	settings["bluestore block path"] = path.Join(prefix, blockPartition)
+
+	return settings, nil
 }
 
 // creates the OSD identity in the cluster via a mon_command
@@ -404,11 +482,11 @@ func getMonMap(bootstrapConn client.Connection) ([]byte, error) {
 }
 
 // creates/initalizes the OSD filesystem and journal via a child process
-func createOSDFileSystem(context *clusterd.Context, clusterName string, osdID int, osdUUID uuid.UUID, osdDataPath string, monMap []byte) error {
-	log.Printf("Initializing OSD %d file system at %s...", osdID, osdDataPath)
+func createOSDFileSystem(context *clusterd.Context, clusterName string, config *osdConfig, monMap []byte) error {
+	log.Printf("Initializing OSD %d file system at %s...", config.id, config.rootPath)
 
 	// the current monmap is needed to create the OSD, save it to a temp location so it is accessible
-	monMapTmpPath := getOSDTempMonMapPath(osdDataPath)
+	monMapTmpPath := getOSDTempMonMapPath(config.rootPath)
 	monMapTmpDir := filepath.Dir(monMapTmpPath)
 	if err := os.MkdirAll(monMapTmpDir, 0744); err != nil {
 		return fmt.Errorf("failed to create monmap tmp file directory at %s: %+v", monMapTmpDir, err)
@@ -417,23 +495,30 @@ func createOSDFileSystem(context *clusterd.Context, clusterName string, osdID in
 		return fmt.Errorf("failed to write mon map to tmp file %s, %+v", monMapTmpPath, err)
 	}
 
+	options := []string{
+		"--mkfs",
+		"--mkkey",
+		fmt.Sprintf("--id=%d", config.id),
+		fmt.Sprintf("--cluster=%s", clusterName),
+		fmt.Sprintf("--conf=%s", getConfFilePath(config.rootPath, clusterName)),
+		fmt.Sprintf("--osd-data=%s", config.rootPath),
+		fmt.Sprintf("--osd-uuid=%s", config.uuid.String()),
+		fmt.Sprintf("--monmap=%s", monMapTmpPath),
+	}
+
+	if !config.bluestore {
+		options = append(options, fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(config.rootPath)))
+		options = append(options, fmt.Sprintf("--keyring=%s", getOSDKeyringPath(config.rootPath)))
+	}
+
 	// create the OSD file system and journal
 	err := context.ProcMan.Run(
 		"osd",
-		"--mkfs",
-		"--mkkey",
-		fmt.Sprintf("--id=%s", strconv.Itoa(osdID)),
-		fmt.Sprintf("--cluster=%s", clusterName),
-		fmt.Sprintf("--osd-data=%s", osdDataPath),
-		fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(osdDataPath)),
-		fmt.Sprintf("--conf=%s", getOSDConfFilePath(osdDataPath, clusterName)),
-		fmt.Sprintf("--keyring=%s", getOSDKeyringPath(osdDataPath)),
-		fmt.Sprintf("--osd-uuid=%s", osdUUID.String()),
-		fmt.Sprintf("--monmap=%s", monMapTmpPath))
+		options...)
 
 	if err != nil {
 		return fmt.Errorf("failed osd mkfs for OSD ID %d, UUID %s, dataDir %s: %+v",
-			osdID, osdUUID.String(), osdDataPath, err)
+			config.id, config.uuid.String(), config.rootPath, err)
 	}
 
 	return nil
@@ -479,10 +564,10 @@ func addOSDToCrushMap(osdConn client.Connection, context *clusterd.Context, osdI
 
 	// weight is ratio of (size in KB) / (1 GB)
 	weight := float64(all/1024) / 1073741824.0
-	weight, _ = strconv.ParseFloat(fmt.Sprintf("%.2f", weight), 64)
+	weight, _ = strconv.ParseFloat(fmt.Sprintf("%.4f", weight), 64)
 
 	osdEntity := fmt.Sprintf("osd.%d", osdID)
-	log.Printf("OSD %s at %s, bytes: %d, weight: %.2f", osdEntity, osdDataPath, all, weight)
+	log.Printf("OSD %s at %s, bytes: %d, weight: %.4f", osdEntity, osdDataPath, all, weight)
 
 	locArgs, err := formatLocation(location)
 	if err != nil {

pkg/cephmgr/osd_test.go

@@ -19,10 +19,10 @@ import (
 
 func TestGetOSDInfo(t *testing.T) {
 	// error when no info is found on disk
-	id, guid, err := getOSDInfo("/tmp")
-	assert.Equal(t, -1, id)
+	config := &osdConfig{rootPath: "/tmp"}
+
+	err := loadOSDInfo(config)
 	assert.NotNil(t, err)
-	assert.Equal(t, uuid.UUID{}, guid)
 
 	// write the info to disk
 	whoFile := "/tmp/whoami"
@@ -34,37 +34,53 @@ func TestGetOSDInfo(t *testing.T) {
 	defer os.Remove(fsidFile)
 
 	// check the successful osd info
-	id, guid, err = getOSDInfo("/tmp")
+	err = loadOSDInfo(config)
 	assert.Nil(t, err)
-	assert.Equal(t, 23, id)
-	assert.Equal(t, testUUID, guid)
+	assert.Equal(t, 23, config.id)
+	assert.Equal(t, testUUID, config.uuid)
 }
 
 func TestDesiredDeviceState(t *testing.T) {
 	etcdClient := util.NewMockEtcdClient()
+	etcdClient.SetValue("/castle/nodes/config/a/disks/serialno/name", "foo")
 
 	// add a device
-	device := &Device{Name: "foo", NodeID: "a"}
-	err := AddDesiredDevice(etcdClient, device)
+	err := AddDesiredDevice(etcdClient, "foo", "a")
 	assert.Nil(t, err)
 	devices := etcdClient.GetChildDirs("/castle/services/ceph/osd/desired/a/device")
 	assert.Equal(t, 1, devices.Count())
-	assert.True(t, devices.Contains("foo"))
+	assert.True(t, devices.Contains("serialno"))
 
 	// remove the device
-	err = RemoveDesiredDevice(etcdClient, device)
+	err = RemoveDesiredDevice(etcdClient, "foo", "a")
 	assert.Nil(t, err)
 	devices = etcdClient.GetChildDirs("/castle/services/ceph/osd/desired/a/device")
 	assert.Equal(t, 0, devices.Count())
 
 	// removing a non-existent device is a no-op
-	err = RemoveDesiredDevice(etcdClient, device)
+	err = RemoveDesiredDevice(etcdClient, "foo", "a")
 	assert.Nil(t, err)
 }
 
+func TestDesiredDirsState(t *testing.T) {
+	etcdClient := util.NewMockEtcdClient()
+
+	// add a dir
+	err := AddDesiredDir(etcdClient, "/my/dir", "a")
+	assert.Nil(t, err)
+	dirs := etcdClient.GetChildDirs("/castle/services/ceph/osd/desired/a/dir")
+	assert.Equal(t, 1, dirs.Count())
+	assert.True(t, dirs.Contains("my_dir"))
+	assert.Equal(t, "/my/dir", etcdClient.GetValue("/castle/services/ceph/osd/desired/a/dir/my_dir/path"))
+
+	loadedDirs, err := loadDesiredDirs(etcdClient, "a")
+	assert.Nil(t, err)
+	assert.True(t, loadedDirs.Equals(util.CreateSet([]string{"/my/dir"})), fmt.Sprintf("dirs=%v", loadedDirs))
+}
+
 func TestOSDBootstrap(t *testing.T) {
 	clusterName := "mycluster"
-	targetPath := getBootstrapOSDKeyringPath(clusterName)
+	targetPath := getBootstrapOSDKeyringPath("/tmp", clusterName)
 	defer os.Remove(targetPath)
 
 	factory := &testceph.MockConnectionFactory{}
@@ -75,7 +91,7 @@ func TestOSDBootstrap(t *testing.T) {
 		return []byte(response), "", nil
 	}
 
-	err := createOSDBootstrapKeyring(conn, clusterName)
+	err := createOSDBootstrapKeyring(conn, "/tmp", clusterName)
 	assert.Nil(t, err)
 
 	contents, err := ioutil.ReadFile(targetPath)

pkg/cephmgr/osdagent.go

@@ -16,16 +16,14 @@ import (
 
 	"github.com/quantum/castle/pkg/cephmgr/client"
 	"github.com/quantum/castle/pkg/clusterd"
-	"github.com/quantum/castle/pkg/clusterd/inventory"
 	"github.com/quantum/castle/pkg/util"
 	"github.com/quantum/castle/pkg/util/proc"
-	"github.com/quantum/castle/pkg/util/sys"
 )
 
 const (
-	osdAgentName    = "osd"
-	devicesKey      = "devices"
-	localDeviceName = "<local>"
+	osdAgentName = "osd"
+	deviceKey    = "device"
+	dirKey       = "dir"
 )
 
 type osdAgent struct {
@@ -37,6 +35,20 @@ type osdAgent struct {
 	devices     string
 }
 
+type osdInfo struct {
+	id     int
+	serial string
+	dir    bool
+}
+
+type osdConfig struct {
+	name      string
+	rootPath  string
+	id        int
+	uuid      uuid.UUID
+	bluestore bool
+}
+
 func newOSDAgent(factory client.ConnectionFactory, devices string, forceFormat bool, location string) *osdAgent {
 	a := &osdAgent{factory: factory, devices: devices, forceFormat: forceFormat, location: location}
 	return a
@@ -49,18 +61,24 @@ func (a *osdAgent) Name() string {
 // set the desired state in etcd
 func (a *osdAgent) Initialize(context *clusterd.Context) error {
 
-	// add the devices to desired state
-	devices := strings.Split(a.devices, ",")
-
-	if len(devices) == 1 && devices[0] == "" {
-		// no devices specified, just use the local file system
-		devices = []string{localDeviceName}
+	if len(a.devices) > 0 {
+		// add the devices to desired state
+		devices := strings.Split(a.devices, ",")
+		for _, device := range devices {
+			log.Printf("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)
+			}
+		}
 	}
 
-	for _, device := range devices {
-		err := AddDesiredDevice(context.EtcdClient, &Device{Name: device, NodeID: context.NodeID})
+	// if no devices or directories were specified, use the current directory for an osd
+	if len(a.devices) == 0 {
+		log.Printf("Adding local path to local directory %s", context.ConfigDir)
+		err := AddDesiredDir(context.EtcdClient, context.ConfigDir, context.NodeID)
 		if err != nil {
-			return fmt.Errorf("failed to add desired device %s", device)
+			return fmt.Errorf("failed to add current dir %s. %v", context.ConfigDir, err)
 		}
 	}
 
@@ -93,28 +111,33 @@ func (a *osdAgent) ConfigureLocalService(context *clusterd.Context) error {
 	}
 
 	// Connect to the ceph cluster
-	adminConn, err := ConnectToClusterAsAdmin(a.factory, a.cluster)
+	adminConn, err := ConnectToClusterAsAdmin(context, a.factory, a.cluster)
 	if err != nil {
 		return err
 	}
 	defer adminConn.Shutdown()
 
 	devices, err := loadDesiredDevices(context.EtcdClient, context.NodeID)
-	if devices.Count() == 0 {
-		return nil
+	if err != nil {
+		return fmt.Errorf("failed to load desired devices. %v", err)
 	}
 
-	if err := a.startDesiredDevices(context, adminConn, devices); err != nil {
+	dirs, err := loadDesiredDirs(context.EtcdClient, context.NodeID)
+	if err != nil {
+		return fmt.Errorf("failed to load desired dirs. %v", err)
+	}
+
+	if err := a.startDesiredDevices(context, adminConn, devices, dirs); err != nil {
 		return err
 	}
 
 	return a.stopUndesiredDevices(context, adminConn, devices)
 }
 
-func (a *osdAgent) startDesiredDevices(context *clusterd.Context, connection client.Connection, devices *util.Set) error {
+func (a *osdAgent) startDesiredDevices(context *clusterd.Context, connection client.Connection, devices map[string]string, dirs *util.Set) error {
 
 	// Start an OSD for each of the specified devices
-	deviceMap, err := a.createOSDs(connection, context, devices)
+	deviceMap, err := a.createOSDs(connection, context, devices, dirs)
 	if err != nil {
 		return fmt.Errorf("failed to create OSDs: %+v", err)
 	}
@@ -127,20 +150,25 @@ func (a *osdAgent) startDesiredDevices(context *clusterd.Context, connection cli
 	return nil
 }
 
-func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection client.Connection, desired *util.Set) error {
+func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection client.Connection, desiredDevices map[string]string) error {
 	applied, err := GetAppliedOSDs(context.NodeID, context.EtcdClient)
 	if err != nil {
 		return fmt.Errorf("failed to get applied OSDs. %v", err)
 	}
 
+	desiredSerial := util.NewSet()
+	for _, serial := range desiredDevices {
+		desiredSerial.Add(serial)
+	}
+
 	var lastErr error
-	for device := range applied {
-		if desired.Contains(device) {
+	for appliedSerial := range applied.Iter() {
+		if desiredSerial.Contains(appliedSerial) {
 			// the osd is desired and applied
 			continue
 		}
 
-		err := a.removeOSD(context, connection, device)
+		err := a.removeOSD(context, connection, appliedSerial)
 		if err != nil {
 			lastErr = err
 		}
@@ -149,23 +177,23 @@ func (a *osdAgent) stopUndesiredDevices(context *clusterd.Context, connection cl
 	return lastErr
 }
 
-func getIDFromName(context *clusterd.Context, name string) (int, error) {
-	key := path.Join(getAppliedKey(context.NodeID), devicesKey, name, "id")
+func getIDFromSerial(context *clusterd.Context, serial string) (int, error) {
+	key := path.Join(getAppliedKey(context.NodeID), deviceKey, serial, "id")
 	val, err := context.EtcdClient.Get(ctx.Background(), key, nil)
 	if err != nil {
-		return -1, fmt.Errorf("failed to get device id from name %s. %v", name, err)
+		return -1, fmt.Errorf("failed to get device id from serial %s. %v", serial, err)
 	}
 
 	id, err := strconv.ParseInt(val.Node.Value, 10, 32)
 	if err != nil {
-		return -1, fmt.Errorf("failed to get parse id from name %s. %v", name, err)
+		return -1, fmt.Errorf("failed to get parse id from serial %s. %v", serial, err)
 	}
 
 	return int(id), nil
 }
 
 func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connection, name string) error {
-	osdID, err := getIDFromName(context, name)
+	osdID, err := getIDFromSerial(context, name)
 	if err != nil {
 		return fmt.Errorf("failed to get osd %s id. %v", name, err)
 	}
@@ -194,7 +222,7 @@ func (a *osdAgent) removeOSD(context *clusterd.Context, connection client.Connec
 	}
 
 	// remove the osd from the applied key
-	appliedKey := path.Join(getAppliedKey(context.NodeID), devicesKey, name)
+	appliedKey := path.Join(getAppliedKey(context.NodeID), deviceKey, name)
 	_, err = context.EtcdClient.Delete(ctx.Background(), appliedKey, &etcd.DeleteOptions{Recursive: true, Dir: true})
 	if err != nil {
 		log.Printf("failed to remove device %s from desired state. %v", name, err)
@@ -223,156 +251,161 @@ func getAppliedKey(nodeID string) string {
 }
 
 // create and initalize OSDs for all the devices specified in the given config
-func (a *osdAgent) createOSDs(adminConn client.Connection, context *clusterd.Context, devices *util.Set) (map[string]int, error) {
+func (a *osdAgent) createOSDs(adminConn client.Connection, context *clusterd.Context, devices map[string]string, dirs *util.Set) (map[string]*osdInfo, error) {
 	// generate and write the OSD bootstrap keyring
-	if err := createOSDBootstrapKeyring(adminConn, a.cluster.Name); err != nil {
+	if err := createOSDBootstrapKeyring(adminConn, context.ConfigDir, a.cluster.Name); err != nil {
 		return nil, err
 	}
 
 	// connect to the cluster using the bootstrap-osd creds, this connection will be used for config operations
-	bootstrapConn, err := connectToCluster(a.factory, a.cluster, getBootstrapOSDDir(), "bootstrap-osd", getBootstrapOSDKeyringPath(a.cluster.Name))
+	bootstrapConn, err := connectToCluster(context, a.factory, a.cluster, getBootstrapOSDDir(context.ConfigDir), "bootstrap-osd", getBootstrapOSDKeyringPath(context.ConfigDir, a.cluster.Name))
 	if err != nil {
 		return nil, err
 	}
 	defer bootstrapConn.Shutdown()
 
+	deviceMap := map[string]*osdInfo{}
+	for dir := range dirs.Iter() {
+		config, err := a.createLocalOSD(bootstrapConn, dir)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create local OSD: %+v", err)
+		}
+
+		err = a.createOrStartOSD(context, bootstrapConn, config)
+		if err != nil {
+			return nil, fmt.Errorf("failed to start osd on device %s. %v", dir, err)
+		}
+
+		deviceMap[dir] = &osdInfo{id: config.id, dir: true}
+	}
+
 	// initialize all the desired OSD volumes
-	deviceMap := map[string]int{}
-	for device := range devices.Iter() {
-		var osdID int
-		var osdUUID uuid.UUID
-		var osdDataRoot string
-
-		if device == localDeviceName {
-			osdDataRoot, osdID, osdUUID, err = a.createLocalOSD(bootstrapConn)
-			if err != nil {
-				return nil, fmt.Errorf("failed to create local OSD: %+v", err)
-			}
-		} else {
-			osdDataRoot, osdID, osdUUID, err = a.createMountedOSD(device, bootstrapConn, context)
-			if err != nil {
-				return nil, fmt.Errorf("failed to create mounted OSD for device %s: %+v", device, err)
-			}
-		}
-
-		// find the OSD data path (under the mount point/data dir)
-		osdDataPath, err := findOSDDataPath(osdDataRoot, a.cluster.Name)
+	for device, serial := range devices {
+		config, err := a.prepareDeviceForOSD(device, serial, bootstrapConn, context)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("failed to create mounted OSD on %s: %+v", device, err)
 		}
 
-		if isOSDDataNotExist(osdDataPath) {
-			// osd_data_dir/whoami does not exist yet, create/initialize the OSD
-			osdDataPath, err = initializeOSD(a.factory, context, osdDataRoot, osdID, osdUUID, device, bootstrapConn, a.cluster, a.location)
-			if err != nil {
-				return nil, fmt.Errorf("failed to initialize OSD at %s: %+v", osdDataRoot, err)
-			}
-		} else {
-			// osd_data_dir/whoami already exists, meaning the OSD is already set up.
-			// look up some basic information about it so we can run it.
-			osdID, osdUUID, err = getOSDInfo(osdDataPath)
-			if err != nil {
-				return nil, fmt.Errorf("failed to get OSD information from %s: %+v", osdDataPath, err)
-			}
-		}
-
-		// run the OSD in a child process now that it is fully initialized and ready to go
-		err = a.runOSD(context, a.cluster.Name, osdID, osdUUID, osdDataPath, device)
+		err = a.createOrStartOSD(context, bootstrapConn, config)
 		if err != nil {
-			return nil, fmt.Errorf("failed to run osd %d: %+v", osdID, err)
+			return nil, fmt.Errorf("failed to start osd on device %s. %v", device, err)
 		}
 
-		deviceMap[device] = osdID
+		deviceMap[device] = &osdInfo{id: config.id, serial: serial}
 	}
 
 	return deviceMap, nil
 }
 
-func (a *osdAgent) createMountedOSD(device string, bootstrapConn client.Connection, context *clusterd.Context) (string, int, uuid.UUID, error) {
+func (a *osdAgent) createOrStartOSD(context *clusterd.Context, connection client.Connection, config *osdConfig) error {
+
+	if isOSDDataNotExist(config.rootPath) {
+
+		// osd_data_dir/whoami does not exist yet, create/initialize the OSD
+		err := initializeOSD(config, a.factory, context, connection, a.cluster, a.location, context.Executor)
+		if err != nil {
+			return fmt.Errorf("failed to initialize OSD at %s: %+v", config.rootPath, err)
+		}
+
+	} else {
+		// osd_data_dir/whoami already exists, meaning the OSD is already set up.
+		// look up some basic information about it so we can run it.
+		err := loadOSDInfo(config)
+		if err != nil {
+			return fmt.Errorf("failed to get OSD information from %s: %+v", config.rootPath, err)
+		}
+	}
+
+	// run the OSD in a child process now that it is fully initialized and ready to go
+	err := a.runOSD(context, a.cluster.Name, config)
+	if err != nil {
+		return fmt.Errorf("failed to run osd %d: %+v", config.id, err)
+	}
+
+	return nil
+}
+
+func (a *osdAgent) prepareDeviceForOSD(device, serial string, bootstrapConn client.Connection, context *clusterd.Context) (*osdConfig, error) {
 	var osdID int
 	var osdUUID uuid.UUID
 
-	mountPoint, err := sys.GetDeviceMountPoint(device, context.Executor)
-	if err != nil {
-		return "", -1, uuid.UUID{}, fmt.Errorf("unable to get mount point for device %s: %+v", device, err)
-	}
-
-	if mountPoint == "" {
-		// the device is not mounted, so proceed with the format and mounting
-		if err := formatOSD(device, a.forceFormat, context.Executor); err != nil {
+	configDir := getOSDConfigDir(context.ConfigDir, serial)
+	if isOSDDataNotExist(configDir) {
+		// the device needs to be formatted
+		err := formatDevice(context, device, serial, configDir, a.forceFormat)
+		if err != nil {
 			// attempting to format the volume failed, bail out with error
-			return "", -1, uuid.UUID{}, err
+			return nil, err
 		}
 
 		// register the OSD with the cluster now to get an official ID
 		osdID, osdUUID, err = registerOSDWithCluster(device, bootstrapConn)
 		if err != nil {
-			return "", -1, uuid.UUID{}, fmt.Errorf("failed to regiser OSD %d with cluster: %+v", osdID, err)
+			return nil, fmt.Errorf("failed to regiser OSD %d with cluster: %+v", osdID, err)
 		}
 
-		// mount the device using a mount point that reflects the OSD's ID
-		mountPoint = getOSDRootDir(osdID)
-		if err := mountOSD(device, mountPoint, context.Executor); err != nil {
-			return "", -1, uuid.UUID{}, err
+		if err := os.MkdirAll(configDir, 0744); err != nil {
+			return nil, fmt.Errorf("failed to create config dir at %s: %+v", configDir, err)
 		}
 	}
 
-	return mountPoint, osdID, osdUUID, nil
+	// configure devices with bluestore, and local directories with filestore
+	return &osdConfig{rootPath: configDir, id: osdID, uuid: osdUUID, name: device, bluestore: true}, nil
 }
 
-func (a *osdAgent) createLocalOSD(bootstrapConn client.Connection) (string, int, uuid.UUID, error) {
+func (a *osdAgent) createLocalOSD(connection client.Connection, root string) (*osdConfig, error) {
 	var osdID int
 	var osdUUID uuid.UUID
 	var osdDataPath string
 
-	root := "/tmp"
 	osdDataRoot, err := findOSDDataRoot(root)
 	if err != nil {
-		return "", -1, uuid.UUID{}, fmt.Errorf("failed to find OSD data root under %s: %+v", root, err)
-	}
-
-	if osdDataRoot != "" {
-		osdDataPath, err = findOSDDataPath(osdDataRoot, a.cluster.Name)
-		if err != nil {
-			return "", -1, uuid.UUID{}, fmt.Errorf("failed to find osd data path under %s: %+v", osdDataRoot, err)
-		}
+		return nil, fmt.Errorf("failed to find OSD data root under %s: %+v", root, err)
 	}
 
 	if isOSDDataNotExist(osdDataPath) {
 		// register the OSD with the cluster now to get an official ID
-		osdID, osdUUID, err = registerOSDWithCluster(osdDataRoot, bootstrapConn)
+		osdID, osdUUID, err = registerOSDWithCluster(root, connection)
 		if err != nil {
-			return "", -1, uuid.UUID{}, fmt.Errorf("failed to register OSD %d with cluster: %+v", osdID, err)
+			return nil, fmt.Errorf("failed to register OSD %d with cluster: %+v", osdID, err)
 		}
 
-		osdDataRoot = getOSDRootDir(osdID)
+		osdDataRoot = getOSDRootDir(root, osdID)
 		if err := os.MkdirAll(osdDataRoot, 0744); err != nil {
-			return "", -1, uuid.UUID{}, fmt.Errorf("failed to make osdDataRoot at %s: %+v", osdDataRoot, err)
+			return nil, fmt.Errorf("failed to make osdDataRoot at %s: %+v", osdDataRoot, err)
 		}
 	}
 
-	return osdDataRoot, osdID, osdUUID, nil
+	return &osdConfig{rootPath: osdDataRoot, id: osdID, uuid: osdUUID, bluestore: false}, nil
 }
 
 // runs an OSD with the given config in a child process
-func (a *osdAgent) runOSD(context *clusterd.Context, clusterName string, osdID int, osdUUID uuid.UUID, osdDataPath, device string) error {
+func (a *osdAgent) runOSD(context *clusterd.Context, clusterName string, config *osdConfig) error {
 	// start the OSD daemon in the foreground with the given config
-	log.Printf("starting osd %d at %s", osdID, osdDataPath)
-	osdUUIDArg := fmt.Sprintf("--osd-uuid=%s", osdUUID.String())
+	log.Printf("starting osd %d at %s", config.id, config.rootPath)
+
+	osdUUIDArg := fmt.Sprintf("--osd-uuid=%s", config.uuid.String())
+
+	params := []string{"--foreground",
+		fmt.Sprintf("--id=%d", config.id),
+		fmt.Sprintf("--cluster=%s", clusterName),
+		fmt.Sprintf("--osd-data=%s", config.rootPath),
+		fmt.Sprintf("--conf=%s", getOSDConfFilePath(config.rootPath, clusterName)),
+		fmt.Sprintf("--keyring=%s", getOSDKeyringPath(config.rootPath)),
+		osdUUIDArg,
+	}
+
+	if !config.bluestore {
+		params = append(params, fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(config.rootPath)))
+	}
+
 	process, err := context.ProcMan.Start(
 		"osd",
 		regexp.QuoteMeta(osdUUIDArg),
 		proc.ReuseExisting,
-		"--foreground",
-		fmt.Sprintf("--id=%s", strconv.Itoa(osdID)),
-		fmt.Sprintf("--cluster=%s", clusterName),
-		fmt.Sprintf("--osd-data=%s", osdDataPath),
-		fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(osdDataPath)),
-		fmt.Sprintf("--conf=%s", getOSDConfFilePath(osdDataPath, clusterName)),
-		fmt.Sprintf("--keyring=%s", getOSDKeyringPath(osdDataPath)),
-		osdUUIDArg)
+		params...)
 	if err != nil {
-		return fmt.Errorf("failed to start osd %d: %+v", osdID, err)
+		return fmt.Errorf("failed to start osd %d: %+v", config.id, err)
 	}
 
 	if a.osdProc == nil {
@@ -381,60 +414,59 @@ func (a *osdAgent) runOSD(context *clusterd.Context, clusterName string, osdID i
 	}
 	if process != nil {
 		// if the process was already running Start will return nil in which case we don't want to overwrite it
-		a.osdProc[device] = process
+		a.osdProc[config.name] = process
 	}
 
 	return nil
 }
 
 // For all applied OSDs, gets a mapping of their device names to their serial numbers
-func GetAppliedOSDs(nodeID string, etcdClient etcd.KeysAPI) (map[string]string, error) {
-	appliedKey := path.Join(getAppliedKey(nodeID), devicesKey)
-	devices := map[string]string{}
+func GetAppliedOSDs(nodeID string, etcdClient etcd.KeysAPI) (*util.Set, error) {
+	appliedKey := path.Join(getAppliedKey(nodeID), deviceKey)
 
-	deviceKeys, err := etcdClient.Get(ctx.Background(), appliedKey, &etcd.GetOptions{Recursive: true})
+	devices, err := util.GetDirChildKeys(etcdClient, appliedKey)
 	if err != nil {
 		if util.IsEtcdKeyNotFound(err) {
-			return devices, nil
+			return util.NewSet(), nil
 		}
 		return nil, err
 	}
 
-	// parse the device info from etcd
-	for _, dev := range deviceKeys.Node.Nodes {
-		name := util.GetLeafKeyPath(dev.Key)
-		for _, setting := range dev.Nodes {
-			if strings.HasSuffix(setting.Key, "/serial") {
-				devices[name] = setting.Value
-			}
-		}
-	}
-
 	return devices, nil
 }
 
-func (a *osdAgent) saveAppliedOSDs(context *clusterd.Context, devices map[string]int) error {
-	appliedKey := path.Join(getAppliedKey(context.NodeID), devicesKey)
+func (a *osdAgent) saveAppliedOSDs(context *clusterd.Context, devices map[string]*osdInfo) error {
 	var settings = make(map[string]string)
-	for name, id := range devices {
-		var serial string
-		var err error
-		if name == localDeviceName {
-			serial = localDeviceName
+	for name, info := range devices {
+		var pseudoName string
+		if info.serial == "" {
+			// convert the local path to an id that will not contain path characters in the etcd key
+			pseudoName = getPseudoDir(name)
 		} else {
-			serial, err = inventory.GetSerialFromDevice(name, context.NodeID, context.EtcdClient)
-			if err != nil {
-				return fmt.Errorf("failed to get serial for device %s: %+v", name, err)
-			}
+			pseudoName = info.serial
+		}
+		baseKey := deviceKey
+		if info.dir {
+			baseKey = dirKey
 		}
 
-		settings[path.Join(name, "serial")] = serial
-		settings[path.Join(name, "id")] = strconv.FormatInt(int64(id), 10)
+		settings[path.Join(baseKey, pseudoName, "name")] = name
+		settings[path.Join(baseKey, pseudoName, "id")] = strconv.FormatInt(int64(info.id), 10)
 	}
 
+	appliedKey := path.Join(getAppliedKey(context.NodeID))
 	if err := util.StoreEtcdProperties(context.EtcdClient, appliedKey, settings); err != nil {
 		return fmt.Errorf("failed to mark devices as applied: %+v", err)
 	}
 
 	return nil
 }
+
+func getPseudoDir(path string) string {
+	// cut off the leading slash so we won't end up with a hidden etcd key
+	if strings.HasPrefix(path, "/") {
+		path = path[1:]
+	}
+
+	return strings.Replace(path, "/", "_", -1)
+}

pkg/cephmgr/osdagent_test.go

@@ -7,7 +7,6 @@ import (
 	"os"
 	"os/exec"
 	"path"
-	"strings"
 	"testing"
 
 	testceph "github.com/quantum/castle/pkg/cephmgr/client/test"
@@ -19,13 +18,14 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+/*
 func TestOSDAgentWithDevices(t *testing.T) {
 	clusterName := "mycluster"
-	bootstrapPath := getBootstrapOSDKeyringPath(clusterName)
+	bootstrapPath := getBootstrapOSDKeyringPath("/tmp", clusterName)
 	defer os.Remove(bootstrapPath)
 	defer os.RemoveAll("/tmp/osd3")
 
-	etcdClient, agent, _ := createTestAgent("sdx,sdy")
+	etcdClient, agent, _ := createTestAgent(t, "sdx,sdy")
 
 	execCount := 0
 	executor := &exectest.MockExecutor{}
@@ -35,36 +35,27 @@ func TestOSDAgentWithDevices(t *testing.T) {
 		nameSuffix := parts[len(parts)-1]
 		switch {
 		case execCount == 0:
+		case (execCount % 9) < 5:
+			assert.Equal(t, "parted", command)
+			assert.Equal(t, "-s", args[0])
+			assert.Equal(t, "/dev/"+nameSuffix, args[1])
+
+		case (execCount % 9) == 5:
 			assert.Equal(t, "format "+nameSuffix, name)
 			assert.Equal(t, "/usr/sbin/mkfs.btrfs", args[0])
-			assert.Equal(t, "/dev/"+nameSuffix, args[6])
-		case execCount == 1:
+			assert.True(t, strings.HasPrefix(args[6], "/dev/"+nameSuffix), args[6])
+		case (execCount % 9) == 6:
 			assert.Equal(t, "mount "+nameSuffix, name)
-			assert.Equal(t, "sudo", command)
-			assert.Equal(t, "mount", args[0])
-			assert.Equal(t, "user_subvol_rm_allowed", args[2])
-			assert.Equal(t, "/tmp/osd3", args[4])
-		case execCount == 2:
-			assert.Equal(t, "chown /tmp/osd3", name)
-			assert.Equal(t, "sudo", command)
-			assert.Equal(t, "chown", args[0])
+			assert.Equal(t, "mount", command)
+			assert.Equal(t, "user_subvol_rm_allowed", args[1])
 			assert.Equal(t, "/tmp/osd3", args[3])
-		case execCount == 3:
-			assert.Equal(t, "format "+nameSuffix, name)
-			assert.Equal(t, "sudo", command)
-			assert.Equal(t, "/usr/sbin/mkfs.btrfs", args[0])
-			assert.Equal(t, "/dev/"+nameSuffix, args[6])
-		case execCount == 4:
-			assert.Equal(t, "mount "+nameSuffix, name)
-			assert.Equal(t, "sudo", command)
-			assert.Equal(t, "mount", args[0])
-			assert.Equal(t, "user_subvol_rm_allowed", args[2])
-			assert.Equal(t, "/tmp/osd3", args[4])
-		case execCount == 5:
+		case (execCount % 9) == 7:
 			assert.Equal(t, "chown /tmp/osd3", name)
-			assert.Equal(t, "sudo", command)
-			assert.Equal(t, "chown", args[0])
-			assert.Equal(t, "/tmp/osd3", args[3])
+			assert.Equal(t, "chown", command)
+			assert.Equal(t, "/tmp/osd3", args[2])
+		case (execCount % 9) == 8:
+			assert.Equal(t, "chown "+nameSuffix, name)
+			assert.Equal(t, "chown", command)
 		default:
 			assert.Fail(t, fmt.Sprintf("unexpected case %d", execCount))
 		}
@@ -97,17 +88,17 @@ func TestOSDAgentWithDevices(t *testing.T) {
 		ProcMan:    &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 3: proc.StartAction, 4: proc.StopAction})},
 	}
 
-	// prep the OSD agent and related orcehstration data
-	prepAgentOrchestrationData(t, agent, etcdClient, context, clusterName)
-
 	// prep the etcd keys that would have been discovered by inventory
 	disksKey := path.Join(inventory.GetNodeConfigKey(context.NodeID), inventory.DisksKey)
 	etcdClient.SetValue(path.Join(disksKey, "sdxserial", "name"), "sdx")
 	etcdClient.SetValue(path.Join(disksKey, "sdyserial", "name"), "sdy")
 
+	// prep the OSD agent and related orcehstration data
+	prepAgentOrchestrationData(t, agent, etcdClient, context, clusterName)
+
 	err := agent.ConfigureLocalService(context)
 	assert.Nil(t, err)
-	assert.Equal(t, 6, execCount)
+	assert.Equal(t, 18, execCount)
 	assert.Equal(t, 2, outputExecCount)
 	assert.Equal(t, 4, procCommands)
 	assert.Equal(t, 2, len(agent.osdProc))
@@ -116,21 +107,22 @@ func TestOSDAgentWithDevices(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 0, len(agent.osdProc))
 }
+*/
 
 func TestOSDAgentNoDevices(t *testing.T) {
 	clusterName := "mycluster"
-	bootstrapPath := getBootstrapOSDKeyringPath(clusterName)
+	bootstrapPath := getBootstrapOSDKeyringPath("/tmp", clusterName)
+	os.MkdirAll("/tmp/osd3", 0744)
 	defer os.Remove(bootstrapPath)
 	defer os.RemoveAll("/tmp/osd3")
 
 	// create a test OSD agent with no devices specified
-	etcdClient, agent, _ := createTestAgent("")
+	etcdClient, agent, _ := createTestAgent(t, "")
 
 	// should be no executeCommand calls
 	execCount := 0
 	executor := &exectest.MockExecutor{}
 	executor.MockExecuteCommand = func(name string, command string, args ...string) error {
-		assert.Fail(t, "executeCommand is not expected for OSD local device")
 		execCount++
 		return nil
 	}
@@ -150,13 +142,17 @@ func TestOSDAgentNoDevices(t *testing.T) {
 		Executor:   executor,
 		NodeID:     "abc",
 		ProcMan:    &proc.ProcManager{Trap: createOSDAgentProcTrap(t, &procCommands, map[int]string{1: proc.StartAction, 2: proc.StopAction})},
+		ConfigDir:  "/tmp",
 	}
 
 	// prep the OSD agent and related orcehstration data
 	prepAgentOrchestrationData(t, agent, etcdClient, context, clusterName)
 
 	// configure the OSD and verify the results
-	err := agent.ConfigureLocalService(context)
+	err := agent.Initialize(context)
+	assert.Nil(t, err)
+
+	err = agent.ConfigureLocalService(context)
 	assert.Nil(t, err)
 	assert.Equal(t, 0, execCount)
 	assert.Equal(t, 0, outputExecCount)
@@ -166,8 +162,8 @@ func TestOSDAgentNoDevices(t *testing.T) {
 	// the local device should be marked as an applied OSD now
 	osds, err := GetAppliedOSDs(context.NodeID, etcdClient)
 	assert.Nil(t, err)
-	assert.Equal(t, 1, len(osds))
-	assert.Equal(t, localDeviceName, osds[localDeviceName])
+	assert.Equal(t, 0, osds.Count())
+	// TODO: applied osds should return the configured dirs
 
 	// destroy the OSD and verify the results
 	err = agent.DestroyLocalService(context)
@@ -182,7 +178,7 @@ func TestAppliedDevices(t *testing.T) {
 	// no applied osds
 	osds, err := GetAppliedOSDs(nodeID, etcdClient)
 	assert.Nil(t, err)
-	assert.Equal(t, 0, len(osds))
+	assert.Equal(t, 0, osds.Count())
 
 	// two applied osds
 	nodeConfigKey := path.Join(inventory.NodesConfigKey, nodeID)
@@ -191,55 +187,54 @@ func TestAppliedDevices(t *testing.T) {
 		100, true, false, "btrfs", "/mnt/xyz", inventory.Disk, "", false)
 	inventory.TestSetDiskInfo(etcdClient, nodeConfigKey, "serial2", "sdb", "ff6d4869-29ee-4bfd-bf21-dfd597bd222e",
 		50, false, false, "ext4", "/mnt/zyx", inventory.Disk, "", false)
-	appliedOSDKey := "/castle/services/ceph/osd/applied/abc/devices"
-	etcdClient.SetValue(path.Join(appliedOSDKey, "sda", "serial"), "serial1")
-	etcdClient.SetValue(path.Join(appliedOSDKey, "sdb", "serial"), "serial2")
+	appliedOSDKey := "/castle/services/ceph/osd/applied/abc/device"
+	etcdClient.SetValue(path.Join(appliedOSDKey, "serial1", "name"), "sda")
+	etcdClient.SetValue(path.Join(appliedOSDKey, "serial2", "name"), "sdb")
 
 	osds, err = GetAppliedOSDs(nodeID, etcdClient)
 	assert.Nil(t, err)
-	assert.Equal(t, 2, len(osds))
-	assert.Equal(t, "serial1", osds["sda"])
-	assert.Equal(t, "serial2", osds["sdb"])
+	assert.Equal(t, 2, osds.Count())
+	assert.True(t, osds.Equals(util.CreateSet([]string{"serial1", "serial2"})))
 }
 
 func TestRemoveDevice(t *testing.T) {
-	etcdClient, agent, conn := createTestAgent("")
+	etcdClient, agent, conn := createTestAgent(t, "")
 	executor := &exectest.MockExecutor{}
 	procTrap := func(action string, c *exec.Cmd) error {
 		return nil
 	}
 
 	context := &clusterd.Context{EtcdClient: etcdClient, NodeID: "a", Executor: executor, ProcMan: &proc.ProcManager{Trap: procTrap}}
-	desired := util.CreateSet([]string{"sda"})
+	desired := map[string]string{"sda": "123"}
 
-	// create two applied osds
-	root := "/castle/services/ceph/osd/applied/a/devices"
-	etcdClient.SetValue(path.Join(root, "sda/serial"), "123")
-	etcdClient.SetValue(path.Join(root, "sdb/serial"), "456")
-	etcdClient.SetValue(path.Join(root, "sdc/serial"), "789")
+	// create two applied osds, one of which is desired
+	appliedRoot := "/castle/services/ceph/osd/applied/a/device"
+	etcdClient.SetValue(path.Join(appliedRoot, "123", "name"), "sda")
+	etcdClient.SetValue(path.Join(appliedRoot, "456", "name"), "sdb")
 
-	// the request will fail without the device id set
+	// removing the device will fail without the id
 	err := agent.stopUndesiredDevices(context, conn, desired)
 	assert.NotNil(t, err)
 
-	etcdClient.SetValue(path.Join(root, "sda/id"), "1")
-	etcdClient.SetValue(path.Join(root, "sdb/id"), "2")
-	etcdClient.SetValue(path.Join(root, "sdc/id"), "3")
-
+	// set the id then successfully remove the device
+	etcdClient.SetValue(path.Join(appliedRoot, "456", "id"), "1")
 	err = agent.stopUndesiredDevices(context, conn, desired)
 	assert.Nil(t, err)
-	applied := etcdClient.GetChildDirs(root)
-	assert.True(t, applied.Equals(desired))
+
+	applied := etcdClient.GetChildDirs(appliedRoot)
+	assert.True(t, applied.Equals(util.CreateSet([]string{"123"})), fmt.Sprintf("applied=%+v", applied))
 }
 
-func createTestAgent(devices string) (*util.MockEtcdClient, *osdAgent, *testceph.MockConnection) {
+func createTestAgent(t *testing.T, devices string) (*util.MockEtcdClient, *osdAgent, *testceph.MockConnection) {
 	location := "root=here"
 	forceFormat := false
 	etcdClient := util.NewMockEtcdClient()
 	factory := &testceph.MockConnectionFactory{}
 	agent := newOSDAgent(factory, devices, forceFormat, location)
 	agent.cluster = &ClusterInfo{Name: "myclust"}
-	agent.Initialize(&clusterd.Context{EtcdClient: etcdClient})
+	agent.Initialize(&clusterd.Context{EtcdClient: etcdClient, ConfigDir: "/tmp"})
+	assert.Equal(t, "/tmp", etcdClient.GetValue("/castle/services/ceph/osd/desired/dir/tmp/path"))
+
 	conn, _ := factory.NewConnWithClusterAndUser("default", "user")
 	mockConn := conn.(*testceph.MockConnection)
 	mockConn.MockMonCommand = func(buf []byte) (buffer []byte, info string, err error) {
@@ -270,6 +265,7 @@ func prepAgentOrchestrationData(t *testing.T, agent *osdAgent, etcdClient *util.
 func createOSDAgentProcTrap(t *testing.T, commands *int, actions map[int]string) func(action string, c *exec.Cmd) error {
 	return func(action string, c *exec.Cmd) error {
 		log.Printf("PROC TRAP %d for %s. %+v", *commands, action, c)
+
 		assert.Equal(t, "daemon", c.Args[1])
 		assert.Equal(t, "--type=osd", c.Args[2])
 		assert.Equal(t, "--", c.Args[3])
@@ -282,7 +278,7 @@ func createOSDAgentProcTrap(t *testing.T, commands *int, actions map[int]string)
 
 		switch {
 		case *commands == 0:
-			err := ioutil.WriteFile("/tmp/osd3/mycluster-3/keyring", []byte("mykeyring"), 0644)
+			err := ioutil.WriteFile("/tmp/osd3/keyring", []byte("mykeyring"), 0644)
 			assert.Nil(t, err)
 			assert.Equal(t, "--mkfs", c.Args[4])
 			assert.Equal(t, "--mkkey", c.Args[5])

pkg/cephmgr/partition/scheme.go

@@ -0,0 +1,145 @@
+package partition
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"path"
+
+	"github.com/google/uuid"
+)
+
+const (
+	WalPartitionName      = "wal"
+	DatabasePartitionName = "db"
+	BlockPartitionName    = "block"
+	walPartition          = 1
+	databasePartition     = 2
+	blockPartition        = 3
+	UseRemainingSpace     = -1
+	SimpleVersion         = 1
+	schemeFilename        = "partition-scheme"
+)
+
+type Scheme struct {
+	Version        int               `json:"version"`
+	SizeMB         int               `json:"sizeMb"`
+	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) {
+	diskUUID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get disk uuid. %+v", err)
+	}
+	walUUID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get wal uuid. %+v", err)
+	}
+	dbUUID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get db uuid. %+v", err)
+	}
+	blockUUID, err := uuid.NewRandom()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get block uuid. %+v", err)
+	}
+
+	uuids := map[string]string{
+		WalPartitionName:      walUUID.String(),
+		DatabasePartitionName: dbUUID.String(),
+		BlockPartitionName:    blockUUID.String(),
+	}
+
+	return &Scheme{
+		Version:        SimpleVersion,
+		SizeMB:         sizeMB,
+		DiskUUID:       diskUUID.String(),
+		PartitionUUIDs: uuids,
+	}, nil
+}
+
+// Turn the scheme into arguments for sgdisk. The result will be three partitions:
+// 1. WAL: Write ahead log (10%)
+// 2. DB: The bluestore database (10%)
+// 3. Block: The raw storage for the data being written to the OSD. (80%)
+//     Uses the remainder of the storage space after small partitions for the wal and db.
+func (s *Scheme) GetArgs(name string) []string {
+	offset := 1
+	tenPercent := s.SizeMB / 10
+
+	// append args for individual partitions
+	args := s.getPartitionArgs(WalPartitionName, walPartition, offset, tenPercent, fmt.Sprintf("osd-wal-%s", s.DiskUUID))
+	offset += tenPercent
+	args = append(args, s.getPartitionArgs(DatabasePartitionName, databasePartition, offset, tenPercent, fmt.Sprintf("osd-db-%s", s.DiskUUID))...)
+	offset += tenPercent
+	args = append(args, s.getPartitionArgs(BlockPartitionName, blockPartition, offset, UseRemainingSpace, fmt.Sprintf("osd-block-%s", s.DiskUUID))...)
+
+	// append args for the whole device
+	args = append(args, fmt.Sprintf("--disk-guid=%s", s.DiskUUID))
+	args = append(args, "/dev/"+name)
+
+	return args
+}
+
+// Save the partition scheme to the config dir
+func (s *Scheme) Save(configDir string) error {
+	b, err := json.Marshal(s)
+	if err != nil {
+		return err
+	}
+
+	err = ioutil.WriteFile(path.Join(configDir, schemeFilename), b, 0644)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Load the persistent partition info from the config directory.
+func LoadScheme(configDir string) (*Scheme, error) {
+	b, err := ioutil.ReadFile(path.Join(configDir, schemeFilename))
+	if err != nil {
+		return nil, err
+	}
+
+	var scheme Scheme
+	err = json.Unmarshal(b, &scheme)
+	if err != nil {
+		return nil, err
+	}
+
+	return &scheme, nil
+}
+
+// Get the arguments necessary to create an sgdisk partition with the given parameters.
+// The id is the partition number.
+// The offset and length are in MB. Under the covers this is translated to sectors.
+// If the length is -1, all remaining space will be assigned to the partition.
+func (s *Scheme) getPartitionArgs(name string, id int, offset, length int, label string) []string {
+	const sectorsPerMB = 2048
+	var newPart string
+	if length == UseRemainingSpace {
+		// The partition gets the remainder of the device
+		newPart = fmt.Sprintf("--largest-new=%d", id)
+	} else {
+		// The partition has a specific offset and length
+		newPart = fmt.Sprintf("--new=%d:%d:+%d", id, offset*sectorsPerMB, length*sectorsPerMB)
+	}
+
+	guid, ok := s.PartitionUUIDs[name]
+	if !ok {
+		log.Printf("could not find uuid for partition %s", name)
+	}
+
+	return []string{
+		newPart,
+		fmt.Sprintf("--change-name=%d:%s", id, label),
+		fmt.Sprintf("--partition-guid=%d:%s", id, guid),
+	}
+}

pkg/cephmgr/partition/scheme_test.go

@@ -0,0 +1,31 @@
+package partition
+
+import (
+	"log"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSimpleScheme(t *testing.T) {
+	scheme, err := GetSimpleScheme(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))
+
+	args := scheme.GetArgs("foo")
+	assert.Equal(t, 11, len(args))
+
+	err = scheme.Save("/tmp")
+	defer os.Remove(path.Join("/tmp", schemeFilename))
+	assert.Nil(t, err)
+
+	loaded, err := LoadScheme("/tmp")
+	assert.Nil(t, err)
+	assert.Equal(t, 3, len(loaded.PartitionUUIDs))
+
+	log.Printf("scheme=%+v", scheme)
+}

pkg/cephmgr/test/conn.go

@@ -1,19 +1,19 @@
 package test
 
 import (
-	etcd "github.com/coreos/etcd/client"
 	"github.com/quantum/castle/pkg/cephmgr/client"
+	"github.com/quantum/castle/pkg/clusterd"
 )
 
 type MockConnectionFactory struct {
-	MockConnectAsAdmin func(cephFactory client.ConnectionFactory, etcdClient etcd.KeysAPI) (client.Connection, error)
+	MockConnectAsAdmin func(context *clusterd.Context, cephFactory client.ConnectionFactory) (client.Connection, error)
 }
 
 func (m *MockConnectionFactory) ConnectAsAdmin(
-	cephFactory client.ConnectionFactory, etcdClient etcd.KeysAPI) (client.Connection, error) {
+	context *clusterd.Context, cephFactory client.ConnectionFactory) (client.Connection, error) {
 
 	if m.MockConnectAsAdmin != nil {
-		return m.MockConnectAsAdmin(cephFactory, etcdClient)
+		return m.MockConnectAsAdmin(context, cephFactory)
 	}
 
 	return cephFactory.NewConnWithClusterAndUser("mycluster", "admin")

pkg/clusterd/clusterMember.go

@@ -2,6 +2,7 @@ package clusterd
 
 import (
 	"errors"
+	"fmt"
 	"log"
 	"path"
 	"sync/atomic"
@@ -15,6 +16,7 @@ import (
 )
 
 const (
+	SystemLogDir                     = "/var/log/castle"
 	leaseVersion                     = 1
 	leaseTtlSeconds                  = 15
 	leaseRetrySeconds                = 5
@@ -71,7 +73,13 @@ func newClusterMember(context *Context, leaseManager Manager, leader Leader) *Cl
 func (r *ClusterMember) initialize() error {
 	err := r.ElectLeader()
 	if err != nil {
-		return err
+		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.
@@ -148,15 +156,16 @@ func (r *ClusterMember) refreshLeader() {
 
 func (r *ClusterMember) discoverHardwareLoop() {
 	for {
+		// sleep until it's time to detect hardware again
+		// assume the initial discovery was already done
+		<-time.After(hardwareDiscoveryInterval)
+
 		err := r.discoverHardware()
 		if err != nil {
 			log.Printf("error while discovering hardware: %+v", err)
 		} else {
 			log.Print("hardware discovery complete")
 		}
-
-		// sleep until it's time to detect hardware again
-		<-time.After(hardwareDiscoveryInterval)
 	}
 }
 

pkg/clusterd/clusterservice.go

@@ -66,6 +66,9 @@ type Context struct {
 
 	// The process manager for launching a process
 	ProcMan *proc.ProcManager
+
+	// The root configuration directory used by services
+	ConfigDir string
 }
 
 func copyContext(c *Context) *Context {
@@ -76,6 +79,7 @@ func copyContext(c *Context) *Context {
 		Executor:   c.Executor,
 		ProcMan:    c.ProcMan,
 		Inventory:  c.Inventory,
+		ConfigDir:  c.ConfigDir,
 	}
 }
 

pkg/clusterd/inventory/disk.go

@@ -51,6 +51,31 @@ func StrToDiskType(diskType string) (DiskType, error) {
 	}
 }
 
+func GetSizeFromSerial(serial, nodeID string, etcdClient etcd.KeysAPI) (uint64, error) {
+	key := path.Join(GetNodeConfigKey(nodeID), DisksKey, serial, 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 GetDeviceFromSerial(serial, nodeID string, etcdClient etcd.KeysAPI) (string, error) {
+	key := path.Join(GetNodeConfigKey(nodeID), DisksKey, serial, DiskNameKey)
+	resp, err := etcdClient.Get(ctx.Background(), key, nil)
+	if err != nil {
+		return "", err
+	}
+
+	return resp.Node.Value, nil
+}
+
 func GetSerialFromDevice(device, nodeID string, etcdClient etcd.KeysAPI) (string, error) {
 	disksKey := path.Join(GetNodeConfigKey(nodeID), DisksKey)
 	serials, err := util.GetDirChildKeys(etcdClient, disksKey)

pkg/clusterd/inventory/node.go

@@ -61,7 +61,7 @@ func loadNodeConfig(etcdClient etcd.KeysAPI) (map[string]*NodeConfig, error) {
 		return nodesConfig, nil
 	}
 
-	log.Printf("Discovered %d nodes", len(nodeInfo.Node.Nodes))
+	// verbose: log.Printf("Discovered %d nodes", len(nodeInfo.Node.Nodes))
 
 	for _, etcdNode := range nodeInfo.Node.Nodes {
 		nodeConfig := &NodeConfig{}
@@ -128,7 +128,7 @@ func loadSingleNodeHealth(node *NodeConfig, health *etcd.Node) error {
 		switch util.GetLeafKeyPath(prop.Key) {
 		case HeartbeatKey:
 			node.HeartbeatAge = HeartbeatTtlDuration - (time.Duration(prop.TTL) * time.Second)
-			log.Printf("Node %s has age of %s", node.IPAddress, node.HeartbeatAge.String())
+			// verbose: log.Printf("Node %s has age of %s", node.IPAddress, node.HeartbeatAge.String())
 		default:
 			return fmt.Errorf("unknown node health key %s", prop.Key)
 		}

pkg/clusterd/join.go

@@ -13,9 +13,9 @@ import (
 )
 
 // StartJoinCluster initializes the cluster services to enable joining the cluster and listening for orchestration.
-func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, nodeID, discoveryURL, etcdMembers, publicIPv4, privateIPv4 string) (*Context, error) {
-	log.Printf("Starting cluster. nodeID=%s, url=%s, members=%s, publicIPv4=%s, privateIPv4=%s",
-		nodeID, discoveryURL, etcdMembers, publicIPv4, privateIPv4)
+func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, configDir, nodeID, discoveryURL, etcdMembers, publicIPv4, privateIPv4 string) (*Context, error) {
+	log.Printf("Starting cluster. config=%s, nodeID=%s, url=%s, members=%s, publicIPv4=%s, privateIPv4=%s",
+		configDir, nodeID, discoveryURL, etcdMembers, publicIPv4, privateIPv4)
 
 	etcdClients := []string{}
 	if etcdMembers != "" {
@@ -25,7 +25,7 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, nod
 
 		// use the discovery URL to query the etcdmgr what the etcd client endpoints should be
 		var err error
-		etcdClients, err = manager.GetEtcdClients(discoveryURL, privateIPv4, nodeID)
+		etcdClients, err = manager.GetEtcdClients(configDir, discoveryURL, privateIPv4, nodeID)
 		if err != nil {
 			return nil, err
 		}
@@ -57,6 +57,7 @@ func StartJoinCluster(services []*ClusterService, procMan *proc.ProcManager, nod
 		NodeID:     nodeID,
 		Services:   services,
 		ProcMan:    procMan,
+		ConfigDir:  configDir,
 	}
 	clusterLeader := newServicesLeader(context)
 	clusterMember := newClusterMember(context, leaseManager, clusterLeader)

pkg/clusterd/status.go

@@ -143,7 +143,6 @@ func GetNodeStatusKey(service, nodeID string) string {
 // If the taskKey is the empty string, set the status for the node.
 func SetNodeConfigStatus(etcdClient etcd.KeysAPI, nodeID, taskKey string, nodeStatus NodeConfigStatus) error {
 	key := GetNodeStatusKey(taskKey, nodeID)
-	fmt.Println("key: ", key)
 	_, err := etcdClient.Set(ctx.Background(), key, nodeStatus.String(), nil)
 	return err
 }

pkg/etcdmgr/bootstrap/config.go

@@ -3,6 +3,7 @@ package bootstrap
 import (
 	"log"
 	"net/url"
+	"path"
 	"time"
 
 	"github.com/coreos/etcd/pkg/types"
@@ -44,7 +45,7 @@ func CopyConfig(conf *Config) Config {
 }
 
 // GenerateConfig automatically generates a config for embedded etcd
-func GenerateConfig(ipAddr, nodeID string) (*Config, error) {
+func GenerateConfig(configDir, ipAddr, nodeID string) (*Config, error) {
 	conf := &Config{}
 	conf.InstanceName = nodeID
 	conf.ListenPeerURLs = append(conf.ListenPeerURLs,
@@ -55,15 +56,15 @@ func GenerateConfig(ipAddr, nodeID string) (*Config, error) {
 		getURLFromSchemeIPPort("http", ipAddr, DefaultPeerPort))
 	conf.AdvertiseClientURLs = append(conf.AdvertiseClientURLs,
 		getURLFromSchemeIPPort("http", ipAddr, DefaultClientPort))
-	conf.DataDir = "/tmp/etcd-data"
+	conf.DataDir = path.Join(configDir, "etcd-data")
 	conf.URLsMap = types.URLsMap{}
 
 	return conf, nil
 }
 
 // GenerateConfigFromExistingCluster automatically generates a config for joining an existing cluster
-func GenerateConfigFromExistingCluster(Context EtcdMgrContext, ipAddr, nodeID string) (*Config, error) {
-	conf, err := GenerateConfig(ipAddr, nodeID)
+func GenerateConfigFromExistingCluster(Context EtcdMgrContext, configDir, ipAddr, nodeID string) (*Config, error) {
+	conf, err := GenerateConfig(configDir, ipAddr, nodeID)
 	if err != nil {
 		return nil, err
 	}

pkg/etcdmgr/etcdmgragent.go

@@ -70,7 +70,7 @@ func (e *etcdMgrAgent) CreateLocalService(context *clusterd.Context, desiredKey
 	}
 	ipAddr := resp.Node.Value
 	log.Println("ipAddress: ", ipAddr)
-	e.conf, err = bootstrap.GenerateConfigFromExistingCluster(e.context, ipAddr, context.NodeID)
+	e.conf, err = bootstrap.GenerateConfigFromExistingCluster(e.context, context.ConfigDir, ipAddr, context.NodeID)
 	log.Println("config: ", e.conf)
 	if err != nil {
 		return err

pkg/etcdmgr/etcdmgrleader_test.go

@@ -81,7 +81,6 @@ func TestEtcdMgrLeaderShrink(t *testing.T) {
 	etcdClient.WatcherResponses["/castle/services/etcd/desired/a/ipaddress"] = "1.2.3.4"
 	etcdClient.WatcherResponses["/castle/services/etcd/desired/b/ipaddress"] = "2.3.4.5"
 	etcdClient.WatcherResponses["/castle/services/etcd/desired/c/ipaddress"] = "2.3.4.5"
-	etcdClient.Dump()
 
 	unhealthyNode := clusterd.UnhealthyNode{AgeSeconds: 10, NodeID: "c"}
 	unhealthyNodeEvt := clusterd.NewUnhealthyNodeEvent(context, []*clusterd.UnhealthyNode{&unhealthyNode})
@@ -89,7 +88,7 @@ func TestEtcdMgrLeaderShrink(t *testing.T) {
 
 	// wait for the event queue to be empty
 	clusterdtest.WaitForEvents(&etcdmgrService)
-	etcdClient.Dump()
+
 	// two possibilities: there might be a&c or b&c in the desired states
 	assert.True(t, etcdClient.GetChildDirs("/castle/services/etcd/desired").Count() == 0)
 }

pkg/etcdmgr/manager/etcdmgr.go

@@ -12,7 +12,7 @@ import (
 
 // GetEtcdClients bootstraps an embedded etcd instance and returns a list of
 // current etcd cluster's client URLs. (entrypoint, when it's used as a library)
-func GetEtcdClients(token, ipAddr, nodeID string) ([]string, error) {
+func GetEtcdClients(configDir, token, ipAddr, nodeID string) ([]string, error) {
 	// GODEBUG setting forcing use of Go's resolver
 	os.Setenv("GODEBUG", "netdns=go+1")
 
@@ -40,7 +40,7 @@ func GetEtcdClients(token, ipAddr, nodeID string) ([]string, error) {
 	}
 
 	log.Println("quorum is not complete, creating a new embedded etcd member...")
-	conf, err := bootstrap.GenerateConfig(ipAddr, nodeID)
+	conf, err := bootstrap.GenerateConfig(configDir, ipAddr, nodeID)
 	if err != nil {
 		return []string{}, err
 	}

pkg/util/proc/procmanager.go

@@ -29,6 +29,7 @@ type ProcManager struct {
 }
 
 func (p *ProcManager) Run(daemon string, args ...string) error {
+
 	log.Printf("Running process %s with args: %v", daemon, args)
 	err := p.runChildProcess(daemon, args...)
 	if err != nil {
@@ -197,8 +198,10 @@ func (p *ProcManager) startChildProcessCmd(cmd *exec.Cmd) (err error) {
 	return cmd.Start()
 }
 
-func createCmd(daemon string, args ...string) (cmd *exec.Cmd) {
-	cmd = exec.Command(os.Args[0], append([]string{"daemon", fmt.Sprintf("--type=%s", daemon), "--"}, args...)...)
+func createCmd(daemon string, args ...string) *exec.Cmd {
+
+	cmd := exec.Command(os.Args[0], append([]string{"daemon", fmt.Sprintf("--type=%s", daemon), "--"}, args...)...)
+
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 

scripts/clean-castled.sh

@@ -1,14 +1,9 @@
 #!/bin/bash
 
-# clean up mon dirs
-rm -fr /tmp/mon*
 
-# clean up bootstrap-osd dir
-rm -fr /tmp/bootstrap-osd
-
-# unmount OSD volumes and delete their directories
-for dir in `mount | grep -E '/tmp/osd[0-9]+ ' | awk '{print $3}'`; do sudo umount -f ${dir}; done
-rm -fr /tmp/osd*
+sudo rm -fr mon0 mon1 mon2 mon3
+sudo rm -fr /var/lib/castle
+sudo rm -fr /var/log/castle
 
 # clean the etcd config
 rm -fr /tmp/etcd-data >/dev/null 2>&1
@@ -19,4 +14,10 @@ rm -fr /tmp/castle-discovery-url
 rm -fr /tmp/castle >/dev/null 2>&1
 
 # ensure castled processes are dead if there was a crash
-ps aux | grep castled | grep -E -v 'grep|clean-castled' | awk '{print $2}' | xargs kill >/dev/null 2>&1
+ps aux | grep castled | grep -E -v 'grep|clean-castled' | awk '{print $2}' | xargs kill >/dev/null 2>&1
+
+# clear the data partitions
+for DEV in sdb sdc sdd; do
+    sudo sgdisk --zap-all /dev/$DEV
+done
+