From 0ad60c065bd1e541d3fe7887549d50cd05c212e8 Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Thu, 26 Oct 2017 15:21:53 -0700 Subject: [PATCH] osd: refactor proc manager for only the osd daemon --- cmd/rook/main.go | 2 -- cmd/rook/mgr.go | 1 - cmd/rook/osd.go | 2 +- cmd/rook/rgw.go | 1 - pkg/ceph/mds/daemon.go | 2 +- pkg/ceph/mgr/daemon.go | 24 +++++++----------------- pkg/ceph/mon/daemon.go | 6 +++--- pkg/ceph/osd/agent.go | 10 ++++------ pkg/ceph/osd/agent_test.go | 10 +++------- pkg/ceph/osd/daemon_test.go | 3 +-- pkg/ceph/osd/filesystem.go | 10 +++------- pkg/ceph/rgw/daemon.go | 18 +++++------------- pkg/clusterd/context.go | 4 ---- 13 files changed, 28 insertions(+), 65 deletions(-) diff --git a/cmd/rook/main.go b/cmd/rook/main.go index 8fc1fc215..c6b832867 100644 --- a/cmd/rook/main.go +++ b/cmd/rook/main.go @@ -32,7 +32,6 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util/exec" "github.com/rook/rook/pkg/util/flags" - "github.com/rook/rook/pkg/util/proc" "github.com/rook/rook/pkg/version" apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" ) @@ -123,7 +122,6 @@ func createContext() *clusterd.Context { executor := &exec.CommandExecutor{} return &clusterd.Context{ Executor: executor, - ProcMan: proc.New(executor), ConfigDir: cfg.dataDir, ConfigFileOverride: cfg.cephConfigOverride, LogLevel: cfg.logLevel, diff --git a/cmd/rook/mgr.go b/cmd/rook/mgr.go index 039b8935e..8712fcf03 100644 --- a/cmd/rook/mgr.go +++ b/cmd/rook/mgr.go @@ -61,7 +61,6 @@ func startMgr(cmd *cobra.Command, args []string) error { Name: mgrName, Keyring: mgrKeyring, ClusterInfo: &clusterInfo, - InProc: true, } err := mgr.Run(createContext(), config) diff --git a/cmd/rook/osd.go b/cmd/rook/osd.go index 9030ea805..35f764dda 100644 --- a/cmd/rook/osd.go +++ b/cmd/rook/osd.go @@ -106,7 +106,7 @@ func startOSD(cmd *cobra.Command, args []string) error { forceFormat := false clusterInfo.Monitors = mon.ParseMonEndpoints(cfg.monEndpoints) - agent := osd.NewAgent(dataDevices, usingDeviceFilter, cfg.metadataDevice, cfg.directories, forceFormat, + agent := osd.NewAgent(context, dataDevices, usingDeviceFilter, cfg.metadataDevice, cfg.directories, forceFormat, crushLocation, cfg.storeConfig, &clusterInfo, cfg.nodeName, kv) err = osd.Run(context, agent) diff --git a/cmd/rook/rgw.go b/cmd/rook/rgw.go index 86fd52c87..f411bbf7e 100644 --- a/cmd/rook/rgw.go +++ b/cmd/rook/rgw.go @@ -77,7 +77,6 @@ func startRGW(cmd *cobra.Command, args []string) error { Port: rgwPort, SecurePort: rgwSecurePort, CertificatePath: rgwCert, - InProc: true, } err := rgw.Run(createContext(), config) diff --git a/pkg/ceph/mds/daemon.go b/pkg/ceph/mds/daemon.go index 6880290ca..b6b295bf2 100644 --- a/pkg/ceph/mds/daemon.go +++ b/pkg/ceph/mds/daemon.go @@ -118,7 +118,7 @@ func startMDS(context *clusterd.Context, config *Config) error { } name := fmt.Sprintf("mds%s", config.ID) - if err := context.ProcMan.Run(name, "ceph-mds", args...); err != nil { + if err := context.Executor.ExecuteCommand(false, name, "ceph-mds", args...); err != nil { return fmt.Errorf("failed to start mds. %+v", err) } return nil diff --git a/pkg/ceph/mgr/daemon.go b/pkg/ceph/mgr/daemon.go index 9914cc9fc..0b6953ed8 100644 --- a/pkg/ceph/mgr/daemon.go +++ b/pkg/ceph/mgr/daemon.go @@ -18,13 +18,11 @@ package mgr import ( "fmt" "path" - "regexp" "github.com/coreos/pkg/capnslog" "github.com/rook/rook/pkg/ceph/mon" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util" - "github.com/rook/rook/pkg/util/proc" ) var ( @@ -41,7 +39,6 @@ const ( ) type Config struct { - InProc bool ClusterInfo *mon.ClusterInfo Name string Keyring string @@ -49,17 +46,15 @@ type Config struct { func Run(context *clusterd.Context, config *Config) error { logger.Infof("Starting MGR %s with keyring %s", config.Name, config.Keyring) - err := generateConfigFiles(context, config) - if err != nil { + if err := generateConfigFiles(context, config); err != nil { return fmt.Errorf("failed to generate mgr config files. %+v", err) } - _, err = startMgr(context, config) - if err != nil { + if err := startMgr(context, config); err != nil { return fmt.Errorf("failed to run mgr. %+v", err) } - return err + return nil } func generateConfigFiles(context *clusterd.Context, config *Config) error { @@ -89,7 +84,7 @@ func generateConfigFiles(context *clusterd.Context, config *Config) error { return nil } -func startMgr(context *clusterd.Context, config *Config) (mgrProc *proc.MonitoredProc, err error) { +func startMgr(context *clusterd.Context, config *Config) error { // start the mgr daemon in the foreground with the given config logger.Infof("starting ceph-mgr") @@ -107,15 +102,10 @@ func startMgr(context *clusterd.Context, config *Config) (mgrProc *proc.Monitore "-i", config.Name, } - if config.InProc { - err = context.ProcMan.Run(cephmgr, cephmgr, args...) - } else { - mgrProc, err = context.ProcMan.Start(cephmgr, cephmgr, regexp.QuoteMeta(cephmgr), proc.ReuseExisting, args...) + if err := context.Executor.ExecuteCommand(false, cephmgr, cephmgr, args...); err != nil { + return fmt.Errorf("failed to start mgr: %+v", err) } - if err != nil { - err = fmt.Errorf("failed to start mgr: %+v", err) - } - return + return nil } func getMgrConfDir(dir, name string) string { diff --git a/pkg/ceph/mon/daemon.go b/pkg/ceph/mon/daemon.go index a53347a6d..0a38dd901 100644 --- a/pkg/ceph/mon/daemon.go +++ b/pkg/ceph/mon/daemon.go @@ -121,7 +121,8 @@ func startMon(context *clusterd.Context, config *Config, confFilePath, monDataDi monNameArg := fmt.Sprintf("--name=mon.%s", config.Name) keyringPath := getMonKeyringPath(context.ConfigDir, config.Name) - err = context.ProcMan.Run( + err = context.Executor.ExecuteCommand( + false, fmt.Sprintf("mkfs-%s", config.Name), "ceph-mon", "--mkfs", @@ -150,8 +151,7 @@ func startMon(context *clusterd.Context, config *Config, confFilePath, monDataDi fmt.Sprintf("--public-addr=%s:%d", context.NetworkInfo.PublicAddrIPv4, config.Port), fmt.Sprintf("--public-bind-addr=%s:%d", context.NetworkInfo.ClusterAddrIPv4, config.Port), } - err = context.ProcMan.Run(config.Name, "ceph-mon", args...) - if err != nil { + if err = context.Executor.ExecuteCommand(false, config.Name, "ceph-mon", args...); err != nil { return fmt.Errorf("failed to start mon: %+v", err) } diff --git a/pkg/ceph/osd/agent.go b/pkg/ceph/osd/agent.go index 1319123de..b08621f17 100644 --- a/pkg/ceph/osd/agent.go +++ b/pkg/ceph/osd/agent.go @@ -59,18 +59,20 @@ type OsdAgent struct { usingDeviceFilter bool metadataDevice string directories string + procMan *proc.ProcManager storeConfig StoreConfig kv kvstore.KeyValueStore configCounter int32 osdsCompleted chan struct{} } -func NewAgent(devices string, usingDeviceFilter bool, metadataDevice, directories string, forceFormat bool, +func NewAgent(context *clusterd.Context, devices string, usingDeviceFilter bool, metadataDevice, directories string, forceFormat bool, location string, storeConfig StoreConfig, cluster *mon.ClusterInfo, nodeName string, kv kvstore.KeyValueStore) *OsdAgent { return &OsdAgent{devices: devices, usingDeviceFilter: usingDeviceFilter, metadataDevice: metadataDevice, directories: directories, forceFormat: forceFormat, location: location, storeConfig: storeConfig, cluster: cluster, nodeName: nodeName, kv: kv, + procMan: proc.New(context.Executor), osdProc: make(map[int]*proc.MonitoredProc), } } @@ -405,7 +407,7 @@ func (a *OsdAgent) runOSD(context *clusterd.Context, clusterName string, config params = append(params, fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(config.rootPath))) } - process, err := context.ProcMan.Start( + process, err := a.procMan.Start( fmt.Sprintf("osd%d", config.id), "ceph-osd", regexp.QuoteMeta(osdUUIDArg), @@ -415,10 +417,6 @@ func (a *OsdAgent) runOSD(context *clusterd.Context, clusterName string, config return fmt.Errorf("failed to start osd %d: %+v", config.id, err) } - if a.osdProc == nil { - // initialize the osd map - a.osdProc = make(map[int]*proc.MonitoredProc) - } 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[config.id] = process diff --git a/pkg/ceph/osd/agent_test.go b/pkg/ceph/osd/agent_test.go index 3ae8a859e..dac6c6e3f 100644 --- a/pkg/ceph/osd/agent_test.go +++ b/pkg/ceph/osd/agent_test.go @@ -31,7 +31,6 @@ import ( "github.com/rook/rook/pkg/clusterd" exectest "github.com/rook/rook/pkg/util/exec/test" "github.com/rook/rook/pkg/util/kvstore" - "github.com/rook/rook/pkg/util/proc" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -170,7 +169,6 @@ func testOSDAgentWithDevicesHelper(t *testing.T, storeConfig StoreConfig) { context := &clusterd.Context{ Executor: executor, ConfigDir: configDir, - ProcMan: proc.New(executor), } // Set sdx as already having an assigned osd id, a UUID and saved to the partition scheme. @@ -211,10 +209,9 @@ func TestOSDAgentNoDevices(t *testing.T) { os.MkdirAll(filepath.Join(configDir, "osd3"), 0744) // create a test OSD agent with no devices specified - agent, _ := createTestAgent(t, "", configDir, nil) + agent, executor := createTestAgent(t, "", configDir, nil) startCount := 0 - executor := &exectest.MockExecutor{} executor.MockStartExecuteCommand = func(debug bool, name string, command string, args ...string) (*exec.Cmd, error) { logger.Infof("StartExecuteCommand: %s %v", command, args) startCount++ @@ -248,7 +245,6 @@ func TestOSDAgentNoDevices(t *testing.T) { context := &clusterd.Context{ Devices: []*clusterd.LocalDisk{}, Executor: executor, - ProcMan: proc.New(executor), ConfigDir: configDir, } dirs := map[string]int{ @@ -270,14 +266,14 @@ func createTestAgent(t *testing.T, devices, configDir string, storeConfig *Store if storeConfig == nil { storeConfig = &StoreConfig{StoreType: Bluestore} } - agent := NewAgent(devices, false, "", "", forceFormat, location, *storeConfig, nil, "myhost", kvstore.NewMockKeyValueStore()) - agent.cluster = &mon.ClusterInfo{Name: "myclust"} executor := &exectest.MockExecutor{ MockExecuteCommandWithOutputFile: func(debug bool, actionName string, command string, outFileArg string, args ...string) (string, error) { return "{\"key\":\"mysecurekey\", \"osdid\":3.0}", nil }, } + agent := NewAgent(&clusterd.Context{Executor: executor}, devices, false, "", "", forceFormat, location, *storeConfig, nil, "myhost", kvstore.NewMockKeyValueStore()) + agent.cluster = &mon.ClusterInfo{Name: "myclust"} return agent, executor } diff --git a/pkg/ceph/osd/daemon_test.go b/pkg/ceph/osd/daemon_test.go index 792a32257..797892fbf 100644 --- a/pkg/ceph/osd/daemon_test.go +++ b/pkg/ceph/osd/daemon_test.go @@ -26,7 +26,6 @@ import ( "github.com/rook/rook/pkg/clusterd" exectest "github.com/rook/rook/pkg/util/exec/test" "github.com/rook/rook/pkg/util/kvstore" - "github.com/rook/rook/pkg/util/proc" "github.com/stretchr/testify/assert" ) @@ -117,7 +116,7 @@ NAME="sdb1" SIZE="30" TYPE="part" PKNAME="sdb"`, nil return "", fmt.Errorf("unknown command %s %+v", command, args) } - context := &clusterd.Context{ProcMan: proc.New(executor), Executor: executor} + context := &clusterd.Context{Executor: executor} context.Devices = []*clusterd.LocalDisk{ {Name: "sda"}, {Name: "sdb"}, diff --git a/pkg/ceph/osd/filesystem.go b/pkg/ceph/osd/filesystem.go index 06549710e..00d41dc9b 100644 --- a/pkg/ceph/osd/filesystem.go +++ b/pkg/ceph/osd/filesystem.go @@ -69,13 +69,9 @@ func createOSDFileSystem(context *clusterd.Context, clusterName string, config * options = append(options, fmt.Sprintf("--osd-journal=%s", getOSDJournalPath(config.rootPath))) } - // create the OSD file system - err = context.ProcMan.Run( - fmt.Sprintf("mkfs-osd%d", config.id), - "ceph-osd", - options...) - - if err != nil { + // create the file system + logName := fmt.Sprintf("mkfs-osd%d", config.id) + if err = context.Executor.ExecuteCommand(false, logName, "ceph-osd", options...); err != nil { return fmt.Errorf("failed osd mkfs for OSD ID %d, UUID %s, dataDir %s: %+v", config.id, config.uuid.String(), config.rootPath, err) } diff --git a/pkg/ceph/rgw/daemon.go b/pkg/ceph/rgw/daemon.go index 2f5fb3995..5b69cd893 100644 --- a/pkg/ceph/rgw/daemon.go +++ b/pkg/ceph/rgw/daemon.go @@ -20,7 +20,6 @@ import ( "io/ioutil" "os" "path" - "regexp" "strconv" @@ -28,7 +27,6 @@ import ( "github.com/rook/rook/pkg/ceph/mon" "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/util" - "github.com/rook/rook/pkg/util/proc" ) var logger = capnslog.NewPackageLogger("github.com/rook/rook", "cephrgw") @@ -40,7 +38,6 @@ type Config struct { SecurePort int Keyring string CertificatePath string - InProc bool ClusterInfo *mon.ClusterInfo } @@ -51,7 +48,7 @@ func Run(context *clusterd.Context, config *Config) error { return fmt.Errorf("failed to generate rgw config files. %+v", err) } - _, err = startRGW(context, config) + err = startRGW(context, config) if err != nil { return fmt.Errorf("failed to run rgw. %+v", err) } @@ -123,7 +120,7 @@ func generateConfigFiles(context *clusterd.Context, config *Config) error { return nil } -func startRGW(context *clusterd.Context, config *Config) (rgwProc *proc.MonitoredProc, err error) { +func startRGW(context *clusterd.Context, config *Config) (err error) { // start the monitor daemon in the foreground with the given config logger.Infof("starting rgw") @@ -139,15 +136,10 @@ func startRGW(context *clusterd.Context, config *Config) (rgwProc *proc.Monitore fmt.Sprintf("--keyring=%s", getRGWKeyringPath(context.ConfigDir)), fmt.Sprintf("--rgw-mime-types-file=%s", getMimeTypesPath(context.ConfigDir)), } - if config.InProc { - err = context.ProcMan.Run("rgw", "radosgw", args...) - } else { - rgwProc, err = context.ProcMan.Start("rgw", "radosgw", regexp.QuoteMeta(rgwNameArg), proc.ReuseExisting, args...) + if err = context.Executor.ExecuteCommand(false, "rgw", "radosgw", args...); err != nil { + return fmt.Errorf("failed to start rgw: %+v", err) } - if err != nil { - err = fmt.Errorf("failed to start rgw: %+v", err) - } - return + return nil } func getRGWConfFilePath(configDir, clusterName string) string { diff --git a/pkg/clusterd/context.go b/pkg/clusterd/context.go index fef4e9a69..8e3d6c22e 100644 --- a/pkg/clusterd/context.go +++ b/pkg/clusterd/context.go @@ -18,7 +18,6 @@ package clusterd import ( "github.com/coreos/pkg/capnslog" "github.com/rook/rook/pkg/util/exec" - "github.com/rook/rook/pkg/util/proc" apiextensionsclient "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset" "k8s.io/client-go/kubernetes" ) @@ -34,9 +33,6 @@ type Context struct { // The implementation of executing a console command Executor exec.Executor - // The process manager for launching a process - ProcMan *proc.ProcManager - // The root configuration directory used by services ConfigDir string