Merge pull request #10934 from subhamkrai/assimilate-conf
core: use assimilate-conf command to run commands
This commit is contained in:
2 files changed
+102
-44
No files matched your search
@@ -20,6 +20,8 @@ package cluster
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path"
|
||||
"strconv"
|
||||
@@ -27,7 +29,9 @@ import (
|
||||
"syscall"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"gopkg.in/ini.v1"
|
||||
|
||||
"github.com/rook/rook/cmd/rook/rook"
|
||||
cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
|
||||
"github.com/rook/rook/pkg/clusterd"
|
||||
"github.com/rook/rook/pkg/daemon/ceph/client"
|
||||
@@ -42,6 +46,7 @@ import (
|
||||
"github.com/rook/rook/pkg/operator/ceph/csi"
|
||||
cephver "github.com/rook/rook/pkg/operator/ceph/version"
|
||||
"github.com/rook/rook/pkg/operator/k8sutil"
|
||||
pkgexec "github.com/rook/rook/pkg/util/exec"
|
||||
rookversion "github.com/rook/rook/pkg/version"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
kerrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
@@ -595,28 +600,89 @@ func (c *cluster) reportTelemetry() {
|
||||
telemetry.ReportKeyValue(c.context, c.ClusterInfo, telemetry.ExternalModeEnabledKey, strconv.FormatBool(c.Spec.External.Enable))
|
||||
}
|
||||
|
||||
func (c *cluster) addSettingsToMonDB(clientName string, settings map[string]string) error {
|
||||
|
||||
assimilateConfPath, err := ioutil.TempFile("", "")
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "failed to create assimilateConf temp dir for %s.", clientName)
|
||||
}
|
||||
|
||||
err = ioutil.WriteFile(assimilateConfPath.Name(), []byte(""), 0600)
|
||||
if err != nil {
|
||||
rook.TerminateFatal(errors.Wrapf(err, "failed to write config file"))
|
||||
}
|
||||
|
||||
defer func() {
|
||||
err := os.Remove(assimilateConfPath.Name())
|
||||
if err != nil {
|
||||
logger.Errorf("failed to remove file %q. %v", assimilateConfPath.Name(), err)
|
||||
}
|
||||
}()
|
||||
|
||||
configFile := ini.Empty()
|
||||
s, err := configFile.NewSection(clientName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for key, val := range settings {
|
||||
if _, err := s.NewKey(key, val); err != nil {
|
||||
return errors.Wrapf(err, "failed to add key %s", key)
|
||||
}
|
||||
}
|
||||
|
||||
if err := configFile.SaveTo(assimilateConfPath.Name()); err != nil {
|
||||
return errors.Wrapf(err, "failed to save config file %s", assimilateConfPath.Name())
|
||||
}
|
||||
|
||||
fileContent, err := ioutil.ReadFile(assimilateConfPath.Name())
|
||||
if err != nil {
|
||||
logger.Errorf("failed to open assimilate input file %s. %c", assimilateConfPath.Name(), err)
|
||||
}
|
||||
logger.Infof("applying ceph settings:\n%s", string(fileContent))
|
||||
|
||||
args := []string{"config", "assimilate-conf", "-i", assimilateConfPath.Name(), "-o", assimilateConfPath.Name() + ".out"}
|
||||
cephCmd := client.NewCephCommand(c.context, c.ClusterInfo, args)
|
||||
|
||||
out, err := cephCmd.RunWithTimeout(pkgexec.CephCommandsTimeout)
|
||||
if err != nil {
|
||||
logger.Errorf("failed to run command ceph %s", args)
|
||||
|
||||
fileContent, err := ioutil.ReadFile(assimilateConfPath.Name() + ".out")
|
||||
if err != nil {
|
||||
logger.Errorf("failed to open assimilate output file %s.out. %v", assimilateConfPath.Name(), err)
|
||||
}
|
||||
logger.Infof("failed to apply ceph settings:\n%s", string(fileContent))
|
||||
|
||||
return errors.Wrapf(err, "failed to set ceph config in the centralized mon configuration database; "+
|
||||
"output: %s", string(out))
|
||||
}
|
||||
|
||||
logger.Info("successfully applied settings to the mon configuration database")
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *cluster) configureMsgr2() error {
|
||||
if c.Spec.Network.Connections == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Set network encryption
|
||||
monStore := config.GetMonStore(c.context, c.ClusterInfo)
|
||||
if c.Spec.Network.Connections.Encryption != nil {
|
||||
encryptionSetting := "crc secure"
|
||||
if c.Spec.Network.Connections.Encryption.Enabled {
|
||||
encryptionSetting = "secure"
|
||||
}
|
||||
logger.Infof("setting msgr2 encryption mode to %q", encryptionSetting)
|
||||
|
||||
if err := monStore.Set("global", "ms_cluster_mode", encryptionSetting); err != nil {
|
||||
return errors.Wrapf(err, "failed to set ms_cluster_mode to %q", encryptionSetting)
|
||||
globalConfigSettings := map[string]string{
|
||||
"ms_cluster_mode": encryptionSetting,
|
||||
"ms_service_mode": encryptionSetting,
|
||||
"ms_client_mode": encryptionSetting,
|
||||
}
|
||||
if err := monStore.Set("global", "ms_service_mode", encryptionSetting); err != nil {
|
||||
return errors.Wrapf(err, "failed to set ms_service_mode to %q", encryptionSetting)
|
||||
}
|
||||
if err := monStore.Set("global", "ms_client_mode", encryptionSetting); err != nil {
|
||||
return errors.Wrapf(err, "failed to set ms_client_mode to %q", encryptionSetting)
|
||||
|
||||
logger.Infof("setting msgr2 encryption mode to %q", encryptionSetting)
|
||||
if err := c.addSettingsToMonDB("global", globalConfigSettings); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -627,9 +693,12 @@ func (c *cluster) configureMsgr2() error {
|
||||
if c.Spec.Network.Connections.Compression.Enabled {
|
||||
compressionSetting = "force"
|
||||
}
|
||||
globalConfigSettings := map[string]string{
|
||||
"ms_osd_compress_mode": compressionSetting,
|
||||
}
|
||||
logger.Infof("setting msgr2 compression mode to %q", compressionSetting)
|
||||
if err := monStore.Set("global", "ms_osd_compress_mode", compressionSetting); err != nil {
|
||||
return errors.Wrapf(err, "failed to set ms_osd_compress_mode to %q", compressionSetting)
|
||||
if err := c.addSettingsToMonDB("global", globalConfigSettings); err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
logger.Warningf("network compression requires Ceph Quincy (v17) or newer, skipping for current ceph %q", c.ClusterInfo.CephVersion.String())
|
||||
|
||||
@@ -153,68 +153,57 @@ func TestPreMonChecks(t *testing.T) {
|
||||
|
||||
func TestConfigureMsgr2(t *testing.T) {
|
||||
type fields struct {
|
||||
encryptionExpected bool
|
||||
compressionExpected bool
|
||||
disabledBothExpected bool
|
||||
cephVersion cephver.CephVersion
|
||||
Spec *cephv1.ClusterSpec
|
||||
encryptionExpected bool
|
||||
compressionExpected bool
|
||||
cephVersion cephver.CephVersion
|
||||
Spec *cephv1.ClusterSpec
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
}{
|
||||
{"default settings", fields{false, false, false, cephver.CephVersion{Major: 15}, &cephv1.ClusterSpec{}}},
|
||||
{"encryption enabled", fields{true, false, false, cephver.CephVersion{Major: 16}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Encryption: &cephv1.EncryptionSpec{Enabled: true}}}}}},
|
||||
{"compression enabled old version", fields{false, false, false, cephver.CephVersion{Major: 16}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Compression: &cephv1.CompressionSpec{Enabled: true}}}}}},
|
||||
{"compression enabled good version", fields{false, true, false, cephver.CephVersion{Major: 17}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Compression: &cephv1.CompressionSpec{Enabled: true}}}}}},
|
||||
{"both enabled", fields{true, true, false, cephver.CephVersion{Major: 17}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Encryption: &cephv1.EncryptionSpec{Enabled: true}, Compression: &cephv1.CompressionSpec{Enabled: true}}}}}},
|
||||
{"both disabled", fields{false, false, true, cephver.CephVersion{Major: 17}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Encryption: &cephv1.EncryptionSpec{Enabled: false}, Compression: &cephv1.CompressionSpec{Enabled: false}}}}}},
|
||||
{"default settings", fields{false, false, cephver.CephVersion{Major: 15}, &cephv1.ClusterSpec{}}},
|
||||
{"encryption enabled", fields{true, false, cephver.CephVersion{Major: 16}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Encryption: &cephv1.EncryptionSpec{Enabled: true}}}}}},
|
||||
{"compression enabled old version", fields{false, false, cephver.CephVersion{Major: 16}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Compression: &cephv1.CompressionSpec{Enabled: true}}}}}},
|
||||
{"compression enabled good version", fields{false, true, cephver.CephVersion{Major: 17}, &cephv1.ClusterSpec{Network: cephv1.NetworkSpec{Connections: &cephv1.ConnectionsSpec{Compression: &cephv1.CompressionSpec{Enabled: true}}}}}},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
enabledCompression := false
|
||||
disabledCompression := false
|
||||
enabledEncryption := false
|
||||
disabledEncryption := false
|
||||
executor := &exectest.MockExecutor{
|
||||
MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
|
||||
logger.Infof("Execute: %s %v", command, args)
|
||||
if args[0] == "config" && args[1] == "set" {
|
||||
if args[3] == "ms_osd_compress_mode" {
|
||||
if args[4] == "force" {
|
||||
enabledCompression = true
|
||||
} else if args[4] == "none" {
|
||||
disabledCompression = true
|
||||
}
|
||||
if args[0] == "config" && args[1] == "assimilate-conf" && args[2] == "-i" && args[4] == "-o" {
|
||||
if tt.fields.Spec.Network.Connections.Encryption != nil && tt.fields.Spec.Network.Connections.Compression == nil {
|
||||
enabledEncryption = true
|
||||
} else if tt.fields.Spec.Network.Connections.Encryption == nil && tt.fields.Spec.Network.Connections.Compression != nil {
|
||||
enabledCompression = true
|
||||
}
|
||||
if args[3] == "ms_cluster_mode" {
|
||||
if args[4] == "secure" {
|
||||
enabledEncryption = true
|
||||
} else if args[4] == "crc secure" {
|
||||
disabledEncryption = true
|
||||
}
|
||||
}
|
||||
|
||||
return "", nil
|
||||
}
|
||||
return "", errors.Errorf("unrecognized command")
|
||||
|
||||
return "", errors.Errorf("unexpected ceph command %q", args)
|
||||
},
|
||||
}
|
||||
clusterInfo := cephclient.AdminTestClusterInfo("rook-ceph")
|
||||
clusterInfo.CephVersion = tt.fields.cephVersion
|
||||
context := &clusterd.Context{Clientset: testop.New(t, 3), Executor: executor}
|
||||
context := &clusterd.Context{
|
||||
Clientset: testop.New(t, 3),
|
||||
Executor: executor,
|
||||
}
|
||||
c := &cluster{
|
||||
ClusterInfo: clusterInfo,
|
||||
Spec: tt.fields.Spec,
|
||||
context: context,
|
||||
Namespace: "rook-ceph",
|
||||
}
|
||||
|
||||
err := c.configureMsgr2()
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, tt.fields.compressionExpected, enabledCompression)
|
||||
assert.Equal(t, tt.fields.encryptionExpected, enabledEncryption)
|
||||
assert.Equal(t, tt.fields.disabledBothExpected, disabledCompression)
|
||||
assert.Equal(t, tt.fields.disabledBothExpected, disabledEncryption)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user