Merge pull request #116 from UtkarshBhatthere/replication_network

Add Cluster config management to microceph

.github/workflows/tests.yml

@@ -130,6 +130,36 @@ jobs:
           s3cmd --host localhost --host-bucket="localhost/%(bucket)" --access_key=fooAccessKey --secret_key=fooSecretKey --no-ssl put -P ~/test.txt s3://testbucket
           curl -s http://localhost/testbucket/test.txt | grep -F hello-radosgw
 
+      - name: Test Cluster Config
+        run: |
+          set -eux
+          cip=$(ip -4 -j route | jq -r '.[] | select(.dst | contains("default")) | .prefsrc' | tr -d '[:space:]')
+          
+          # pre config set timestamp for service age
+          ts=$(sudo systemctl show --property ActiveEnterTimestampMonotonic snap.microceph.osd.service | cut -d= -f2)
+
+          # set config
+          sudo microceph cluster config set cluster_network $cip/8 --wait
+
+          # post config set timestamp for service age
+          ts2=$(sudo systemctl show --property ActiveEnterTimestampMonotonic snap.microceph.osd.service | cut -d= -f2)
+
+          # Check config output
+          output=$(sudo microceph cluster config get cluster_network | grep -cim1 'cluster_network')
+          if [[ $output -lt 1 ]] ; then echo "config check failed: $output"; exit 1; fi
+
+          # Check service restarted
+          if [ $ts2 -lt $ts ]; then echo "config check failed: TS1: $ts2 TS2: $ts3"; exit 1; fi
+
+          # reset config
+          sudo microceph cluster config reset cluster_network --wait
+
+          # post config reset timestamp for service age
+          ts3=$(sudo systemctl show --property ActiveEnterTimestampMonotonic snap.microceph.osd.service | cut -d= -f2)
+
+          # Check service restarted
+          if [ $ts3 -lt $ts2 ]; then echo "config check failed: TS2: $ts2 TS3: $ts3"; exit 1; fi
+
       - name: Upload artifacts
         if: always()
         uses: actions/upload-artifact@v3

microceph/api/configs.go

@@ -0,0 +1,89 @@
+package api
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/canonical/microcluster/rest"
+	"github.com/canonical/microcluster/state"
+	"github.com/lxc/lxd/lxd/response"
+
+	"github.com/canonical/microceph/microceph/api/types"
+	"github.com/canonical/microceph/microceph/ceph"
+	"github.com/canonical/microceph/microceph/client"
+)
+
+// /1.0/configs endpoint.
+var configsCmd = rest.Endpoint{
+	Path: "configs",
+
+	Get:  rest.EndpointAction{Handler: cmdConfigsGet, ProxyTarget: true},
+	Put: rest.EndpointAction{Handler: cmdConfigsPut, ProxyTarget: true},
+	Delete: rest.EndpointAction{Handler: cmdConfigsDelete, ProxyTarget: true},
+}
+
+func cmdConfigsGet(s *state.State, r *http.Request) response.Response {
+	var err error
+	var req types.Config
+	var configs types.Configs
+
+	err = json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		return response.InternalError(err)
+	}
+
+	// If a valid key string is passed, fetch that key.
+	if len(req.Key) > 0 {
+		configs, err = ceph.GetConfigItem(req)
+	} else {
+		// Fetch all configs.
+		configs, err = ceph.ListConfigs()
+	}
+	if err != nil {
+		return response.SmartError(err)
+	}
+
+	return response.SyncResponse(true, configs)
+}
+
+func cmdConfigsPut(s *state.State, r *http.Request) response.Response {
+	var req types.Config
+	configTable := ceph.GetConstConfigTable()
+
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		return response.InternalError(err)
+	}
+
+	// Configure the key/value
+	err = ceph.SetConfigItem(req)
+	if err != nil {
+		return response.SmartError(err)
+	}
+
+	services := configTable[req.Key].Daemons
+	client.ConfigChangeRefresh(s, services, req.Wait)
+
+	return response.EmptySyncResponse
+}
+
+func cmdConfigsDelete(s *state.State, r *http.Request) response.Response {
+	var req types.Config
+	configTable := ceph.GetConstConfigTable()
+
+	err := json.NewDecoder(r.Body).Decode(&req)
+	if err != nil {
+		return response.InternalError(err)
+	}
+
+	// Clean the key/value
+	err = ceph.RemoveConfigItem(req)
+	if err != nil {
+		return response.SmartError(err)
+	}
+
+	services := configTable[req.Key].Daemons
+	client.ConfigChangeRefresh(s, services, req.Wait)
+
+	return response.EmptySyncResponse
+}

microceph/api/endpoints.go

@@ -11,4 +11,6 @@ var Endpoints = []rest.Endpoint{
 	resourcesCmd,
 	servicesCmd,
 	rgwServiceCmd,
+	configsCmd,
+	restartServiceCmd,
 }

microceph/api/services.go

@@ -2,13 +2,16 @@ package api
 
 import (
 	"encoding/json"
+	"fmt"
+	"net/http"
+
 	"github.com/canonical/microceph/microceph/api/types"
 	"github.com/canonical/microceph/microceph/common"
-	"net/http"
 
 	"github.com/canonical/microcluster/rest"
 	"github.com/canonical/microcluster/state"
 	"github.com/lxc/lxd/lxd/response"
+	"github.com/lxc/lxd/shared/logger"
 
 	"github.com/canonical/microceph/microceph/ceph"
 )
@@ -29,6 +32,43 @@ func cmdServicesGet(s *state.State, r *http.Request) response.Response {
 	return response.SyncResponse(true, services)
 }
 
