diff --git a/.gitignore b/.gitignore index 3f42f554e..75fb01e0c 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,5 @@ debug.test .vagrant/ demo/vagrant/cloud-config.yml + +.virtualbox/ \ No newline at end of file diff --git a/cmd/castle/castle.go b/cmd/castle/castle.go index 04734d50e..bb648ea84 100644 --- a/cmd/castle/castle.go +++ b/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) diff --git a/cmd/castled/main.go b/cmd/castled/main.go index ee617f56b..f4ec58a5b 100644 --- a/cmd/castled/main.go +++ b/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 diff --git a/demo/vagrant/Vagrantfile b/demo/vagrant/Vagrantfile index 3e363a8a2..39674728d 100644 --- a/demo/vagrant/Vagrantfile +++ b/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'] diff --git a/demo/vagrant/cloud-config.yml.in b/demo/vagrant/cloud-config.yml.in index e10c5616d..299013fc6 100644 --- a/demo/vagrant/cloud-config.yml.in +++ b/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 \ diff --git a/demo/vagrant/util.rb b/demo/vagrant/util.rb index 99c9e9821..75267cd8c 100644 --- a/demo/vagrant/util.rb +++ b/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 + diff --git a/pkg/api/device.go b/pkg/api/device.go index b030001bb..ae6f12924 100644 --- a/pkg/api/device.go +++ b/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) diff --git a/pkg/api/handlers.go b/pkg/api/handlers.go index 4ab0b773f..a80bab919 100644 --- a/pkg/api/handlers.go +++ b/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) diff --git a/pkg/api/handlers_test.go b/pkg/api/handlers_test.go index 3a2613836..f56e0f955 100644 --- a/pkg/api/handlers_test.go +++ b/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()) diff --git a/pkg/api/node.go b/pkg/api/node.go index 7d5f1d3e0..162449df9 100644 --- a/pkg/api/node.go +++ b/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 diff --git a/pkg/cephmgr/cephleader.go b/pkg/cephmgr/cephleader.go index 4c97fc040..ba862c403 100644 --- a/pkg/cephmgr/cephleader.go +++ b/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 } } diff --git a/pkg/cephmgr/cephleader_test.go b/pkg/cephmgr/cephleader_test.go index 138b10b33..8744b3f24 100644 --- a/pkg/cephmgr/cephleader_test.go +++ b/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 diff --git a/pkg/cephmgr/config.go b/pkg/cephmgr/config.go index 1514a8981..72d3cd0f2 100644 --- a/pkg/cephmgr/config.go +++ b/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 } diff --git a/pkg/cephmgr/conn.go b/pkg/cephmgr/conn.go index 9ce272bca..e613a8b2f 100644 --- a/pkg/cephmgr/conn.go +++ b/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) } diff --git a/pkg/cephmgr/conn_test.go b/pkg/cephmgr/conn_test.go index fffab6fbd..97990ddb1 100644 --- a/pkg/cephmgr/conn_test.go +++ b/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) } diff --git a/pkg/cephmgr/mon.go b/pkg/cephmgr/mon.go index a376f6d7a..05bd0636c 100644 --- a/pkg/cephmgr/mon.go +++ b/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)) } diff --git a/pkg/cephmgr/monagent.go b/pkg/cephmgr/monagent.go index edc081aad..8e6233d7e 100644 --- a/pkg/cephmgr/monagent.go +++ b/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) } diff --git a/pkg/cephmgr/monagent_test.go b/pkg/cephmgr/monagent_test.go index 57b0005dd..166e3e8f5 100644 --- a/pkg/cephmgr/monagent_test.go +++ b/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"} diff --git a/pkg/cephmgr/monleader.go b/pkg/cephmgr/monleader.go index a75ea8ac3..619782d14 100644 --- a/pkg/cephmgr/monleader.go +++ b/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 } diff --git a/pkg/cephmgr/monleader_test.go b/pkg/cephmgr/monleader_test.go index 904cabf72..f8f1bf054 100644 --- a/pkg/cephmgr/monleader_test.go +++ b/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' diff --git a/pkg/cephmgr/osd.go b/pkg/cephmgr/osd.go index d2328b819..448a96615 100644 --- a/pkg/cephmgr/osd.go +++ b/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 { diff --git a/pkg/cephmgr/osd_test.go b/pkg/cephmgr/osd_test.go index b95bb52c1..355728d49 100644 --- a/pkg/cephmgr/osd_test.go +++ b/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) diff --git a/pkg/cephmgr/osdagent.go b/pkg/cephmgr/osdagent.go index 1e6adcc1d..ae15aca74 100644 --- a/pkg/cephmgr/osdagent.go +++ b/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 = "" + 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) +} diff --git a/pkg/cephmgr/osdagent_test.go b/pkg/cephmgr/osdagent_test.go index 5d9a7d82f..fe827e40d 100644 --- a/pkg/cephmgr/osdagent_test.go +++ b/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]) diff --git a/pkg/cephmgr/partition/scheme.go b/pkg/cephmgr/partition/scheme.go new file mode 100644 index 000000000..deda50f83 --- /dev/null +++ b/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), + } +} diff --git a/pkg/cephmgr/partition/scheme_test.go b/pkg/cephmgr/partition/scheme_test.go new file mode 100644 index 000000000..f14f4e5ac --- /dev/null +++ b/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) +} diff --git a/pkg/cephmgr/test/conn.go b/pkg/cephmgr/test/conn.go index b4cac6fe6..6affecd7d 100644 --- a/pkg/cephmgr/test/conn.go +++ b/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") diff --git a/pkg/clusterd/clusterMember.go b/pkg/clusterd/clusterMember.go index 47213a532..065da379a 100644 --- a/pkg/clusterd/clusterMember.go +++ b/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) } } diff --git a/pkg/clusterd/clusterservice.go b/pkg/clusterd/clusterservice.go index 0e699328e..e8cea8bc4 100644 --- a/pkg/clusterd/clusterservice.go +++ b/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, } } diff --git a/pkg/clusterd/inventory/disk.go b/pkg/clusterd/inventory/disk.go index 1a2743fa2..8111c0bad 100644 --- a/pkg/clusterd/inventory/disk.go +++ b/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) diff --git a/pkg/clusterd/inventory/node.go b/pkg/clusterd/inventory/node.go index 56936cb8a..b15697271 100644 --- a/pkg/clusterd/inventory/node.go +++ b/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) } diff --git a/pkg/clusterd/join.go b/pkg/clusterd/join.go index e438db0af..a4d16ba0e 100644 --- a/pkg/clusterd/join.go +++ b/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) diff --git a/pkg/clusterd/status.go b/pkg/clusterd/status.go index 63b4d2e79..b6be6a5f8 100644 --- a/pkg/clusterd/status.go +++ b/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 } diff --git a/pkg/etcdmgr/bootstrap/config.go b/pkg/etcdmgr/bootstrap/config.go index 65457ade3..9b64a6646 100644 --- a/pkg/etcdmgr/bootstrap/config.go +++ b/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 } diff --git a/pkg/etcdmgr/etcdmgragent.go b/pkg/etcdmgr/etcdmgragent.go index 0c2360d33..01815505a 100644 --- a/pkg/etcdmgr/etcdmgragent.go +++ b/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 diff --git a/pkg/etcdmgr/etcdmgrleader_test.go b/pkg/etcdmgr/etcdmgrleader_test.go index c4ea09169..1cd2584e4 100644 --- a/pkg/etcdmgr/etcdmgrleader_test.go +++ b/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) } diff --git a/pkg/etcdmgr/manager/etcdmgr.go b/pkg/etcdmgr/manager/etcdmgr.go index f6af75d7c..b3d0e760a 100644 --- a/pkg/etcdmgr/manager/etcdmgr.go +++ b/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 } diff --git a/pkg/util/proc/procmanager.go b/pkg/util/proc/procmanager.go index 452fd61fd..0ca9948c3 100644 --- a/pkg/util/proc/procmanager.go +++ b/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 diff --git a/scripts/clean-castled.sh b/scripts/clean-castled.sh index 60e7be9aa..81769fece 100755 --- a/scripts/clean-castled.sh +++ b/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 \ No newline at end of file +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 +