diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 6ce261fa9..d54c6ed26 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -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()) diff --git a/pkg/operator/ceph/cluster/cluster_test.go b/pkg/operator/ceph/cluster/cluster_test.go index a53a29014..7d5ed9c8e 100644 --- a/pkg/operator/ceph/cluster/cluster_test.go +++ b/pkg/operator/ceph/cluster/cluster_test.go @@ -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) }) } }