diff --git a/ceph b/ceph index 5e221dbee..49fdbcfdb 160000 --- a/ceph +++ b/ceph @@ -1 +1 @@ -Subproject commit 5e221dbee418242fa07f16315cf44d5f8d69632b +Subproject commit 49fdbcfdbbe6afa6cc0447c49fa2d61465507a89 diff --git a/cmd/rookd/daemon.go b/cmd/rookd/daemon.go index 867451c5d..a7bf0f2ea 100644 --- a/cmd/rookd/daemon.go +++ b/cmd/rookd/daemon.go @@ -47,7 +47,7 @@ func runDaemon(cmd *cobra.Command, args []string) error { if err := flags.VerifyRequiredFlags(daemonCmd, []string{"type"}); err != nil { return err } - if daemonType != "mon" && daemonType != "osd" && daemonType != "mds" && daemonType != "rgw" { + if daemonType != "mon" && daemonType != "osd" && daemonType != "mds" && daemonType != "rgw" && daemonType != "rgw-admin" { return fmt.Errorf("unknown daemon type: %s", daemonType) } diff --git a/pkg/cephmgr/cephd/cephd.go b/pkg/cephmgr/cephd/cephd.go index 9bba714f3..b2abf72e1 100644 --- a/pkg/cephmgr/cephd/cephd.go +++ b/pkg/cephmgr/cephd/cephd.go @@ -159,6 +159,8 @@ func (c *ceph) RunDaemon(daemon string, args ...string) error { ret = C.cephd_run_mds(C.int(len(finalArgs)), (**C.char)(ptr)) } else if daemon == "rgw" { ret = C.cephd_run_rgw(C.int(len(finalArgs)), (**C.char)(ptr)) + } else if daemon == "rgw-admin" { + ret = C.cephd_run_rgw_admin(C.int(len(finalArgs)), (**C.char)(ptr)) } if ret < 0 { return GetCephdError(int(ret)) diff --git a/pkg/cephmgr/mon/agent.go b/pkg/cephmgr/mon/agent.go index 1f946ce82..547ca8594 100644 --- a/pkg/cephmgr/mon/agent.go +++ b/pkg/cephmgr/mon/agent.go @@ -122,7 +122,8 @@ func (a *agent) DestroyLocalService(context *clusterd.Context) error { // creates and initializes the given monitors file systems func (a *agent) makeMonitorFileSystem(context *clusterd.Context, cluster *ClusterInfo, monName string) error { // write the keyring to disk - if err := writeMonitorKeyring(context.ConfigDir, monName, cluster); err != nil { + keyringPath := getMonKeyringPath(context.ConfigDir, monName) + if err := writeMonitorKeyring(monName, cluster, keyringPath); err != nil { return err } @@ -194,9 +195,8 @@ func (a *agent) runMonitor(context *clusterd.Context, cluster *ClusterInfo, moni } // writes the monitor keyring to disk -func writeMonitorKeyring(configDir, monName string, c *ClusterInfo) error { +func writeMonitorKeyring(monName string, c *ClusterInfo, keyringPath string) error { keyring := fmt.Sprintf(monitorKeyringTemplate, c.MonitorSecret, c.AdminSecret) - 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/mon/config.go b/pkg/cephmgr/mon/config.go index 355d75d3f..cb05d36d4 100644 --- a/pkg/cephmgr/mon/config.go +++ b/pkg/cephmgr/mon/config.go @@ -90,7 +90,8 @@ func ConnectToClusterAsAdmin(context *clusterd.Context, factory client.Connectio } // write the monitor keyring to disk monName := getFirstMonitor(cluster) - if err := writeMonitorKeyring(context.ConfigDir, monName, cluster); err != nil { + keyringPath := getMonKeyringPath(context.ConfigDir, monName) + if err := writeMonitorKeyring(monName, cluster, keyringPath); err != nil { return nil, err } @@ -103,6 +104,19 @@ func GetConfFilePath(root, clusterName string) string { return fmt.Sprintf("%s/%s.config", root, clusterName) } +func GenerateTempConfigFiles(context *clusterd.Context, cluster *ClusterInfo) (string, string, string, error) { + root := path.Join(context.ConfigDir, "tmp") + keyring := path.Join(root, "keyring") + user := getFirstMonitor(cluster) + err := writeMonitorKeyring(user, cluster, keyring) + if err != nil { + return "", "", "", fmt.Errorf("failed to write keyring to %s", root) + } + + configFile, err := GenerateConfigFile(context, cluster, root, user, keyring, false, nil, nil) + return configFile, keyring, user, err +} + // generates and writes the monitor config file to disk func GenerateConnectionConfigFile(context *clusterd.Context, cluster *ClusterInfo, pathRoot, user, keyringPath string) (string, error) { return GenerateConfigFile(context, cluster, pathRoot, user, keyringPath, false, nil, nil) diff --git a/pkg/cephmgr/rgw/admin.go b/pkg/cephmgr/rgw/admin.go new file mode 100644 index 000000000..592621fb3 --- /dev/null +++ b/pkg/cephmgr/rgw/admin.go @@ -0,0 +1,53 @@ +/* +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package rgw + +import ( + "fmt" + + "github.com/rook/rook/pkg/cephmgr/mon" + "github.com/rook/rook/pkg/clusterd" +) + +// Runs the embedded radosgw-admin command and returns its output +func RunAdminCommand(context *clusterd.Context, command, subcommand string, args ...string) (string, error) { + cluster, err := mon.LoadClusterInfo(context.EtcdClient) + if err != nil { + return "", fmt.Errorf("failed to get cluster name. %+v", err) + } + + confFile, keyringFile, _, err := mon.GenerateTempConfigFiles(context, cluster) + if err != nil { + return "", fmt.Errorf("failed to generate config file. %+v", err) + } + + options := []string{ + command, + subcommand, + fmt.Sprintf("--cluster=%s", cluster.Name), + fmt.Sprintf("--conf=%s", confFile), + fmt.Sprintf("--keyring=%s", keyringFile), + } + options = append(options, args...) + + // start the monitor daemon in the foreground with the given config + output, err := context.ProcMan.RunWithOutput("rgw-admin", "rgw-admin", options...) + if err != nil { + return "", fmt.Errorf("failed to run rgw-admin: %+v", err) + } + + return output, nil +} diff --git a/pkg/cephmgr/rgw/leader.go b/pkg/cephmgr/rgw/leader.go index f45f31433..ddf60dd0f 100644 --- a/pkg/cephmgr/rgw/leader.go +++ b/pkg/cephmgr/rgw/leader.go @@ -89,6 +89,12 @@ func RemoveObjectStore(context *clusterd.Context) error { return fmt.Errorf("failed to remove object store from desired state. %+v", err) } + key = path.Join(mon.CephKey, ObjectStoreKey, clusterd.AppliedKey, builtinUserKey) + _, err = context.EtcdClient.Delete(ctx.Background(), key, &etcd.DeleteOptions{Dir: true, Recursive: true}) + if err != nil { + return fmt.Errorf("failed to remove object store from desired state. %+v", err) + } + _, err = context.EtcdClient.Delete(ctx.Background(), getRGWNodesKey(false), &etcd.DeleteOptions{Dir: true, Recursive: true}) if err != nil { return fmt.Errorf("failed to remove rgw nodes from desired state. %+v", err) @@ -136,6 +142,7 @@ func markUnapplied(context *clusterd.Context) error { } func (r *Leader) enable(context *clusterd.Context, factory client.ConnectionFactory) error { + // start an instance of rgw on every node count := len(context.Inventory.Nodes) nodes, err := r.getDesiredRGWNodes(context, count) @@ -157,6 +164,12 @@ func (r *Leader) enable(context *clusterd.Context, factory client.ConnectionFact } } + // create the built-in rgw user + err = createBuiltinUser(context) + if err != nil { + return fmt.Errorf("failed to get create first user. %+v", err) + } + return markApplied(context) } diff --git a/pkg/cephmgr/rgw/leader_test.go b/pkg/cephmgr/rgw/leader_test.go index cd975eac4..e37fbdef8 100644 --- a/pkg/cephmgr/rgw/leader_test.go +++ b/pkg/cephmgr/rgw/leader_test.go @@ -16,11 +16,15 @@ limitations under the License. package rgw import ( + "os" "path" "testing" testceph "github.com/rook/rook/pkg/cephmgr/client/test" + cephtest "github.com/rook/rook/pkg/cephmgr/test" "github.com/rook/rook/pkg/clusterd/inventory" + exectest "github.com/rook/rook/pkg/util/exec/test" + "github.com/rook/rook/pkg/util/proc" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util" @@ -34,9 +38,21 @@ func TestRGWConfig(t *testing.T) { "a": &inventory.NodeConfig{PublicIP: "1.2.3.4"}, "b": &inventory.NodeConfig{PublicIP: "2.3.4.5"}, } - context := &clusterd.Context{EtcdClient: etcdClient, Inventory: &inventory.Config{Nodes: nodes}} + executor := &exectest.MockExecutor{} + context := &clusterd.Context{EtcdClient: etcdClient, Inventory: &inventory.Config{Nodes: nodes}, + ProcMan: proc.New(executor), Executor: executor, ConfigDir: "/tmp/rgw"} factory := &testceph.MockConnectionFactory{Fsid: "f", SecretKey: "k"} leader := NewLeader() + executor.MockExecuteCommandWithOutput = func(actionName, command string, args ...string) (string, error) { + response := `{"keys": [ + {"access_key": "myaccessid", "secret_key": "mybigsecretkey"} + ]}` + + return response, nil + } + defer os.RemoveAll("/tmp/rgw") + // mock a monitor + cephtest.CreateClusterInfo(etcdClient, []string{"mymon"}) // Nothing happens when not in desired state err := leader.Configure(context, factory) @@ -61,10 +77,14 @@ func TestRGWConfig(t *testing.T) { err = leader.Configure(context, factory) assert.Nil(t, err) verifyObjectConfigured(t, context, true) + assert.Equal(t, "myaccessid", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/id")) + assert.Equal(t, "mybigsecretkey", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/_secret")) // Remove the object service err = RemoveObjectStore(context) assert.Nil(t, err) + assert.Equal(t, "", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/id")) + assert.Equal(t, "", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/_secret")) err = leader.Configure(context, factory) assert.Nil(t, err) verifyObjectConfigured(t, context, false) diff --git a/pkg/cephmgr/rgw/user.go b/pkg/cephmgr/rgw/user.go new file mode 100644 index 000000000..287317cfd --- /dev/null +++ b/pkg/cephmgr/rgw/user.go @@ -0,0 +1,80 @@ +/* +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package rgw + +import ( + "encoding/json" + "fmt" + "path" + + ctx "golang.org/x/net/context" + + "github.com/rook/rook/pkg/cephmgr/mon" + "github.com/rook/rook/pkg/clusterd" +) + +const ( + builtinUserKey = "admin" +) + +type user struct { + Keys []keyInfo `json:"keys"` +} + +type keyInfo struct { + User string `json:"user"` + AccessID string `json:"access_key"` + SecretKey string `json:"secret_key"` +} + +func createBuiltinUser(context *clusterd.Context) error { + logger.Infof("creating the built-in rgw user") + result, err := RunAdminCommand(context, + "user", + "create", + fmt.Sprintf("--uid=%s", "rookadmin"), + fmt.Sprintf("--display-name=%s", "rook rgw builtin user")) + if err != nil { + return fmt.Errorf("failed to create user: %+v", err) + } + + // Parse the creds from the json response + var u user + if err := json.Unmarshal([]byte(result), &u); err != nil { + return fmt.Errorf("failed to read user info. %+v, result=%s", err, result) + } + + if len(u.Keys) == 0 { + return fmt.Errorf("missing keys in %s", result) + } + + userkey := u.Keys[0] + if userkey.AccessID == "" || userkey.SecretKey == "" { + return fmt.Errorf("missing user properties in %s", result) + } + + // store the creds in etcd + key := path.Join(mon.CephKey, ObjectStoreKey, clusterd.AppliedKey, builtinUserKey) + if _, err := context.EtcdClient.Set(ctx.Background(), path.Join(key, "id"), userkey.AccessID, nil); err != nil { + return fmt.Errorf("failed to store access id. %+v", err) + } + + if _, err := context.EtcdClient.Set(ctx.Background(), path.Join(key, "_secret"), userkey.SecretKey, nil); err != nil { + return fmt.Errorf("failed to store access id. %+v", err) + } + + return nil +} diff --git a/pkg/cephmgr/rgw/user_test.go b/pkg/cephmgr/rgw/user_test.go new file mode 100644 index 000000000..8f815a967 --- /dev/null +++ b/pkg/cephmgr/rgw/user_test.go @@ -0,0 +1,116 @@ +/* +Copyright 2016 The Rook Authors. All rights reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ +package rgw + +import ( + "os" + "testing" + + cephtest "github.com/rook/rook/pkg/cephmgr/test" + exectest "github.com/rook/rook/pkg/util/exec/test" + "github.com/rook/rook/pkg/util/proc" + + "github.com/rook/rook/pkg/clusterd" + "github.com/rook/rook/pkg/util" + + "github.com/stretchr/testify/assert" +) + +func TestBuiltinUser(t *testing.T) { + etcdClient := util.NewMockEtcdClient() + executor := &exectest.MockExecutor{} + context := &clusterd.Context{EtcdClient: etcdClient, + ProcMan: proc.New(executor), Executor: executor, ConfigDir: "/tmp/rgw"} + response := `{ + "user_id": "testadmin", + "display_name": "my display name", + "max_buckets": 1000, + "auid": 0, + "subusers": [], + "keys": [ + { + "user": "rookadmin", + "access_key": "myaccessid", + "secret_key": "mybigsecretkey" + } + ], + "temp_url_keys": [], + "type": "rgw" + }` + executor.MockExecuteCommandWithOutput = func(actionName, command string, args ...string) (string, error) { + assert.Equal(t, "user", args[3]) + assert.Equal(t, "create", args[4]) + assert.Equal(t, 10, len(args)) + + return response, nil + } + defer os.RemoveAll("/tmp/rgw") + + // mock a monitor + cephtest.CreateClusterInfo(etcdClient, []string{"mymon"}) + + // create a valid user + err := createBuiltinUser(context) + assert.Nil(t, err) + assert.Equal(t, "myaccessid", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/id")) + assert.Equal(t, "mybigsecretkey", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/_secret")) + + // test various forms of invalid responses for the user + response = "{}" + err = createBuiltinUser(context) + assert.NotNil(t, err) + + // no keys + response = `"{"keys": []}` + err = createBuiltinUser(context) + assert.NotNil(t, err) + + // no access key + response = `"{"keys": [ + { + "accss_key": "foo", + "secret_key": "bar" + } + ]}` + err = createBuiltinUser(context) + assert.NotNil(t, err) + + // no secret key + response = `"{"keys": [ + { + "access_key": "foo", + "secrt_key": "bar" + } + ]}` + err = createBuiltinUser(context) + assert.NotNil(t, err) + assert.Equal(t, "myaccessid", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/id")) + assert.Equal(t, "mybigsecretkey", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/_secret")) + + // valid key + response = `{ + "keys": [ + { + "access_key": "foo", + "secret_key": "bar" + } + ] + }` + err = createBuiltinUser(context) + assert.Nil(t, err) + assert.Equal(t, "foo", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/id")) + assert.Equal(t, "bar", etcdClient.GetValue("/rook/services/ceph/object/applied/admin/_secret")) +} diff --git a/pkg/util/proc/procmanager.go b/pkg/util/proc/procmanager.go index 6bc8d1a41..935cc03a0 100644 --- a/pkg/util/proc/procmanager.go +++ b/pkg/util/proc/procmanager.go @@ -46,6 +46,18 @@ func New(executor exec.Executor) *ProcManager { return &ProcManager{executor: executor} } +// Start a child process and wait for its completion +func (p *ProcManager) RunWithOutput(logName, daemon string, args ...string) (string, error) { + + logger.Infof("Running process %s with args: %v", daemon, args) + output, err := p.executor.ExecuteCommandWithOutput(logName, os.Args[0], createDaemonArgs(daemon, args...)...) + if err != nil { + return "", fmt.Errorf("failed to run %s: %+v", daemon, err) + } + + return output, nil +} + // Start a child process and wait for its completion func (p *ProcManager) Run(logName, daemon string, args ...string) error {