mon: run ceph commands to mon with timeout (backport #8939)

pkg/operator/ceph/cluster/cephstatus_test.go

@@ -164,7 +164,11 @@ func TestConfigureHealthSettings(t *testing.T) {
 	getGlobalIDReclaim := false
 	setGlobalIDReclaim := false
 	c.context.Executor = &exectest.MockExecutor{
+<<<<<<< HEAD
 		MockExecuteCommandWithOutputFile: func(command, outfile string, args ...string) (string, error) {
+=======
+		MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			logger.Infof("Command: %s %v", command, args)
 			if args[0] == "config" && args[3] == "auth_allow_insecure_global_id_reclaim" {
 				if args[1] == "get" {

pkg/operator/ceph/cluster/mgr/mgr_test.go

@@ -22,6 +22,7 @@ import (
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	rookv1 "github.com/rook/rook/pkg/apis/rook.io/v1"
@@ -257,12 +258,15 @@ func TestConfigureModules(t *testing.T) {
 					}
 					lastModuleConfigured = args[3]
 				}
-				if args[0] == "config" && args[1] == "set" && args[2] == "global" {
-					configSettings[args[3]] = args[4]
-				}
 			}
 			return "", nil //return "{\"key\":\"mysecurekey\"}", nil
 		},
+		MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
+			if args[0] == "config" && args[1] == "set" && args[2] == "global" {
+				configSettings[args[3]] = args[4]
+			}
+			return "", nil
+		},
 	}
 
 	clientset := testop.New(t, 3)

pkg/operator/ceph/config/monstore.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/util/exec"
 )
 
 // MonStore provides methods for setting Ceph configurations in the centralized mon
@@ -58,7 +59,7 @@ func (m *MonStore) Set(who, option, value string) error {
 	logger.Infof("setting %q=%q=%q option to the mon configuration database", who, option, value)
 	args := []string{"config", "set", who, normalizeKey(option), value}
 	cephCmd := client.NewCephCommand(m.context, m.clusterInfo, args)
-	out, err := cephCmd.Run()
+	out, err := cephCmd.RunWithTimeout(exec.CephCommandsTimeout)
 	if err != nil {
 		return errors.Wrapf(err, "failed to set ceph config in the centralized mon configuration database; "+
 			"you may need to use the rook-config-override ConfigMap. output: %s", string(out))
@@ -73,7 +74,7 @@ func (m *MonStore) Delete(who, option string) error {
 	logger.Infof("deleting %q option from the mon configuration database", option)
 	args := []string{"config", "rm", who, normalizeKey(option)}
 	cephCmd := client.NewCephCommand(m.context, m.clusterInfo, args)
-	out, err := cephCmd.Run()
+	out, err := cephCmd.RunWithTimeout(exec.CephCommandsTimeout)
 	if err != nil {
 		return errors.Wrapf(err, "failed to delete ceph config in the centralized mon configuration database. output: %s",
 			string(out))
@@ -88,7 +89,7 @@ func (m *MonStore) Delete(who, option string) error {
 func (m *MonStore) Get(who, option string) (string, error) {
 	args := []string{"config", "get", who, normalizeKey(option)}
 	cephCmd := client.NewCephCommand(m.context, m.clusterInfo, args)
-	out, err := cephCmd.Run()
+	out, err := cephCmd.RunWithTimeout(exec.CephCommandsTimeout)
 	if err != nil {
 		return "", errors.Wrapf(err, "failed to get config setting %q for user %q", option, who)
 	}
@@ -99,7 +100,7 @@ func (m *MonStore) Get(who, option string) (string, error) {
 func (m *MonStore) GetDaemon(who string) ([]Option, error) {
 	args := []string{"config", "get", who}
 	cephCmd := client.NewCephCommand(m.context, m.clusterInfo, args)
-	out, err := cephCmd.Run()
+	out, err := cephCmd.RunWithTimeout(exec.CephCommandsTimeout)
 	if err != nil {
 		return []Option{}, errors.Wrapf(err, "failed to get config for daemon %q. output: %s", who, string(out))
 	}

pkg/operator/ceph/config/monstore_test.go

@@ -20,6 +20,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
@@ -41,8 +42,13 @@ func TestMonStore_Set(t *testing.T) {
 	// us to cause it to return an error when it detects a keyword.
 	execedCmd := ""
 	execInjectErr := false
+<<<<<<< HEAD
 	executor.MockExecuteCommandWithOutputFile =
 		func(command string, outfile string, args ...string) (string, error) {
+=======
+	executor.MockExecuteCommandWithTimeout =
+		func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			execedCmd = command + " " + strings.Join(args, " ")
 			if execInjectErr {
 				return "output from cmd with error", errors.New("mocked error")
@@ -86,8 +92,13 @@ func TestMonStore_Delete(t *testing.T) {
 	// us to cause it to return an error when it detects a keyword.
 	execedCmd := ""
 	execInjectErr := false
+<<<<<<< HEAD
 	executor.MockExecuteCommandWithOutputFile =
 		func(command string, outfile string, args ...string) (string, error) {
+=======
+	executor.MockExecuteCommandWithTimeout =
+		func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			execedCmd = command + " " + strings.Join(args, " ")
 			if execInjectErr {
 				return "output from cmd with error", errors.New("mocked error")
@@ -125,8 +136,13 @@ func TestMonStore_GetDaemon(t *testing.T) {
 		"\"rgw_enable_usage_log\":{\"value\":\"true\",\"section\":\"client.rgw.test.a\",\"mask\":{}," +
 		"\"can_update_at_runtime\":true}}"
 	execInjectErr := false
+<<<<<<< HEAD
 	executor.MockExecuteCommandWithOutputFile =
 		func(command string, outfile string, args ...string) (string, error) {
+=======
+	executor.MockExecuteCommandWithTimeout =
+		func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			execedCmd = command + " " + strings.Join(args, " ")
 			if execInjectErr {
 				return "output from cmd with error", errors.New("mocked error")
@@ -171,8 +187,13 @@ func TestMonStore_DeleteDaemon(t *testing.T) {
 		"\"can_update_at_runtime\":true}," +
 		"\"rgw_enable_usage_log\":{\"value\":\"true\",\"section\":\"client.rgw.test.a\",\"mask\":{}," +
 		"\"can_update_at_runtime\":true}}"
+<<<<<<< HEAD
 	executor.MockExecuteCommandWithOutputFile =
 		func(command string, outfile string, args ...string) (string, error) {
+=======
+	executor.MockExecuteCommandWithTimeout =
+		func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			execedCmd = command + " " + strings.Join(args, " ")
 			return execReturn, nil
 		}
@@ -197,8 +218,13 @@ func TestMonStore_SetAll(t *testing.T) {
 	// us to cause it to return an error when it detects a keyword.
 	execedCmds := []string{}
 	execInjectErrOnKeyword := "donotinjectanerror"
+<<<<<<< HEAD
 	executor.MockExecuteCommandWithOutputFile =
 		func(command string, outfile string, args ...string) (string, error) {
+=======
+	executor.MockExecuteCommandWithTimeout =
+		func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			execedCmd := command + " " + strings.Join(args, " ")
 			execedCmds = append(execedCmds, execedCmd)
 			k := execInjectErrOnKeyword

pkg/operator/ceph/pool/controller_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/coreos/pkg/capnslog"
 	"github.com/pkg/errors"
@@ -406,7 +407,11 @@ func TestConfigureRBDStats(t *testing.T) {
 	)
 
 	executor := &exectest.MockExecutor{
+<<<<<<< HEAD
 		MockExecuteCommandWithOutputFile: func(command, outfile string, args ...string) (string, error) {
+=======
+		MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			logger.Infof("Command: %s %v", command, args)
 			switch {
 			case args[0] == "config" && args[1] == "set" && args[2] == "mgr." && args[3] == "mgr/prometheus/rbd_stats_pools" && args[4] != "":
@@ -470,7 +475,11 @@ func TestConfigureRBDStats(t *testing.T) {
 	// Case 5: Two CephBlockPools with EnableRBDStats:false & EnableRBDStats:true.
 	// SetConfig returns an error
 	context.Executor = &exectest.MockExecutor{
+<<<<<<< HEAD
 		MockExecuteCommandWithOutputFile: func(command, outfile string, args ...string) (string, error) {
+=======
+		MockExecuteCommandWithTimeout: func(timeout time.Duration, command string, args ...string) (string, error) {
+>>>>>>> 8da68bfb7 (mon: run ceph commands to mon with timeout)
 			logger.Infof("Command: %s %v", command, args)
 			return "", errors.New("mock error to simulate failure of SetConfig() function")
 		},