+// Service Reload Endpoint.
+var restartServiceCmd = rest.Endpoint{
+	Path: "services/restart",
+	Post: rest.EndpointAction{Handler: cmdRestartServicePost, ProxyTarget: true},
+}
+
+func cmdRestartServicePost(s *state.State, r *http.Request) response.Response {
+	var services types.Services
+
+	err := json.NewDecoder(r.Body).Decode(&services)
+	if err != nil {
+		logger.Errorf("Failed decoding restart services: %v", err)
+		return response.InternalError(err)
+	}
+
+	// Check if provided services are valid and available in microceph
+	for _, service := range(services) {
+		valid_services := ceph.GetConfigTableServiceSet()
+		if _, ok := valid_services[service.Service]; !ok {
+			err := fmt.Errorf("%s is not a valid ceph service", service.Service)
+			logger.Errorf("%v", err)
+			return response.InternalError(err)
+		}
+	}
+
+	for _, service := range services {
+		err = ceph.RestartCephService(service.Service)
+		if err != nil {
+			url := s.Address().String()
+			logger.Errorf("Failed restarting %s on host %s", service.Service, url)
+			return response.SyncResponse(false, err)
+		}
+	}
+
+	return response.EmptySyncResponse
+}
+
 var rgwServiceCmd = rest.Endpoint{
 	Path: "services/rgw",
 

microceph/api/types/configs.go

@@ -0,0 +1,12 @@
+// Package types provides shared types and structs.
+package types
+
+// Configs holds the key value pair
+type Config struct {
+	Key   string  `json:"key" yaml:"key"`
+	Value string  `json:"value" yaml:"value"`
+	Wait  bool    `json:"wait" yaml:"wait"`
+}
+
+// Configs is a slice of configs
+type Configs []Config

microceph/ceph/config.go

@@ -3,15 +3,164 @@ package ceph
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 
+	"github.com/canonical/microceph/microceph/api/types"
 	"github.com/canonical/microceph/microceph/common"
 	"github.com/canonical/microceph/microceph/database"
 )
 
+// Config Table is the source of additional information for each supported config key
+// Refer to GetConfigTable()
+type ConfigTable map[string]struct{
+	Who     string // Ceph Config internal <who> against each key
+	Daemons []string // List of Daemons that need to be restarted across the cluster for the config change to take effect.
+}
+
+// Check if certain key is present in the map.
+func (c ConfigTable) isKeyPresent(key string) bool {
+	if _, ok := c[key]; !ok {
+		return false
+	}
+
+	return true
+}
+
+// Return keys of the given set
+func (c ConfigTable) Keys() (keys []string) {
+	for k := range c {
+		keys = append(keys, k)
+	}
+	return keys
+}
+
+// Since we can't have const maps, we encapsulate the map into a func
+// so that each request for the map gaurantees consistent definition.
+func GetConstConfigTable() ConfigTable {
+	return ConfigTable{
+		"public_network":  {"global", []string{"mon", "osd"}},
+		"cluster_network": {"global", []string{"osd"}},
+	}
+}
+
+func GetConfigTableServiceSet() Set {
+	return Set{
+		"mon": struct{}{},
+		"mgr": struct{}{},
+		"osd": struct{}{},
+		"mds": struct{}{},
+		"rgw": struct{}{},
+	}
+}
+
+// Struct to get Config Items from config dump json output.
+type ConfigDumpItem struct{
+	Section string
+	Name    string
+	Value   string
+}
+type ConfigDump []ConfigDumpItem
+
+func SetConfigItem(c types.Config) error {
+	configTable := GetConstConfigTable()
+
+	args := []string{
+		"config",
+		"set",
+		configTable[c.Key].Who,
+		c.Key,
+		c.Value,
+		"-f",
+		"json-pretty",
+	}
+
+	_, err := processExec.RunCommand("ceph", args...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func GetConfigItem(c types.Config) (types.Configs, error) {
+	var err error
+	configTable := GetConstConfigTable()
+	ret := make(types.Configs, 1)
+	who := "mon"
+
+	// workaround to query global configs from mon entity
+	if configTable[c.Key].Who != "global" {
+		who = configTable[c.Key].Who
+	}
+
+	args := []string{
+		"config",
+		"get",
+		who,
+		c.Key,
+	}
+
+	ret[0].Key = c.Key
+	ret[0].Value, err = processExec.RunCommand("ceph", args...)
+	if err != nil {
+		return nil, err
+	}
+
+	return ret, nil
+}
+
+func RemoveConfigItem(c types.Config) error {
+	configTable := GetConstConfigTable()
+	args := []string{
+		"config",
+		"rm",
+		configTable[c.Key].Who,
+		c.Key,
+	}
+
+	_, err := processExec.RunCommand("ceph", args...)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func ListConfigs() (types.Configs, error) {
+	var dump ConfigDump
+	var configs types.Configs
+	configTable := GetConstConfigTable()
+	args := []string{
+		"config",
+		"dump",
+		"-f",
+		"json-pretty",
+	}
+
+	output, err := processExec.RunCommand("ceph", args...)
+	if err != nil {
+		return configs, err
+	}
+
+	json.Unmarshal([]byte(output), &dump)
+	// Only take configs permitted in config table.
+	for _, configItem := range dump {
+		if configTable.isKeyPresent(configItem.Name) {
+			configs = append(configs, types.Config{
+				Key: configItem.Name,
+				Value:  configItem.Value,
+			})
+		}
+	}
+
+	return configs, nil
+}
+
+// updates the ceph config file.
 func updateConfig(s common.StateInterface) error {
 	confPath := filepath.Join(os.Getenv("SNAP_DATA"), "conf")
 	runPath := filepath.Join(os.Getenv("SNAP_DATA"), "run")