Merge pull request #1143 from travisn/refactor-proc

Refactor proc manager for only the osd daemon

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,

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)

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)

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)

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

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 {

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)
 	}
 

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

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
 }

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"},

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)
 	}

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 {

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