Merge pull request kubernetes#2222 from Argh4k/poll-exec-timeout

Add timeout to exec command used for polling single master health

clusterloader2/pkg/execservice/exec_service.go

@@ -134,10 +134,10 @@ func TearDownExecService(f *framework.Framework) error {
 	return nil
 }
 
-// RunCommand executes given command on a pod in cluster.
-func RunCommand(pod *corev1.Pod, cmd string) (string, error) {
+// RunCommand executes given command on a pod in cluster. Context is passed to the exec command.
+func RunCommand(ctx context.Context, pod *corev1.Pod, cmd string) (string, error) {
 	var stdout, stderr bytes.Buffer
-	c := exec.Command("kubectl", "exec", fmt.Sprintf("--namespace=%v", pod.Namespace), pod.Name, "--", "/bin/sh", "-x", "-c", cmd)
+	c := exec.CommandContext(ctx, "kubectl", "exec", fmt.Sprintf("--namespace=%v", pod.Namespace), pod.Name, "--", "/bin/sh", "-x", "-c", cmd)
 	c.Stdout, c.Stderr = &stdout, &stderr
 	if err := c.Run(); err != nil {
 		return stderr.String(), err

clusterloader2/pkg/measurement/common/api_availability_measurement.go

@@ -59,10 +59,11 @@ type apiAvailabilityMeasurement struct {
 	clusterLevelMetrics *apiAvailabilityMetrics
 	threshold           float64
 	// Metrics per host internal IP.
-	hostLevelMetrics       map[string]*apiAvailabilityMetrics
-	hostPollTimeoutSeconds int
-	wg                     sync.WaitGroup
-	lock                   sync.Mutex
+	hostLevelMetrics           map[string]*apiAvailabilityMetrics
+	hostPollTimeoutSeconds     int
+	hostPollExecTimeoutSeconds int
+	wg                         sync.WaitGroup
+	lock                       sync.Mutex
 }
 
 func (a *apiAvailabilityMeasurement) updateHostAvailabilityMetrics(c clientset.Interface, provider provider.Provider) {
@@ -95,7 +96,9 @@ func (a *apiAvailabilityMeasurement) pollHost(hostIP string) (string, error) {
 		return "", fmt.Errorf("problem with GetPod(): %w", err)
 	}
 	cmd := fmt.Sprintf("curl --connect-timeout %d -s -k -w \"%%{http_code}\" -o /dev/null https://%s:443/readyz", a.hostPollTimeoutSeconds, hostIP)
-	output, err := execservice.RunCommand(pod, cmd)
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(a.hostPollExecTimeoutSeconds)*time.Second)
+	defer cancel()
+	output, err := execservice.RunCommand(ctx, pod, cmd)
 	if err != nil {
 		return "", fmt.Errorf("problem with RunCommand(): output=%q, err=%w", output, err)
 	}
@@ -210,6 +213,11 @@ func (a *apiAvailabilityMeasurement) initFields(config *measurement.Config) erro
 			return err
 		}
 		a.hostPollTimeoutSeconds = hostPollTimeoutSeconds
+		hostPollExecTimeoutSeconds, err := util.GetIntOrDefault(config.Params, "hostPollExecTimeoutSeconds", 10)
+		if err != nil {
+			return err
+		}
+		a.hostPollExecTimeoutSeconds = hostPollExecTimeoutSeconds
 	} else {
 		klog.V(2).Infof("%s: exec service is not enabled, therefore only cluster-level availability will be measured", a)
 	}

clusterloader2/pkg/measurement/common/service_creation_latency.go

@@ -380,7 +380,7 @@ func (p *pingChecker) run() {
 			for _, ip := range ips {
 				address := net.JoinHostPort(ip, fmt.Sprint(port))
 				command := fmt.Sprintf("curl %s", address)
-				_, err = execservice.RunCommand(pod, command)
+				_, err = execservice.RunCommand(context.TODO(), pod, command)
 				if err != nil {
 					break
 				}