Added retransmissions in service call by e2e resource consumer library.

test/e2e/common/autoscaling_utils.go

@@ -24,6 +24,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/kubernetes/pkg/api/v1"
 	autoscalingv1 "k8s.io/kubernetes/pkg/apis/autoscaling/v1"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -52,6 +53,8 @@ const (
 	rsIsNil                         = "ERROR: replicaset = nil"
 	invalidKind                     = "ERROR: invalid workload kind for resource consumer"
 	customMetricName                = "QPS"
+	serviceInitializationTimeout    = 2 * time.Minute
+	serviceInitializationInterval   = 15 * time.Second
 )
 
 const (
@@ -223,6 +226,7 @@ func (rc *ResourceConsumer) sendConsumeCPURequest(millicores int) {
 	ctx, cancel := context.WithTimeout(context.Background(), framework.SingleCallTimeout)
 	defer cancel()
 
+	err = wait.PollImmediate(serviceInitializationInterval, serviceInitializationTimeout, func() (bool, error) {
 		req := proxyRequest.Namespace(rc.framework.Namespace.Name).
 			Context(ctx).
 			Name(rc.controllerName).
@@ -230,8 +234,15 @@ func (rc *ResourceConsumer) sendConsumeCPURequest(millicores int) {
 			Param("millicores", strconv.Itoa(millicores)).
 			Param("durationSec", strconv.Itoa(rc.consumptionTimeInSeconds)).
 			Param("requestSizeMillicores", strconv.Itoa(rc.requestSizeInMillicores))
-	framework.Logf("URL: %v", *req.URL())
-	_, err = req.DoRaw()
+		framework.Logf("ConsumeCPU URL: %v", *req.URL())
+		_, err := req.DoRaw()
+		if err != nil {
+			framework.Logf("ConsumeCPU failure: %v", err)
+			return false, nil
+		}
+		return true, nil
+	})
+
 	framework.ExpectNoError(err)
 }
 
@@ -243,6 +254,7 @@ func (rc *ResourceConsumer) sendConsumeMemRequest(megabytes int) {
 	ctx, cancel := context.WithTimeout(context.Background(), framework.SingleCallTimeout)
 	defer cancel()
 
+	err = wait.PollImmediate(serviceInitializationInterval, serviceInitializationTimeout, func() (bool, error) {
 		req := proxyRequest.Namespace(rc.framework.Namespace.Name).
 			Context(ctx).
 			Name(rc.controllerName).
@@ -250,8 +262,15 @@ func (rc *ResourceConsumer) sendConsumeMemRequest(megabytes int) {
 			Param("megabytes", strconv.Itoa(megabytes)).
 			Param("durationSec", strconv.Itoa(rc.consumptionTimeInSeconds)).
 			Param("requestSizeMegabytes", strconv.Itoa(rc.requestSizeInMegabytes))
-	framework.Logf("URL: %v", *req.URL())
-	_, err = req.DoRaw()
+		framework.Logf("ConsumeMem URL: %v", *req.URL())
+		_, err := req.DoRaw()
+		if err != nil {
+			framework.Logf("ConsumeMem failure: %v", err)
+			return false, nil
+		}
+		return true, nil
+	})
+
 	framework.ExpectNoError(err)
 }
 
@@ -263,6 +282,7 @@ func (rc *ResourceConsumer) sendConsumeCustomMetric(delta int) {
 	ctx, cancel := context.WithTimeout(context.Background(), framework.SingleCallTimeout)
 	defer cancel()
 
+	err = wait.PollImmediate(serviceInitializationInterval, serviceInitializationTimeout, func() (bool, error) {
 		req := proxyRequest.Namespace(rc.framework.Namespace.Name).
 			Context(ctx).
 			Name(rc.controllerName).
@@ -271,8 +291,14 @@ func (rc *ResourceConsumer) sendConsumeCustomMetric(delta int) {
 			Param("delta", strconv.Itoa(delta)).
 			Param("durationSec", strconv.Itoa(rc.consumptionTimeInSeconds)).
 			Param("requestSizeMetrics", strconv.Itoa(rc.requestSizeCustomMetric))
-	framework.Logf("URL: %v", *req.URL())
-	_, err = req.DoRaw()
+		framework.Logf("ConsumeCustomMetric URL: %v", *req.URL())
+		_, err := req.DoRaw()
+		if err != nil {
+			framework.Logf("ConsumeCustomMetric failure: %v", err)
+			return false, nil
+		}
+		return true, nil
+	})
 	framework.ExpectNoError(err)
 }