Merge pull request #147 from UtkarshBhatthere/customServicePlacement

Added Service Placement Interface and Intf Implementation

.github/workflows/tests.yml

@@ -160,6 +160,8 @@ jobs:
     - name: Exercise RGW
       run: |
         set -eux
+        sudo microceph.ceph status
+        sudo systemctl status snap.microceph.rgw
         sudo microceph.radosgw-admin user create --uid=test --display-name=test
         sudo microceph.radosgw-admin key create --uid=test --key-type=s3 --access-key fooAccessKey --secret-key fooSecretKey
         sudo apt-get -qq install s3cmd

microceph/api/endpoints.go

@@ -13,4 +13,8 @@ var Endpoints = []rest.Endpoint{
 	rgwServiceCmd,
 	configsCmd,
 	restartServiceCmd,
+	mdsServiceCmd,
+	mgrServiceCmd,
+	monServiceCmd,
+	rgwServiceCmd,
 }

microceph/api/services.go

@@ -32,6 +32,45 @@ func cmdServicesGet(s *state.State, r *http.Request) response.Response {
 	return response.SyncResponse(true, services)
 }
 
+// Service Enable Endpoint.
+var monServiceCmd = rest.Endpoint{
+	Path: "services/mon",
+	Put:  rest.EndpointAction{Handler: cmdEnableServicePut, ProxyTarget: true},
+}
+
+var mgrServiceCmd = rest.Endpoint{
+	Path: "services/mgr",
+	Put:  rest.EndpointAction{Handler: cmdEnableServicePut, ProxyTarget: true},
+}
+
+var mdsServiceCmd = rest.Endpoint{
+	Path: "services/mds",
+	Put:  rest.EndpointAction{Handler: cmdEnableServicePut, ProxyTarget: true},
+}
+
+var rgwServiceCmd = rest.Endpoint{
+	Path:   "services/rgw",
+	Put:    rest.EndpointAction{Handler: cmdEnableServicePut, ProxyTarget: true},
+	Delete: rest.EndpointAction{Handler: cmdRGWServiceDelete, ProxyTarget: true},
+}
+
+func cmdEnableServicePut(s *state.State, r *http.Request) response.Response {
+	var payload types.EnableService
+
+	err := json.NewDecoder(r.Body).Decode(&payload)
+	if err != nil {
+		logger.Errorf("Failed decoding enable service request: %v", err)
+		return response.InternalError(err)
+	}
+
+	err = ceph.ServicePlacementHandler(common.CephState{State: s}, payload)
+	if err != nil {
+		return response.SyncResponse(false, err)
+	}
+
+	return response.SyncResponse(true, nil)
+}
+
 // Service Reload Endpoint.
 var restartServiceCmd = rest.Endpoint{
 	Path: "services/restart",
@@ -69,26 +108,15 @@ func cmdRestartServicePost(s *state.State, r *http.Request) response.Response {
 	return response.EmptySyncResponse
 }
 
-var rgwServiceCmd = rest.Endpoint{
-	Path: "services/rgw",
-
-	Put: rest.EndpointAction{Handler: cmdRGWServicePut, ProxyTarget: true},
-}
-
-// cmdRGWServicePutRGW is the handler for PUT /1.0/services/rgw.
-func cmdRGWServicePut(s *state.State, r *http.Request) response.Response {
+func cmdRGWServiceDelete(s *state.State, r *http.Request) response.Response {
 	var req types.RGWService
 
 	err := json.NewDecoder(r.Body).Decode(&req)
 	if err != nil {
 		return response.InternalError(err)
 	}
 
-	if req.Enabled {
-		err = ceph.EnableRGW(common.CephState{State: s}, req.Port)
-	} else {
-		err = ceph.DisableRGW(common.CephState{State: s})
-	}
+	err = ceph.DisableRGW(common.CephState{State: s})
 	if err != nil {
 		return response.SmartError(err)
 	}

microceph/api/types/services.go

@@ -10,6 +10,16 @@ type Service struct {
 	Location string `json:"location" yaml:"location"`
 }
 
+// Name: Name of the service to be enabled
+// Wait: Whether the operation is to be performed in sync or async
+// Payload: Service specific additional data encoded as a json string.
+type EnableService struct {
+	Name    string `json:"name" yaml:"name"`
+	Wait    bool   `json:"bool" yaml:"bool"`
+	Payload string `json:"payload" yaml:"payload"`
+	// Enable Service passes all additional data as a json payload string.
+}
+
 // RGWService holds a port number and enable/disable flag
 type RGWService struct {
 	Service

microceph/ceph/bootstrap.go

@@ -16,21 +16,11 @@ import (
 
 // Bootstrap will initialize a new Ceph deployment.
 func Bootstrap(s common.StateInterface) error {
-
-	confPath := filepath.Join(os.Getenv("SNAP_DATA"), "conf")
-	runPath := filepath.Join(os.Getenv("SNAP_DATA"), "run")
-	dataPath := filepath.Join(os.Getenv("SNAP_COMMON"), "data")
-	logPath := filepath.Join(os.Getenv("SNAP_COMMON"), "logs")
+	pathConsts := common.GetPathConst()
+	pathFileMode := common.GetPathFileMode()
 
 	// Create our various paths.
-	paths := map[string]os.FileMode{
-		confPath: 0755,
-		runPath:  0700,
-		dataPath: 0700,
-		logPath:  0700,
-	}
-
-	for path, perm := range paths {
+	for path, perm := range pathFileMode {
 		err := os.MkdirAll(path, perm)
 		if err != nil {
 			return fmt.Errorf("Unable to create %q: %w", path, err)
@@ -39,12 +29,11 @@ func Bootstrap(s common.StateInterface) error {
 
 	// Generate a new FSID.
 	fsid := uuid.NewRandom().String()
-
-	conf := newCephConfig(confPath)
+	conf := newCephConfig(pathConsts.ConfPath)
 	err := conf.WriteConfig(
 		map[string]any{
 			"fsid":     fsid,
-			"runDir":   runPath,
+			"runDir":   pathConsts.RunPath,
 			"monitors": s.ClusterState().Address().Hostname(),
 			"addr":     s.ClusterState().Address().Hostname(),
 		},
@@ -53,14 +42,14 @@ func Bootstrap(s common.StateInterface) error {
 		return err
 	}
 
-	path, err := createKeyrings(confPath)
+	path, err := createKeyrings(pathConsts.ConfPath)
 	if err != nil {
 		return err
 	}
 
 	defer os.RemoveAll(path)
 
-	adminKey, err := parseKeyring(filepath.Join(confPath, "ceph.client.admin.keyring"))
+	adminKey, err := parseKeyring(filepath.Join(pathConsts.ConfPath, "ceph.client.admin.keyring"))
 	if err != nil {
 		return fmt.Errorf("Failed parsing admin keyring: %w", err)
 	}
@@ -70,17 +59,17 @@ func Bootstrap(s common.StateInterface) error {
 		return err
 	}
 
-	err = initMon(s, dataPath, path)
+	err = initMon(s, pathConsts.DataPath, path)
 	if err != nil {
 		return err
 	}
 
-	err = initMgr(s, dataPath)
+	err = initMgr(s, pathConsts.DataPath)
 	if err != nil {
 		return err
 	}
 
-	err = initMds(s, dataPath)
+	err = initMds(s, pathConsts.DataPath)
 	if err != nil {
 		return err
 	}
@@ -90,7 +79,7 @@ func Bootstrap(s common.StateInterface) error {
 		return err
 	}
 
-	err = startOSDs(s, dataPath)
+	err = startOSDs(s, pathConsts.DataPath)
 	if err != nil {
 		return err
 	}

microceph/ceph/join.go

@@ -5,28 +5,19 @@ import (
 	"database/sql"
 	"fmt"
 	"os"
-	"path/filepath"
 
+	"github.com/canonical/lxd/shared/logger"
 	"github.com/canonical/microceph/microceph/common"
 	"github.com/canonical/microceph/microceph/database"
 )
 
 // Join will join an existing Ceph deployment.
 func Join(s common.StateInterface) error {
-	confPath := filepath.Join(os.Getenv("SNAP_DATA"), "conf")
-	runPath := filepath.Join(os.Getenv("SNAP_DATA"), "run")
-	dataPath := filepath.Join(os.Getenv("SNAP_COMMON"), "data")
-	logPath := filepath.Join(os.Getenv("SNAP_COMMON"), "logs")
+	pathFileMode := common.GetPathFileMode()
+	var spt = GetServicePlacementTable()
 
 	// Create our various paths.
-	paths := map[string]os.FileMode{
-		confPath: 0755,
-		runPath:  0700,
-		dataPath: 0700,
-		logPath:  0700,
-	}
-
-	for path, perm := range paths {
+	for path, perm := range pathFileMode {
 		err := os.MkdirAll(path, perm)
 		if err != nil {
 			return fmt.Errorf("Unable to create %q: %w", path, err)
@@ -82,63 +73,30 @@ func Join(s common.StateInterface) error {
 	services := []string{}
 
 	if srvMon < 3 {
-		monDataPath := filepath.Join(dataPath, "mon", fmt.Sprintf("ceph-%s", s.ClusterState().Name()))
-
-		err = os.MkdirAll(monDataPath, 0700)
+		err := spt["mon"].ServiceInit(s)
 		if err != nil {
-			return fmt.Errorf("Failed to join monitor: %w", err)
-		}
-
-		err = joinMon(s.ClusterState().Name(), monDataPath)
-		if err != nil {
-			return fmt.Errorf("Failed to join monitor: %w", err)
-		}
-
-		err = snapStart("mon", true)
-		if err != nil {
-			return fmt.Errorf("Failed to start monitor: %w", err)
+			logger.Errorf("%v", err)
+			return err
 		}
 
 		services = append(services, "mon")
 	}
 
 	if srvMgr < 3 {
-		mgrDataPath := filepath.Join(dataPath, "mgr", fmt.Sprintf("ceph-%s", s.ClusterState().Name()))
-
-		err = os.MkdirAll(mgrDataPath, 0700)
-		if err != nil {
-			return fmt.Errorf("Failed to join manager: %w", err)
-		}
-
-		err = joinMgr(s.ClusterState().Name(), mgrDataPath)
-		if err != nil {
-			return fmt.Errorf("Failed to join manager: %w", err)
-		}
-
-		err = snapStart("mgr", true)
+		err := spt["mgr"].ServiceInit(s)
 		if err != nil {
-			return fmt.Errorf("Failed to start manager: %w", err)
+			logger.Errorf("%v", err)
+			return err
 		}
 
 		services = append(services, "mgr")
 	}
 
 	if srvMds < 3 {
-		mdsDataPath := filepath.Join(dataPath, "mds", fmt.Sprintf("ceph-%s", s.ClusterState().Name()))
-
-		err = os.MkdirAll(mdsDataPath, 0700)
+		err := spt["mds"].ServiceInit(s)
 		if err != nil {
-			return fmt.Errorf("Failed to join metadata server: %w", err)
-		}
-
-		err = joinMds(s.ClusterState().Name(), mdsDataPath)
-		if err != nil {
-			return fmt.Errorf("Failed to join metadata server: %w", err)
-		}
-
-		err = snapStart("mds", true)
-		if err != nil {
-			return fmt.Errorf("Failed to start metadata server: %w", err)
+			logger.Errorf("%v", err)
+			return err
 		}
 
 		services = append(services, "mds")

microceph/ceph/monitor.go

@@ -57,20 +57,20 @@ func bootstrapMon(hostname string, path string, monmap string, keyring string) e
 func joinMon(hostname string, path string) error {
 	tmpPath, err := os.MkdirTemp("", "")
 	if err != nil {
-		return fmt.Errorf("Unable to create temporary path: %w", err)
+		return fmt.Errorf("unable to create temporary path: %w", err)
 	}
 	defer os.RemoveAll(tmpPath)
 
 	monmap := filepath.Join(tmpPath, "mon.map")
 	_, err = cephRun("mon", "getmap", "-o", monmap)
 	if err != nil {
-		return fmt.Errorf("Failed to retrieve monmap: %w", err)
+		return fmt.Errorf("failed to retrieve monmap: %w", err)
 	}
 
 	keyring := filepath.Join(tmpPath, "mon.keyring")
 	_, err = cephRun("auth", "get", "mon.", "-o", keyring)
 	if err != nil {
-		return fmt.Errorf("Failed to retrieve mon keyring: %w", err)
+		return fmt.Errorf("failed to retrieve mon keyring: %w", err)
 	}
 
 	return bootstrapMon(hostname, path, monmap, keyring)